busya
54a0317e98
- executors: 13 tests, 100% coverage (init/shutdown/run_blocking/run_cpu_blocking) - fileio_utils: 33 tests, 45% coverage (sanitize_filename, get_filename_from_headers, calculate_crc32, create_temp_file, remove_empty_directories, create_dashboard_export, consolidate_archive_folders) - client_registry: 14 tests, 92% coverage (get_client, get_superset_client, get_semaphore, get_auth_lock, shutdown) - cot_logger: 24 tests, 100% coverage (seed/set/get trace_id, push/pop span, structured log with markers, MarkerLogger proxy) - encryption: 12 tests, 100% coverage (key validation, encrypt/decrypt cycle) - rate_limiter: 9 tests, 100% coverage (ban logic, window pruning, per-IP isolation) - auth/logger: 10 tests, 100% coverage (_mask_details, log_security_event)
188 lines
7.6 KiB
Python
188 lines
7.6 KiB
Python
# #region Test.BlockingExecutors [C:3] [TYPE Module] [SEMANTICS test,async,executors,blocking,threadpool]
|
|
# @BRIEF Tests for core/utils/executors.py — init, shutdown, run_blocking, run_cpu_blocking.
|
|
# @RELATION BINDS_TO -> [BlockingExecutorsModule]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
import asyncio
|
|
import pytest
|
|
|
|
|
|
class TestInitShutdown:
|
|
"""init_executors and shutdown_executors."""
|
|
|
|
def test_init_creates_executors(self):
|
|
from src.core.utils.executors import init_executors
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor") as mock_tpe:
|
|
init_executors(db_workers=5, file_workers=3, git_workers=2)
|
|
assert mock_tpe.call_count == 3
|
|
|
|
def test_shutdown_without_init(self):
|
|
# shutdown should not raise when executors are None
|
|
from src.core.utils.executors import shutdown_executors
|
|
shutdown_executors()
|
|
|
|
def test_shutdown_calls_shutdown(self):
|
|
from src.core.utils.executors import init_executors, shutdown_executors
|
|
import src.core.utils.executors as exec_mod
|
|
with patch.object(exec_mod, "_db_executor", None), \
|
|
patch.object(exec_mod, "_file_executor", None), \
|
|
patch.object(exec_mod, "_git_executor", None):
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor") as mock_tpe:
|
|
mock_executor = MagicMock()
|
|
mock_tpe.return_value = mock_executor
|
|
init_executors()
|
|
shutdown_executors()
|
|
assert mock_executor.shutdown.call_count >= 1
|
|
for call in mock_executor.shutdown.call_args_list:
|
|
assert call == unittest.mock.call(wait=True, cancel_futures=True)
|
|
|
|
def test_shutdown_no_wait(self):
|
|
from src.core.utils.executors import init_executors, shutdown_executors
|
|
import src.core.utils.executors as exec_mod
|
|
with patch.object(exec_mod, "_db_executor", None), \
|
|
patch.object(exec_mod, "_file_executor", None), \
|
|
patch.object(exec_mod, "_git_executor", None):
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor") as mock_tpe:
|
|
mock_executor = MagicMock()
|
|
mock_tpe.return_value = mock_executor
|
|
init_executors()
|
|
shutdown_executors(wait=False, cancel_futures=False)
|
|
assert mock_executor.shutdown.call_count >= 1
|
|
for call in mock_executor.shutdown.call_args_list:
|
|
assert call == unittest.mock.call(wait=False, cancel_futures=False)
|
|
|
|
|
|
class TestGetExecutor:
|
|
"""_get_executor — returns executor and semaphore by kind."""
|
|
|
|
def test_db_kind(self):
|
|
from src.core.utils.executors import _get_executor, init_executors, shutdown_executors
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
executor, sem = _get_executor("db")
|
|
assert executor is not None
|
|
assert sem is not None
|
|
shutdown_executors()
|
|
|
|
def test_file_kind(self):
|
|
from src.core.utils.executors import _get_executor, init_executors, shutdown_executors
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
executor, sem = _get_executor("file")
|
|
assert executor is not None
|
|
assert sem is not None
|
|
shutdown_executors()
|
|
|
|
def test_git_kind(self):
|
|
from src.core.utils.executors import _get_executor, init_executors, shutdown_executors
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
executor, sem = _get_executor("git")
|
|
assert executor is not None
|
|
assert sem is not None
|
|
shutdown_executors()
|
|
|
|
def test_unknown_kind_raises(self):
|
|
from src.core.utils.executors import _get_executor, init_executors, shutdown_executors
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
with pytest.raises(ValueError, match="Unknown executor kind"):
|
|
_get_executor("unknown")
|
|
shutdown_executors()
|
|
|
|
|
|
class TestRunBlocking:
|
|
"""run_blocking — execute blocking function in named executor."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runs_function(self):
|
|
from src.core.utils.executors import run_blocking, init_executors, shutdown_executors
|
|
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor") as mock_tpe:
|
|
mock_executor = MagicMock()
|
|
mock_tpe.return_value = mock_executor
|
|
init_executors()
|
|
|
|
# Make run_in_executor work
|
|
mock_loop = AsyncMock()
|
|
mock_loop.run_in_executor.return_value = "result"
|
|
get_running_loop = asyncio.get_running_loop
|
|
|
|
with patch.object(asyncio, "get_running_loop", return_value=mock_loop):
|
|
result = await run_blocking("db", lambda: "hello")
|
|
assert result == "result"
|
|
|
|
shutdown_executors()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_semaphore_acquisition_failure(self):
|
|
from src.core.utils.executors import run_blocking, init_executors, shutdown_executors
|
|
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors(queue_timeout=0.1)
|
|
|
|
with (
|
|
patch("src.core.utils.executors._db_semaphore") as mock_sem,
|
|
pytest.raises(TimeoutError, match="queue timeout"),
|
|
):
|
|
mock_sem.acquire.side_effect = TimeoutError()
|
|
await run_blocking("db", lambda: "hello", timeout=0.1)
|
|
|
|
shutdown_executors()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_semaphore_release_on_success(self):
|
|
from src.core.utils.executors import run_blocking, init_executors, shutdown_executors
|
|
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
mock_loop = AsyncMock()
|
|
mock_loop.run_in_executor.return_value = "done"
|
|
|
|
# Track semaphore acquire/release
|
|
with patch.object(asyncio, "get_running_loop", return_value=mock_loop):
|
|
result = await run_blocking("file", lambda: "work")
|
|
assert result == "done"
|
|
|
|
shutdown_executors()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancelled_error_propagates(self):
|
|
from src.core.utils.executors import run_blocking, init_executors, shutdown_executors
|
|
|
|
with patch("src.core.utils.executors.ThreadPoolExecutor"):
|
|
init_executors()
|
|
mock_loop = AsyncMock()
|
|
mock_loop.run_in_executor.side_effect = asyncio.CancelledError()
|
|
|
|
with (
|
|
patch.object(asyncio, "get_running_loop", return_value=mock_loop),
|
|
pytest.raises(asyncio.CancelledError),
|
|
):
|
|
await run_blocking("git", lambda: "fail")
|
|
|
|
shutdown_executors()
|
|
|
|
|
|
class TestRunCpuBlocking:
|
|
"""run_cpu_blocking — CPU-bound work in default executor."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_runs_in_default_executor(self):
|
|
from src.core.utils.executors import run_cpu_blocking
|
|
|
|
mock_loop = AsyncMock()
|
|
mock_loop.run_in_executor.return_value = 42
|
|
|
|
with patch.object(asyncio, "get_running_loop", return_value=mock_loop):
|
|
result = await run_cpu_blocking(lambda: 42)
|
|
assert result == 42
|
|
mock_loop.run_in_executor.assert_called_once_with(None, mock_loop.run_in_executor.call_args[0][1])
|
|
# #endregion Test.BlockingExecutors
|