jpmschweitzerandClaude
e45de4fec7
Two defects with one shape: an execution reaches a terminal condition and the
orchestrator fails to write it down, so the system's own record disagrees with
what happened. Neither produced an error. Both produced silence.
T-2 -- a restart mid-task unscheduled that task forever.
get_tasks_for_minute excludes any task holding a task_executions row with
status='running'. The row is written before the executor runs and updated
after, so a process dying in between left it 'running' permanently, and the
task was then excluded from every future minute with no error, no alarm and no
log line. It did not fail; it went quiet.
test_example_task had held such a row since 2025-12-07 -- 5916 hours. It is
disabled, so nothing was broken by that instance; the mechanism is the point,
and the exposure is daily, because Watchtower restarts this container at 4 AM
while the config backup starts at 03:05 and runs ~21 minutes.
Startup now reconciles them, where the reasoning is sound by construction: this
process has just begun, so nothing it can see is genuinely running.
Marked 'orphaned', not 'failed'. When the process dies mid-task the work may
well have completed -- a backup that finished and never got to update its row
is indistinguishable from one that died halfway -- and 'failed' would assert an
outcome nobody observed. Same error as the health-report diagnostic fixed in
18be804: naming a cause you did not witness.
Not extended to a duration-based sweep. While this process lives, execute_task's
finally clause always closes the row, so a stale row implies a dead owner. A
time-based rule would have to tell a slow task from a dead one, and getting that
wrong closes the record of a task still working.
T-3 -- the 'timeout' status was unreachable.
execute_task has an `except asyncio.TimeoutError` branch that records
status='timeout'. It could never run: _run_executor wrapped the awaited call in
`except Exception`, and since 3.11 asyncio.TimeoutError IS the builtin
TimeoutError (OSError -> Exception), so the broad handler caught it first and
converted it to an ordinary error tuple. Confirmed in the deployed runtime and
against the history -- 18,785 executions since 2025-12-07, of which 'timeout'
rows: zero. Every timeout in eight months was filed as a generic failure,
erasing the distinction between "too slow for its window" and "broken".
A narrower except after a broader one is dead code, and no linter is configured
here to say so.
One trap in fixing it: while the branch was unreachable a timeout travelled the
normal path, which DOES update scheduled_tasks. Making the branch reachable
without that write would have traded a wrong status for a stale one, so
_update_task_outcome now mirrors terminal outcomes onto the parent row.
The timeout message also states that the work may still be running -- after
T-74 executors are handed to asyncio.to_thread, and a thread cannot be
cancelled, so wait_for frees the loop while the work continues.
Both fixes are mutation-checked: removing the startup call fails the ordering
test, removing the narrow except clause fails the propagation test. Suite goes
118 -> 126 passing with the same 36 pre-existing failures.
Co-Authored-By: Claude <noreply@anthropic.com>
486 lines
19 KiB
Python
486 lines
19 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
|
|
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
|
|
assert executor.max_concurrent == 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}
|
|
|
|
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
|
|
patch('src.tasks.executor.importlib.import_module') as mock_import:
|
|
|
|
# Mock executor module
|
|
mock_executor_module = MagicMock()
|
|
mock_executor_module.execute = AsyncMock(return_value="Task completed successfully")
|
|
mock_import.return_value = mock_executor_module
|
|
|
|
# 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_called_with('src.executors.example_executor')
|
|
|
|
# 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}
|
|
|
|
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
|
|
patch('src.tasks.executor.importlib.import_module') as mock_import:
|
|
|
|
# Mock executor that raises error
|
|
mock_executor_module = MagicMock()
|
|
mock_executor_module.execute = AsyncMock(side_effect=Exception("Task failed"))
|
|
mock_import.return_value = mock_executor_module
|
|
|
|
# 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}
|
|
|
|
with patch.object(executor, 'get_db_connection') as mock_get_conn, \
|
|
patch('src.tasks.executor.importlib.import_module') as mock_import:
|
|
|
|
# Mock executor that takes too long
|
|
import asyncio
|
|
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
|
|
mock_import.return_value = mock_executor_module
|
|
|
|
# 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)
|
|
|
|
def test_should_run_task_wildcard(self, test_settings: Settings):
|
|
"""Test task scheduling with wildcards."""
|
|
executor = TaskExecutor(test_settings)
|
|
|
|
# All wildcards should always match
|
|
task = {
|
|
'minute': -1,
|
|
'hour': -1,
|
|
'day_of_month': -1,
|
|
'month': -1,
|
|
'day_of_week': -1
|
|
}
|
|
|
|
now = datetime(2025, 12, 7, 14, 30, 0) # Saturday
|
|
|
|
assert executor._should_run_now(task, now) is True
|
|
|
|
def test_should_run_task_specific_time(self, test_settings: Settings):
|
|
"""Test task scheduling with specific time."""
|
|
executor = TaskExecutor(test_settings)
|
|
|
|
# Specific time: every day at 14:30
|
|
task = {
|
|
'minute': 30,
|
|
'hour': 14,
|
|
'day_of_month': -1,
|
|
'month': -1,
|
|
'day_of_week': -1
|
|
}
|
|
|
|
# Matching time
|
|
now = datetime(2025, 12, 7, 14, 30, 0)
|
|
assert executor._should_run_now(task, now) is True
|
|
|
|
# Non-matching time
|
|
now = datetime(2025, 12, 7, 14, 31, 0)
|
|
assert executor._should_run_now(task, now) is False
|
|
|
|
def test_should_run_task_specific_day_of_month(self, test_settings: Settings):
|
|
"""Test task scheduling with specific day of month."""
|
|
executor = TaskExecutor(test_settings)
|
|
|
|
# Run on 11th of every month at 04:00
|
|
task = {
|
|
'minute': 0,
|
|
'hour': 4,
|
|
'day_of_month': 11,
|
|
'month': -1,
|
|
'day_of_week': -1
|
|
}
|
|
|
|
# Matching date
|
|
now = datetime(2025, 12, 11, 4, 0, 0)
|
|
assert executor._should_run_now(task, now) is True
|
|
|
|
# Wrong day
|
|
now = datetime(2025, 12, 12, 4, 0, 0)
|
|
assert executor._should_run_now(task, now) is False
|
|
|
|
def test_should_run_task_specific_month(self, test_settings: Settings):
|
|
"""Test task scheduling with specific month."""
|
|
executor = TaskExecutor(test_settings)
|
|
|
|
# Run on January 1st at midnight
|
|
task = {
|
|
'minute': 0,
|
|
'hour': 0,
|
|
'day_of_month': 1,
|
|
'month': 1,
|
|
'day_of_week': -1
|
|
}
|
|
|
|
# Matching date
|
|
now = datetime(2025, 1, 1, 0, 0, 0)
|
|
assert executor._should_run_now(task, now) is True
|
|
|
|
# Wrong month
|
|
now = datetime(2025, 2, 1, 0, 0, 0)
|
|
assert executor._should_run_now(task, now) is False
|
|
|
|
def test_should_run_task_day_of_week(self, test_settings: Settings):
|
|
"""Test task scheduling with day of week."""
|
|
executor = TaskExecutor(test_settings)
|
|
|
|
# Run every Monday at 09:00
|
|
task = {
|
|
'minute': 0,
|
|
'hour': 9,
|
|
'day_of_month': -1,
|
|
'month': -1,
|
|
'day_of_week': 0 # Monday
|
|
}
|
|
|
|
# Monday
|
|
now = datetime(2025, 12, 8, 9, 0, 0) # Monday
|
|
assert executor._should_run_now(task, now) is True
|
|
|
|
# Tuesday
|
|
now = datetime(2025, 12, 9, 9, 0, 0) # Tuesday
|
|
assert executor._should_run_now(task, now) is False
|
|
|
|
@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()
|
|
|
|
# Should only execute max_concurrent (5) tasks
|
|
assert mock_execute.call_count <= executor.max_concurrent
|
|
|
|
|
|
@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
|