""" 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