jpmschweitzerandClaude
3572a45322
_should_run_now has no definition in src/ in any commit in this repo's history — checked with `git log --all -S` across the whole tree, not just the current worktree. Twelve assertion sites across six tests called it, so these have never passed and never protected anything. The behaviour they describe is real: cron-wildcard matching of minute, hour, day_of_month, month and day_of_week. It lives inside the WHERE clause of get_tasks_for_minute, not as a Python predicate, so there was nothing to rename them onto. KNOWN GAP, stated rather than left implied: that matching is now covered by no test at all. Testing it means either asserting against the SQL and params a mocked cursor receives, or extracting the predicate out of the query — the second changes what decides, every minute, which scheduled work runs, and is not a refactor to do casually. Deleting was chosen over rewriting because a test that has never run is not coverage, and leaving it in place claimed some. Co-Authored-By: Claude <noreply@anthropic.com>
430 lines
18 KiB
Python
430 lines
18 KiB
Python
"""
|
|
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
|