"""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 the three properties that makes safe: it reaches the reporter, it survives the worker thread T-74 introduced, and concurrent executions cannot read each other's. """ 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", 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", 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"}