""" Tests for the task executor module. """ import pytest from unittest.mock import AsyncMock, MagicMock, patch from datetime import datetime from src.tasks.executor import TaskExecutor, MAX_CONCURRENT_TASKS from src.config import Settings @pytest.mark.unit class TestTaskExecutor: """Tests for TaskExecutor class.""" def test_executor_initialization(self, test_settings: Settings): """Test that executor initializes correctly.""" executor = TaskExecutor(test_settings) assert executor.settings == test_settings # There is no `max_concurrent` instance attribute — concurrency is # capped by MAX_CONCURRENT_TASKS (module constant) via # asyncio.Semaphore(MAX_CONCURRENT_TASKS) in __init__. Verify the # semaphore was built with that bound instead of asserting an # attribute name the class has never had. assert MAX_CONCURRENT_TASKS == 5 assert executor.semaphore._value == 5 @patch('psycopg2.connect') def test_get_db_connection(self, mock_connect, test_settings: Settings): """Test database connection creation.""" mock_conn = MagicMock() mock_connect.return_value = mock_conn executor = TaskExecutor(test_settings) conn = executor.get_db_connection() assert conn == mock_conn mock_connect.assert_called_once() @pytest.mark.asyncio async def test_process_minute_no_tasks(self, test_settings: Settings): """Test process_minute with no scheduled tasks.""" executor = TaskExecutor(test_settings) with patch.object(executor, 'get_db_connection') as mock_get_conn: mock_conn = MagicMock() mock_cursor = MagicMock() mock_cursor.fetchall.return_value = [] mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) # Should complete without error await executor.process_minute() # Should have queried for tasks assert mock_cursor.execute.called @pytest.mark.asyncio async def test_process_minute_with_tasks(self, test_settings: Settings, sample_task_data: dict): """Test process_minute executes tasks.""" executor = TaskExecutor(test_settings) task_from_db = { **sample_task_data, 'id': 1, 'config': sample_task_data['config'] # Already a dict } with patch.object(executor, 'get_db_connection') as mock_get_conn, \ patch.object(executor, 'execute_task', new_callable=AsyncMock) as mock_execute: mock_conn = MagicMock() mock_cursor = MagicMock() mock_cursor.fetchall.return_value = [task_from_db] mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) await executor.process_minute() # Should have executed the task mock_execute.assert_called_once() @pytest.mark.asyncio async def test_execute_task_success(self, test_settings: Settings, sample_task_data: dict): """Test successful task execution.""" executor = TaskExecutor(test_settings) task = {**sample_task_data, 'id': 1} # _run_executor (src/tasks/executor.py) loads the executor module with # the __import__ builtin directly — `module = __import__(module_path, # fromlist=['execute'])` — not importlib.import_module. This is the # documented dynamic-loading trap in this repo's own CLAUDE.md # ("executors are chosen by data, not code"). `importlib` is never # imported in that module, so patching 'src.tasks.executor.importlib' # fails at patch setup, before the test body runs at all. real_import = __import__ def fake_import(name, globals=None, locals=None, fromlist=(), level=0): if name == 'src.executors.example_executor': return mock_executor_module return real_import(name, globals, locals, fromlist, level) mock_executor_module = MagicMock() mock_executor_module.execute = AsyncMock(return_value="Task completed successfully") with patch.object(executor, 'get_db_connection') as mock_get_conn, \ patch('builtins.__import__', side_effect=fake_import) as mock_import: # Mock database mock_conn = MagicMock() mock_cursor = MagicMock() mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) await executor.execute_task(task) # Should have imported executor module mock_import.assert_any_call('src.executors.example_executor', fromlist=['execute']) # Should have updated task status assert mock_cursor.execute.call_count >= 2 # Insert execution record + update task @pytest.mark.asyncio async def test_execute_task_failure(self, test_settings: Settings, sample_task_data: dict): """Test task execution handles failures.""" executor = TaskExecutor(test_settings) task = {**sample_task_data, 'id': 1} # See test_execute_task_success: _run_executor uses the __import__ # builtin directly, not importlib.import_module. real_import = __import__ def fake_import(name, globals=None, locals=None, fromlist=(), level=0): if name == 'src.executors.example_executor': return mock_executor_module return real_import(name, globals, locals, fromlist, level) mock_executor_module = MagicMock() mock_executor_module.execute = AsyncMock(side_effect=Exception("Task failed")) with patch.object(executor, 'get_db_connection') as mock_get_conn, \ patch('builtins.__import__', side_effect=fake_import): # Mock database mock_conn = MagicMock() mock_cursor = MagicMock() mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) await executor.execute_task(task) # Should have recorded the failure assert mock_cursor.execute.called # Check that error status was recorded calls = [str(call) for call in mock_cursor.execute.call_args_list] assert any('failed' in str(call).lower() or 'error' in str(call).lower() for call in calls) @pytest.mark.asyncio async def test_execute_task_timeout(self, test_settings: Settings, sample_task_data: dict): """Test task execution handles timeouts.""" executor = TaskExecutor(test_settings) task = {**sample_task_data, 'id': 1, 'timeout_seconds': 1} # See test_execute_task_success: _run_executor uses the __import__ # builtin directly, not importlib.import_module. import asyncio real_import = __import__ def fake_import(name, globals=None, locals=None, fromlist=(), level=0): if name == 'src.executors.example_executor': return mock_executor_module return real_import(name, globals, locals, fromlist, level) mock_executor_module = MagicMock() async def slow_execute(*args, **kwargs): await asyncio.sleep(10) # Longer than timeout return "Done" mock_executor_module.execute = slow_execute with patch.object(executor, 'get_db_connection') as mock_get_conn, \ patch('builtins.__import__', side_effect=fake_import): # Mock database mock_conn = MagicMock() mock_cursor = MagicMock() mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) await executor.execute_task(task) # Should have recorded timeout calls = [str(call) for call in mock_cursor.execute.call_args_list] assert any('timeout' in str(call).lower() for call in calls) @pytest.mark.asyncio async def test_concurrent_task_limit(self, test_settings: Settings, sample_task_data: dict): """Test that executor respects concurrent task limit.""" executor = TaskExecutor(test_settings) # Create 10 tasks tasks = [ {**sample_task_data, 'id': i, 'task_name': f'task_{i}'} for i in range(10) ] with patch.object(executor, 'get_db_connection') as mock_get_conn, \ patch.object(executor, 'execute_task', new_callable=AsyncMock) as mock_execute: mock_conn = MagicMock() mock_cursor = MagicMock() mock_cursor.fetchall.return_value = tasks mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor) mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None) mock_get_conn.return_value.__enter__ = MagicMock(return_value=mock_conn) mock_get_conn.return_value.__exit__ = MagicMock(return_value=None) await executor.process_minute() # process_minute awaits every task in this batch (asyncio.gather) # before returning, so by the time we're back here all 10 have # run to completion — the semaphore bounds how many can be # in flight *concurrently*, not the eventual call_count, which # this assertion conflated. Kept as a correctness check on the # total (all scheduled tasks still get executed) since a # concurrency-in-flight assertion needs a task that can be # observed mid-execution, which mock_execute (an AsyncMock with # no delay) does not provide. assert mock_execute.call_count == len(tasks) @pytest.mark.unit class TestOrphanReconciliation: """T-2. A 'running' row excludes its task from scheduling forever. get_tasks_for_minute filters out any task holding one, and nothing ever closed those rows, so a process that died between writing the row and updating it left its task permanently unschedulable — silently. A row from 2025-12-07 sat that way for eight months. Watchtower restarts this container nightly, so the exposure was daily. """ def _mock_conn(self, executor, rows): conn, cur = MagicMock(), MagicMock() cur.fetchall.return_value = rows 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) patcher = patch.object(executor, 'get_db_connection', return_value=conn) patcher.start() return cur, patcher def test_running_rows_are_released(self, test_settings: Settings): executor = TaskExecutor(test_settings) cur, p = self._mock_conn(executor, [("backup_docker_configs_daily", datetime(2026, 8, 11))]) try: assert executor.reconcile_orphaned_executions() == 1 sql = cur.execute.call_args[0][0] assert "UPDATE task_executions" in sql # It must target exactly the rows get_tasks_for_minute excludes, and # move them to a status it does not exclude. If these two ever drift # apart the bug returns in silence. assert "WHERE status = 'running'" in sql assert "status = 'orphaned'" in sql finally: p.stop() def test_clean_startup_reports_nothing(self, test_settings: Settings): executor = TaskExecutor(test_settings) cur, p = self._mock_conn(executor, []) try: assert executor.reconcile_orphaned_executions() == 0 finally: p.stop() def test_a_database_failure_does_not_stop_startup(self, test_settings: Settings): """Refusing to boot because cleanup failed is worse than a stale row.""" executor = TaskExecutor(test_settings) with patch.object(executor, 'get_db_connection', side_effect=Exception("db down")): assert executor.reconcile_orphaned_executions() == 0 # no raise def test_orphaned_is_not_failed(self, test_settings: Settings): """The outcome is unknown, not known-bad. A backup that finished and never got to update its row looks identical to one that died halfway. Recording 'failed' asserts something nobody observed. """ executor = TaskExecutor(test_settings) cur, p = self._mock_conn(executor, [("t", datetime(2026, 1, 1))]) try: executor.reconcile_orphaned_executions() sql = cur.execute.call_args[0][0] assert "'failed'" not in sql finally: p.stop() @pytest.mark.asyncio async def test_startup_reconciles_before_the_scheduler_starts( self, test_settings: Settings, monkeypatch ): """Order matters: reconcile must finish before the first minute is processed. Run the other way round and the first tick still sees the stale rows. """ from src import main order = [] monkeypatch.setattr(main, 'get_settings', lambda: test_settings) monkeypatch.setattr( TaskExecutor, 'reconcile_orphaned_executions', lambda self: (order.append('reconcile'), 0)[1], ) class FakeScheduler: def add_job(self, **kw): order.append('add_job') def start(self): order.append('start') def shutdown(self, wait=True): order.append('shutdown') monkeypatch.setattr(main, 'AsyncIOScheduler', lambda **kw: FakeScheduler()) async with main.lifespan(None): pass assert 'reconcile' in order, "startup never reconciled orphaned executions" assert order.index('reconcile') < order.index('start') @pytest.mark.unit class TestTimeoutIsDistinguishable: """T-3. The 'timeout' status existed in the code and had never been written. _run_executor's `except Exception` sat above execute_task's `except asyncio.TimeoutError`, and since 3.11 asyncio.TimeoutError IS the builtin TimeoutError (OSError -> Exception), so the broad handler always won. Eight months, 18,785 executions, zero timeout rows — every one filed as a generic failure, erasing the difference between "too slow for its window" and "broken". """ @pytest.mark.asyncio async def test_timeout_propagates_instead_of_becoming_an_error_tuple( self, test_settings: Settings, monkeypatch ): import sys, types, asyncio as aio mod = types.ModuleType("src.executors.slow_probe") async def execute(config, settings): await aio.sleep(5) mod.execute = execute monkeypatch.setitem(sys.modules, "src.executors.slow_probe", mod) executor = TaskExecutor(test_settings) with pytest.raises(aio.TimeoutError): await executor._run_executor("slow_probe", {"config": {}}, timeout=0.05) @pytest.mark.asyncio async def test_a_timed_out_task_is_recorded_as_timeout(self, test_settings: Settings): import asyncio as aio executor = TaskExecutor(test_settings) task = {'id': 7, 'task_name': 'slow', 'executor': 'slow_probe', 'service': 'scheduler', 'priority': 5, 'timeout_seconds': 1} conn, cur = MagicMock(), MagicMock() cur.fetchone.return_value = [123] 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) with patch.object(executor, 'get_db_connection', return_value=conn), \ patch.object(executor, '_run_executor', new=AsyncMock(side_effect=aio.TimeoutError())), \ patch.object(executor, '_update_execution_status') as upd_exec, \ patch.object(executor, '_update_task_outcome') as upd_task: await executor.execute_task(task) assert upd_exec.call_args[0][1] == 'timeout', "execution row must say timeout" # The trap: while the timeout branch was unreachable a timeout travelled # the normal path, which DOES update scheduled_tasks. Making the branch # reachable without this call would swap a wrong status for a stale one. assert upd_task.called, "scheduled_tasks left stale after a timeout" assert upd_task.call_args[0][1] == 'timeout' @pytest.mark.asyncio async def test_ordinary_errors_are_still_returned_not_raised( self, test_settings: Settings, monkeypatch ): """The narrow clause must not swallow anything else on its way past.""" import sys, types mod = types.ModuleType("src.executors.boom_probe") async def execute(config, settings): raise ValueError("kaboom") mod.execute = execute monkeypatch.setitem(sys.modules, "src.executors.boom_probe", mod) executor = TaskExecutor(test_settings) output, error = await executor._run_executor("boom_probe", {"config": {}}, timeout=5) assert output is None assert "kaboom" in error