Files

932 lines
32 KiB
Python
Raw Permalink Normal View History

"""Comprehensive tests for the Executor module.
Tests cover:
- Executor initialization
- add_function method
- add_coroutine method
- add_strategy and add_strategies methods
- run_strategy static method (async and sync strategies)
- run_coroutine_tasks method
- run_coroutine_task static method
- run_function static method
- sigint_handle method
- exit method
- execute method
- Integration tests
"""
import asyncio
import inspect
2026-02-21 07:43:33 +01:00
import time
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import MagicMock, AsyncMock, patch, call
import pytest
from aiomql.lib.executor import Executor
from aiomql.lib.strategy import Strategy
from aiomql.core.config import Config
class MockAsyncStrategy:
"""Mock async strategy for testing."""
def __init__(self):
self.running = True
async def run_strategy(self):
"""Async run_strategy method."""
pass
class MockSyncStrategy:
"""Mock sync strategy for testing."""
def __init__(self):
self.running = True
def run_strategy(self):
"""Sync run_strategy method."""
pass
class TestExecutorInitialization:
"""Test Executor class initialization."""
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_creates_empty_strategy_runners(self, mock_config, mock_signal):
"""Test Executor init creates empty strategy_runners list."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.strategy_runners == []
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_creates_empty_coroutines(self, mock_config, mock_signal):
"""Test Executor init creates empty coroutines dict."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.coroutines == {}
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_creates_empty_coroutine_threads(self, mock_config, mock_signal):
"""Test Executor init creates empty coroutine_threads dict."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.coroutine_threads == {}
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_creates_empty_functions(self, mock_config, mock_signal):
"""Test Executor init creates empty functions dict."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.functions == {}
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_creates_config(self, mock_config, mock_signal):
"""Test Executor init creates config."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.config is not None
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_sets_timeout_none(self, mock_config, mock_signal):
"""Test Executor init sets timeout to None."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
assert executor.timeout is None
@patch('aiomql.lib.executor.signal')
@patch.object(Config, '__new__')
def test_init_registers_signal_handler(self, mock_config, mock_signal):
"""Test Executor init registers SIGINT handler."""
config = MagicMock()
mock_config.return_value = config
executor = Executor()
mock_signal.assert_called()
class TestAddFunction:
"""Test Executor add_function method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
mock_config.return_value = config
return Executor()
def test_add_function_without_kwargs(self, executor):
"""Test add_function without kwargs."""
def my_function():
pass
executor.add_function(function=my_function)
assert my_function in executor.functions
assert executor.functions[my_function] == {}
def test_add_function_with_kwargs(self, executor):
"""Test add_function with kwargs."""
def my_function(a, b):
pass
kwargs = {"a": 1, "b": 2}
executor.add_function(function=my_function, kwargs=kwargs)
assert my_function in executor.functions
assert executor.functions[my_function] == kwargs
def test_add_multiple_functions(self, executor):
"""Test adding multiple functions."""
def func1():
pass
def func2():
pass
executor.add_function(function=func1, kwargs={"x": 1})
executor.add_function(function=func2, kwargs={"y": 2})
assert len(executor.functions) == 2
assert func1 in executor.functions
assert func2 in executor.functions
2026-02-21 07:43:33 +01:00
def test_add_function_none_kwargs_becomes_empty_dict(self, executor):
"""Test add_function with None kwargs defaults to empty dict."""
def my_function():
pass
executor.add_function(function=my_function, kwargs=None)
assert executor.functions[my_function] == {}
def test_add_function_replaces_if_same_key(self, executor):
"""Test add_function overwrites kwargs if same function is added twice."""
def my_function():
pass
executor.add_function(function=my_function, kwargs={"a": 1})
executor.add_function(function=my_function, kwargs={"a": 2})
assert executor.functions[my_function] == {"a": 2}
assert len(executor.functions) == 1
class TestAddCoroutine:
"""Test Executor add_coroutine method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
mock_config.return_value = config
return Executor()
def test_add_coroutine_without_kwargs(self, executor):
"""Test add_coroutine without kwargs."""
async def my_coroutine():
pass
executor.add_coroutine(coroutine=my_coroutine)
assert my_coroutine in executor.coroutines
assert executor.coroutines[my_coroutine] == {}
def test_add_coroutine_with_kwargs(self, executor):
"""Test add_coroutine with kwargs."""
async def my_coroutine(a, b):
pass
kwargs = {"a": 1, "b": 2}
executor.add_coroutine(coroutine=my_coroutine, kwargs=kwargs)
assert my_coroutine in executor.coroutines
assert executor.coroutines[my_coroutine] == kwargs
def test_add_coroutine_on_separate_thread(self, executor):
"""Test add_coroutine with on_separate_thread=True."""
async def my_coroutine():
pass
executor.add_coroutine(coroutine=my_coroutine, on_separate_thread=True)
assert my_coroutine in executor.coroutine_threads
assert my_coroutine not in executor.coroutines
def test_add_coroutine_default_not_separate_thread(self, executor):
"""Test add_coroutine defaults to not separate thread."""
async def my_coroutine():
pass
executor.add_coroutine(coroutine=my_coroutine)
assert my_coroutine in executor.coroutines
assert my_coroutine not in executor.coroutine_threads
2026-02-21 07:43:33 +01:00
def test_add_coroutine_none_kwargs_becomes_empty_dict(self, executor):
"""Test add_coroutine with None kwargs defaults to empty dict."""
async def my_coroutine():
pass
executor.add_coroutine(coroutine=my_coroutine, kwargs=None)
assert executor.coroutines[my_coroutine] == {}
def test_add_coroutine_on_separate_thread_with_kwargs(self, executor):
"""Test add_coroutine on separate thread with kwargs."""
async def my_coroutine(x):
pass
executor.add_coroutine(coroutine=my_coroutine, kwargs={"x": 42}, on_separate_thread=True)
assert my_coroutine in executor.coroutine_threads
assert executor.coroutine_threads[my_coroutine] == {"x": 42}
class TestAddStrategy:
"""Test Executor add_strategy and add_strategies methods."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
mock_config.return_value = config
return Executor()
def test_add_strategy(self, executor):
"""Test add_strategy adds single strategy."""
strategy = MagicMock(spec=Strategy)
executor.add_strategy(strategy=strategy)
assert len(executor.strategy_runners) == 1
assert strategy in executor.strategy_runners
def test_add_multiple_strategies_one_by_one(self, executor):
"""Test adding multiple strategies one by one."""
strategy1 = MagicMock(spec=Strategy)
strategy2 = MagicMock(spec=Strategy)
executor.add_strategy(strategy=strategy1)
executor.add_strategy(strategy=strategy2)
assert len(executor.strategy_runners) == 2
assert strategy1 in executor.strategy_runners
assert strategy2 in executor.strategy_runners
def test_add_strategies_batch(self, executor):
"""Test add_strategies adds multiple strategies at once."""
strategy1 = MagicMock(spec=Strategy)
strategy2 = MagicMock(spec=Strategy)
strategy3 = MagicMock(spec=Strategy)
executor.add_strategies(strategies=(strategy1, strategy2, strategy3))
assert len(executor.strategy_runners) == 3
assert strategy1 in executor.strategy_runners
assert strategy2 in executor.strategy_runners
assert strategy3 in executor.strategy_runners
def test_add_strategies_extends_existing(self, executor):
"""Test add_strategies extends existing strategies."""
strategy1 = MagicMock(spec=Strategy)
strategy2 = MagicMock(spec=Strategy)
executor.add_strategy(strategy=strategy1)
executor.add_strategies(strategies=(strategy2,))
assert len(executor.strategy_runners) == 2
2026-02-21 07:43:33 +01:00
def test_add_strategies_preserves_order(self, executor):
"""Test add_strategies preserves insertion order."""
strategy1 = MagicMock(spec=Strategy)
strategy2 = MagicMock(spec=Strategy)
strategy3 = MagicMock(spec=Strategy)
executor.add_strategies(strategies=(strategy1, strategy2, strategy3))
assert executor.strategy_runners[0] == strategy1
assert executor.strategy_runners[1] == strategy2
assert executor.strategy_runners[2] == strategy3
class TestRunStrategy:
"""Test Executor run_strategy static method."""
def test_run_strategy_async(self):
"""Test run_strategy with async strategy."""
strategy = MockAsyncStrategy()
strategy.run_strategy = AsyncMock()
with patch('asyncio.run') as mock_asyncio_run:
Executor.run_strategy(strategy)
mock_asyncio_run.assert_called_once()
def test_run_strategy_sync(self):
"""Test run_strategy with sync strategy."""
strategy = MockSyncStrategy()
strategy.run_strategy = MagicMock()
with patch('asyncio.run') as mock_asyncio_run:
Executor.run_strategy(strategy)
# asyncio.run should NOT be called for sync
mock_asyncio_run.assert_not_called()
strategy.run_strategy.assert_called_once()
def test_run_strategy_detects_async_correctly(self):
"""Test run_strategy correctly detects async method."""
async_strategy = MockAsyncStrategy()
assert inspect.iscoroutinefunction(async_strategy.run_strategy)
def test_run_strategy_detects_sync_correctly(self):
"""Test run_strategy correctly detects sync method."""
sync_strategy = MockSyncStrategy()
assert not inspect.iscoroutinefunction(sync_strategy.run_strategy)
2026-02-21 07:43:33 +01:00
def test_run_strategy_calls_sync_directly(self):
"""Test run_strategy calls sync strategy's run_strategy directly."""
strategy = MockSyncStrategy()
call_tracker = {"called": False}
original_run = strategy.run_strategy
def tracking_run():
call_tracker["called"] = True
strategy.run_strategy = tracking_run
Executor.run_strategy(strategy)
assert call_tracker["called"] is True
class TestRunCoroutineTasks:
"""Test Executor run_coroutine_tasks method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
mock_config.return_value = config
return Executor()
async def test_run_coroutine_tasks_runs_all(self, executor):
"""Test run_coroutine_tasks runs all coroutines."""
call_tracker = {"coro1": False, "coro2": False}
async def coro1():
call_tracker["coro1"] = True
async def coro2():
call_tracker["coro2"] = True
executor.add_coroutine(coroutine=coro1)
executor.add_coroutine(coroutine=coro2)
await executor.run_coroutine_tasks()
assert call_tracker["coro1"] is True
assert call_tracker["coro2"] is True
async def test_run_coroutine_tasks_passes_kwargs(self, executor):
"""Test run_coroutine_tasks passes kwargs to coroutines."""
received_kwargs = {}
async def my_coro(a, b):
received_kwargs["a"] = a
received_kwargs["b"] = b
executor.add_coroutine(coroutine=my_coro, kwargs={"a": 1, "b": 2})
await executor.run_coroutine_tasks()
assert received_kwargs["a"] == 1
assert received_kwargs["b"] == 2
async def test_run_coroutine_tasks_handles_exception(self, executor):
"""Test run_coroutine_tasks handles exceptions gracefully."""
async def failing_coro():
raise Exception("Test error")
executor.add_coroutine(coroutine=failing_coro)
# Should not raise
await executor.run_coroutine_tasks()
async def test_run_coroutine_tasks_empty(self, executor):
"""Test run_coroutine_tasks with no coroutines."""
# Should not raise
await executor.run_coroutine_tasks()
2026-02-21 07:43:33 +01:00
async def test_run_coroutine_tasks_only_runs_coroutines_not_threads(self, executor):
"""Test run_coroutine_tasks only runs coroutines, not coroutine_threads."""
call_tracker = {"coro": False, "thread_coro": False}
async def regular_coro():
call_tracker["coro"] = True
async def thread_coro():
call_tracker["thread_coro"] = True
executor.add_coroutine(coroutine=regular_coro)
executor.add_coroutine(coroutine=thread_coro, on_separate_thread=True)
await executor.run_coroutine_tasks()
assert call_tracker["coro"] is True
assert call_tracker["thread_coro"] is False
class TestRunCoroutineTask:
"""Test Executor run_coroutine_task static method."""
def test_run_coroutine_task_runs_with_asyncio(self):
"""Test run_coroutine_task uses asyncio.run."""
async def my_coro(x):
return x
with patch('asyncio.run') as mock_asyncio_run:
Executor.run_coroutine_task(my_coro, {"x": 42})
mock_asyncio_run.assert_called_once()
2026-02-21 07:43:33 +01:00
def test_run_coroutine_task_passes_kwargs(self):
"""Test run_coroutine_task passes kwargs to the coroutine."""
received = {}
async def my_coro(a, b):
received["a"] = a
received["b"] = b
with patch('asyncio.run', side_effect=lambda coro: asyncio.get_event_loop().run_until_complete(coro)) as mock_run:
# Just verify the coroutine is called with kwargs
Executor.run_coroutine_task(my_coro, {"a": 1, "b": 2})
mock_run.assert_called_once()
class TestRunFunction:
"""Test Executor run_function static method."""
def test_run_function_calls_function(self):
"""Test run_function calls the function."""
mock_func = MagicMock()
Executor.run_function(mock_func, {})
mock_func.assert_called_once_with()
def test_run_function_passes_kwargs(self):
"""Test run_function passes kwargs."""
mock_func = MagicMock()
kwargs = {"a": 1, "b": "test"}
Executor.run_function(mock_func, kwargs)
mock_func.assert_called_once_with(a=1, b="test")
2026-02-21 07:43:33 +01:00
def test_run_function_with_multiple_kwargs(self):
"""Test run_function with multiple keyword arguments."""
received = {}
def capture_func(**kwargs):
received.update(kwargs)
Executor.run_function(capture_func, {"x": 10, "y": 20, "z": 30})
assert received == {"x": 10, "y": 20, "z": 30}
class TestSigintHandle:
"""Test Executor sigint_handle method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
config.shutdown = False
mock_config.return_value = config
return Executor()
def test_sigint_handle_sets_shutdown(self, executor):
"""Test sigint_handle sets config.shutdown to True."""
executor.sigint_handle(None, None)
assert executor.config.shutdown is True
2026-02-21 07:43:33 +01:00
def test_sigint_handle_accepts_signum_and_frame(self, executor):
"""Test sigint_handle accepts signum and frame parameters."""
mock_frame = MagicMock()
# Should not raise
executor.sigint_handle(2, mock_frame)
assert executor.config.shutdown is True
class TestExit:
"""Test Executor exit method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
config.shutdown = False
config.force_shutdown = False
config.task_queue = MagicMock()
mock_config.return_value = config
2026-02-21 07:43:33 +01:00
exec_ = Executor()
exec_.executor = MagicMock(spec=ThreadPoolExecutor)
return exec_
def test_exit_with_timeout(self, executor):
"""Test exit respects timeout."""
executor.timeout = 0.1
executor.config.shutdown = False
executor.exit()
assert executor.config.shutdown is True
def test_exit_stops_strategies(self, executor):
"""Test exit sets running=False on all strategies."""
strategy1 = MagicMock()
strategy1.running = True
strategy2 = MagicMock()
strategy2.running = True
executor.strategy_runners = [strategy1, strategy2]
executor.timeout = 0.1
executor.exit()
assert strategy1.running is False
assert strategy2.running is False
def test_exit_cancels_task_queue(self, executor):
"""Test exit cancels task queue."""
executor.timeout = 0.1
executor.exit()
executor.config.task_queue.cancel.assert_called_once()
def test_exit_shuts_down_executor(self, executor):
"""Test exit shuts down thread pool executor."""
executor.timeout = 0.1
executor.exit()
executor.executor.shutdown.assert_called_once_with(wait=False, cancel_futures=False)
2026-02-21 07:43:33 +01:00
def test_exit_force_shutdown(self, executor):
"""Test exit with force_shutdown."""
executor.config.force_shutdown = True
executor.timeout = 0.1
2026-02-21 07:43:33 +01:00
with patch('os._exit') as mock_exit:
executor.exit()
mock_exit.assert_called_once_with(1)
def test_exit_on_shutdown_flag(self, executor):
"""Test exit when shutdown is already True."""
executor.config.shutdown = True
executor.exit()
2026-02-21 07:43:33 +01:00
# Should still stop strategies and clean up
executor.config.task_queue.cancel.assert_called_once()
executor.executor.shutdown.assert_called_once_with(wait=False, cancel_futures=False)
2026-02-21 07:43:33 +01:00
def test_exit_no_strategies(self, executor):
"""Test exit with no strategies."""
executor.timeout = 0.1
2026-02-21 07:43:33 +01:00
executor.strategy_runners = []
# Should not raise
executor.exit()
executor.config.task_queue.cancel.assert_called_once()
def test_exit_exception_calls_os_exit(self, executor):
"""Test exit calls os._exit on exception during shutdown."""
executor.config.shutdown = True
executor.config.task_queue.cancel.side_effect = Exception("Cancel error")
with patch('os._exit') as mock_exit:
executor.exit()
mock_exit.assert_called_once_with(1)
2026-02-21 07:43:33 +01:00
def test_exit_timeout_duration(self, executor):
"""Test exit completes within timeout duration."""
executor.timeout = 0.05
executor.config.shutdown = False
start = time.time()
executor.exit()
elapsed = time.time() - start
# Should exit within timeout + small buffer
assert elapsed < 0.3
assert executor.config.shutdown is True
class TestExecute:
"""Test Executor execute method."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
2026-02-21 07:43:33 +01:00
config.shutdown = True # Set shutdown True so exit loop terminates immediately
config.force_shutdown = False
config.task_queue = MagicMock()
mock_config.return_value = config
return Executor()
def test_execute_calculates_workers(self, executor):
"""Test execute calculates minimum workers correctly."""
strategy = MagicMock()
executor.add_strategy(strategy=strategy)
def func():
pass
executor.add_function(function=func)
async def coro():
pass
executor.add_coroutine(coroutine=coro, on_separate_thread=True)
2026-02-21 07:43:33 +01:00
# workers_ = 1 strategy + 1 function + 1 coroutine_thread + 3 = 6
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute(workers=2)
# max(2, 6) = 6
mock_pool.assert_called_once_with(max_workers=6)
def test_execute_uses_minimum_workers(self, executor):
2026-02-21 07:43:33 +01:00
"""Test execute uses at least the calculated minimum workers."""
# With no strategies/functions/threads, workers_ = 0 + 0 + 0 + 3 = 3
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute(workers=1)
# max(1, 3) = 3
mock_pool.assert_called_once_with(max_workers=3)
def test_execute_respects_custom_workers(self, executor):
"""Test execute uses custom workers when larger than calculated."""
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute(workers=20)
# max(20, 3) = 20
mock_pool.assert_called_once_with(max_workers=20)
def test_execute_default_workers(self, executor):
"""Test execute default workers parameter is 5."""
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute()
# max(5, 3) = 5
mock_pool.assert_called_once_with(max_workers=5)
def test_execute_submits_strategies(self, executor):
"""Test execute submits each strategy to the thread pool."""
strategy1 = MagicMock()
strategy2 = MagicMock()
executor.add_strategy(strategy=strategy1)
executor.add_strategy(strategy=strategy2)
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute()
# Check strategies were submitted
submit_calls = mock_tpe.submit.call_args_list
strategy_calls = [c for c in submit_calls if len(c.args) >= 2 and c.args[0] == executor.run_strategy]
assert len(strategy_calls) == 2
def test_execute_submits_functions(self, executor):
"""Test execute submits functions to the thread pool."""
def my_func(x):
pass
executor.add_function(function=my_func, kwargs={"x": 1})
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute()
# Check function was submitted
submit_calls = mock_tpe.submit.call_args_list
func_calls = [c for c in submit_calls if len(c.args) >= 1 and c.args[0] == my_func]
assert len(func_calls) == 1
def test_execute_submits_coroutine_threads(self, executor):
"""Test execute submits coroutine threads to the thread pool."""
async def my_coro():
pass
executor.add_coroutine(coroutine=my_coro, on_separate_thread=True)
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
2026-02-21 07:43:33 +01:00
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
2026-02-21 07:43:33 +01:00
executor.execute()
2026-02-21 07:43:33 +01:00
# Check coroutine thread was submitted
submit_calls = mock_tpe.submit.call_args_list
coro_thread_calls = [c for c in submit_calls if len(c.args) >= 1 and c.args[0] == executor.run_coroutine_task]
assert len(coro_thread_calls) == 1
def test_execute_submits_coroutine_tasks(self, executor):
"""Test execute submits run_coroutine_tasks via asyncio.run."""
async def my_coro():
pass
executor.add_coroutine(coroutine=my_coro)
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute()
# Check asyncio.run was submitted for coroutine tasks
submit_calls = mock_tpe.submit.call_args_list
asyncio_calls = [c for c in submit_calls if len(c.args) >= 1 and c.args[0] == asyncio.run]
assert len(asyncio_calls) == 1
def test_execute_sets_executor_attribute(self, executor):
"""Test execute sets the executor attribute on the Executor instance."""
with patch('aiomql.lib.executor.ThreadPoolExecutor') as mock_pool:
mock_tpe = MagicMock()
mock_pool.return_value.__enter__.return_value = mock_tpe
mock_pool.return_value.__exit__ = MagicMock(return_value=None)
executor.execute()
assert executor.executor == mock_tpe
class TestIntegration:
"""Integration tests for Executor."""
@pytest.fixture
def executor(self):
"""Create an Executor for testing."""
with patch('aiomql.lib.executor.signal'):
with patch.object(Config, '__new__') as mock_config:
config = MagicMock()
config.shutdown = False
config.force_shutdown = False
config.task_queue = MagicMock()
mock_config.return_value = config
return Executor()
def test_full_setup(self, executor):
"""Test complete executor setup."""
# Add strategies
strategy1 = MagicMock(spec=Strategy)
strategy2 = MagicMock(spec=Strategy)
executor.add_strategies(strategies=(strategy1, strategy2))
# Add functions
def my_func(x):
pass
executor.add_function(function=my_func, kwargs={"x": 1})
# Add coroutines
async def my_coro():
pass
executor.add_coroutine(coroutine=my_coro)
async def my_thread_coro():
pass
executor.add_coroutine(coroutine=my_thread_coro, on_separate_thread=True)
assert len(executor.strategy_runners) == 2
assert len(executor.functions) == 1
assert len(executor.coroutines) == 1
assert len(executor.coroutine_threads) == 1
def test_async_and_sync_strategies(self, executor):
"""Test executor handles both async and sync strategies."""
async_strategy = MockAsyncStrategy()
sync_strategy = MockSyncStrategy()
executor.add_strategy(strategy=async_strategy)
executor.add_strategy(strategy=sync_strategy)
assert len(executor.strategy_runners) == 2
# Both should be runnable via run_strategy
with patch('asyncio.run'):
# Should not raise for either
Executor.run_strategy(async_strategy)
Executor.run_strategy(sync_strategy)
async def test_coroutines_with_different_kwargs(self, executor):
"""Test running coroutines with different kwargs."""
results = []
async def collector(value):
results.append(value)
executor.add_coroutine(coroutine=collector, kwargs={"value": 1})
executor.add_coroutine(coroutine=collector, kwargs={"value": 2})
2026-02-21 07:43:33 +01:00
# Note: dicts can't have duplicate keys, so second call overwrites first
await executor.run_coroutine_tasks()
2026-02-21 07:43:33 +01:00
# Only the last kwargs will be used
assert 2 in results
def test_timeout_functionality(self, executor):
"""Test timeout functionality in exit."""
executor.timeout = 0.05
executor.executor = MagicMock(spec=ThreadPoolExecutor)
start = time.time()
executor.exit()
elapsed = time.time() - start
# Should exit within timeout + small buffer
assert elapsed < 0.2
assert executor.config.shutdown is True
2026-02-21 07:43:33 +01:00
def test_sigint_then_exit(self, executor):
"""Test SIGINT handler followed by exit."""
executor.executor = MagicMock(spec=ThreadPoolExecutor)
strategy = MagicMock()
strategy.running = True
executor.add_strategy(strategy=strategy)
# Simulate SIGINT
executor.sigint_handle(2, None)
assert executor.config.shutdown is True
# Now exit should process immediately
executor.exit()
assert strategy.running is False
executor.config.task_queue.cancel.assert_called_once()