jpmschweitzerandClaude Opus 4.5
64574bcc39
Build and Push / build (release) Failing after 17s
Extracted standalone scheduler service with: - FastAPI REST API for task management - APScheduler-based task execution - PostgreSQL persistence - Docker container support - Gitea Actions CI/CD workflow 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
308 lines
12 KiB
Python
308 lines
12 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
|