jpmschweitzer
68bea0cceb
1. `executor.max_concurrent` has never existed. Concurrency is capped by the
module-level MAX_CONCURRENT_TASKS constant via asyncio.Semaphore(
MAX_CONCURRENT_TASKS) in TaskExecutor.__init__ — confirmed with git log -p
across this file's whole history (three commits), the name has always
been the module constant, never an instance attribute.
test_executor_initialization now asserts MAX_CONCURRENT_TASKS == 5 and the
semaphore's initial count, instead of a name the class never had.
test_concurrent_task_limit asserted `mock_execute.call_count <=
executor.max_concurrent`, which — separately from the AttributeError — was
asserting the wrong observable: process_minute() awaits the full batch via
asyncio.gather before returning, so by the time the assertion runs all 10
scheduled tasks have executed; the semaphore bounds how many run
concurrently mid-flight, not the eventual call_count. Reworded to assert
all scheduled tasks still run (call_count == len(tasks)); a concurrency-
in-flight assertion would need a task that can be observed mid-execution,
which the AsyncMock stand-in does not provide.
2. _run_executor (src/tasks/executor.py) loads the executor module with the
__import__ builtin directly (`__import__(module_path, fromlist=
['execute'])`), not importlib.import_module — this repo's own CLAUDE.md
documents it as "the thing that will mislead you" about this module.
importlib is never imported there, so patch('src.tasks.executor.importlib.
import_module') failed at patch setup, before the three
test_execute_task_* bodies ran at all. Switched to patch('builtins.
__import__', side_effect=...) with a routing function that falls through
to the real import for anything other than the target module — verified
the call-recording shape empirically first (call('name', fromlist=[...])).
Source is unchanged in both cases; both are test-only defects present since
this file's initial commit.
526 lines
21 KiB
Python
526 lines
21 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)
|
|
|
|
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()
|
|
|
|
# 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
|