scheduler/tests/test_task_attribution.py

"""A check_history row must say which task produced it.

On 2026-08-11 a 425 MB probe task and the 5 GB nightly backup both ran through
config_backup_executor. The probe failed, and the row it wrote was
indistinguishable from a nightly-backup failure — same `source`, same `domain`,
nothing naming the task. The health record could not answer which job broke,
which is most of what a health record is for.

Executors are called as `execute(config, settings)` and are never told which
task they are, so the name travels in a ContextVar. These tests pin what makes
that safe: it reaches the reporter, it survives the worker thread T-74
introduced, and concurrent executions cannot read each other's. They also pin
where the summary lives, since two producers write this table.
"""
import asyncio
import json

import pytest
from unittest.mock import AsyncMock, MagicMock, patch

from src.config import Settings
from src.executors import health_report
from src.task_context import current_task_name, task_scope
from src.tasks.executor import TaskExecutor


@pytest.fixture
def captured_row(monkeypatch):
    """Capture the JSON payload report() would insert, without a database."""
    box = {}
    conn, cur = MagicMock(), MagicMock()

    def execute(sql, params):
        box['result'] = json.loads(params[4])

    cur.execute.side_effect = execute
    conn.cursor.return_value.__enter__ = MagicMock(return_value=cur)
    conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
    conn.__enter__ = MagicMock(return_value=conn)
    conn.__exit__ = MagicMock(return_value=None)
    monkeypatch.setattr(health_report.psycopg2, 'connect', lambda **kw: conn)
    return box


def _report(settings, **kw):
    health_report.report(
        settings, domain="backup", status=health_report.OK,
        source="scheduler/config_backup_executor",
        summary="backed up 3 sources", metrics={}, **kw
    )


@pytest.mark.unit
class TestAttribution:

    def test_the_row_names_the_task(self, test_settings: Settings, captured_row):
        with task_scope("backup_docker_configs_daily"):
            _report(test_settings)
        assert captured_row['result']['task'] == "backup_docker_configs_daily"

    def test_source_still_names_the_code(self, test_settings: Settings, captured_row):
        """Task and source answer different questions; both are needed.

        `source` says which code wrote the row, `task` says which schedule
        invoked it. Replacing one with the other loses a distinction.
        """
        with task_scope("t74_loop_probe"):
            _report(test_settings)
        r = captured_row['result']
        assert r['source'] == "scheduler/config_backup_executor"
        assert r['task'] == "t74_loop_probe"

    def test_the_field_is_omitted_rather_than_nulled(self, test_settings: Settings, captured_row):
        """report() is callable from a script with no task around it.

        A null would claim there was a task and it had no name. Absence says the
        question does not apply.
        """
        _report(test_settings)
        assert 'task' not in captured_row['result']

    @pytest.mark.asyncio
    async def test_attribution_survives_the_worker_thread(
        self, test_settings: Settings, captured_row
    ):
        """T-74 moved reporting into asyncio.to_thread. Attribution has to follow.

        If the context did not propagate, every row written by a real executor
        would silently lose its task while the tests above still passed.
        """
        with task_scope("backup_portainer_daily"):
            await health_report.report_async(
                test_settings, domain="backup", status=health_report.OK,
                source="scheduler/portainer_backup_executor",
                summary="backed up Portainer", metrics={},
            )
        assert captured_row['result']['task'] == "backup_portainer_daily"


@pytest.mark.unit
class TestAttributionIsolation:
    """MAX_CONCURRENT_TASKS is 5, so mixing values between them is a live risk."""

    def _executor_with_db(self, test_settings):
        executor = TaskExecutor(test_settings)
        conn, cur = MagicMock(), MagicMock()
        cur.fetchone.return_value = [1]
        conn.cursor.return_value.__enter__ = MagicMock(return_value=cur)
        conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
        conn.__enter__ = MagicMock(return_value=conn)
        conn.__exit__ = MagicMock(return_value=None)
        return executor, patch.object(executor, 'get_db_connection', return_value=conn)

    def _task(self, name):
        return {'id': hash(name) % 1000, 'task_name': name, 'executor': 'x',
                'service': 'scheduler', 'priority': 5, 'timeout_seconds': 60}

    @pytest.mark.asyncio
    async def test_the_engine_names_the_task_and_clears_it_after(
        self, test_settings: Settings
    ):
        executor, db = self._executor_with_db(test_settings)
        seen = {}

        async def fake_run(name, task, timeout):
            seen['during'] = current_task_name()
            return "done", None

        with db, patch.object(executor, '_run_executor', new=fake_run):
            await executor.execute_task(self._task("nightly_backup"))

        assert seen['during'] == "nightly_backup", "executor ran unattributed"
        assert current_task_name() is None, "attribution leaked past the execution"

    @pytest.mark.asyncio
    async def test_concurrent_executions_do_not_see_each_other(
        self, test_settings: Settings
    ):
        executor, db = self._executor_with_db(test_settings)
        seen = {}

        async def fake_run(name, task, timeout):
            who = task['task_name']
            # Yield mid-flight so the executions genuinely interleave; without
            # this they would run to completion one at a time and the test would
            # pass even with a shared global.
            await asyncio.sleep(0.01 if who == "slow_one" else 0)
            seen[who] = current_task_name()
            return "done", None

        with db, patch.object(executor, '_run_executor', new=fake_run):
            await asyncio.gather(
                executor.execute_task(self._task("slow_one")),
                executor.execute_task(self._task("fast_one")),
            )

        assert seen == {"slow_one": "slow_one", "fast_one": "fast_one"}


@pytest.mark.unit
class TestSummaryPlacement:
    """The two writers of check_history must agree where the substance lives.

    sysmon-go writes `summary` at the top level, beside `status`. This module
    wrote it under `metrics` until 2026-08-11, so a reader had to know which
    producer wrote a row before it could find out what the row said — and a
    query written the obvious way silently found half the data. That is the T-36
    failure exactly, where per-domain queries returned nothing because the value
    was nested somewhere else.
    """

    def test_summary_is_top_level(self, test_settings: Settings, captured_row):
        _report(test_settings)
        r = captured_row['result']
        assert r['summary'] == "backed up 3 sources"
        assert 'summary' not in r['metrics'], "summary must not also live under metrics"

    def test_summary_is_required(self, test_settings: Settings, captured_row):
        """Omitting it is an error at the call, not a silently empty column.

        sysmon-go enforces this through Domain.Run's signature; a parameter with
        no default is the equivalent here. A row whose substance is missing looks
        exactly like a row whose check found nothing to say.
        """
        with pytest.raises(TypeError):
            health_report.report(
                test_settings, domain="backup", status=health_report.OK,
                source="scheduler/x", metrics={},
            )