conftest.py
4.22 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
"""Shared pytest setup for the repository test suite."""
import atexit
import importlib
import shutil
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
import pytest
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
@pytest.fixture()
def isolated_research_task_store(monkeypatch: pytest.MonkeyPatch):
"""Provide an isolated research task store and patch app-level globals to use it."""
import backend.research_routes as research_routes
import backend.research_tasks as research_tasks_backend
import services.application.analysis.analysis_service as analysis_service_module
import services.application.analysis.run_store as analysis_run_store_module
temp_dir = Path(tempfile.mkdtemp(prefix="web-api-test-"))
store_path = temp_dir / "venue_research_tasks.json"
analysis_runs_path = temp_dir / "analysis_runs.json"
fresh_store = research_tasks_backend.ResearchTaskStore(store_path)
monkeypatch.setattr(fresh_store, "_save_locked", lambda: None, raising=False)
fresh_analysis_run_store = analysis_run_store_module.InMemoryAnalysisRunStore()
monkeypatch.setattr(research_tasks_backend, "RESEARCH_TASKS_PATH", store_path, raising=False)
monkeypatch.setattr(research_tasks_backend, "research_task_service", fresh_store, raising=False)
monkeypatch.setattr(analysis_run_store_module, "ANALYSIS_RUNS_PATH", analysis_runs_path, raising=False)
monkeypatch.setattr(analysis_run_store_module, "ANALYSIS_RUN_STORE", fresh_analysis_run_store, raising=False)
monkeypatch.setattr(analysis_service_module, "ANALYSIS_RUN_STORE", fresh_analysis_run_store, raising=False)
monkeypatch.setattr(
research_routes,
"RESEARCH_TASK_APP_SERVICE",
research_routes.ResearchTaskService(fresh_store),
raising=False,
)
fresh_analysis_query_service = research_routes.AnalysisRunQueryService(
analysis_run_store=fresh_analysis_run_store,
)
monkeypatch.setattr(
research_routes,
"RESEARCH_ANALYSIS_APP_SERVICE",
fresh_analysis_query_service,
raising=False,
)
monkeypatch.setattr(
research_routes,
"_RESEARCH_ANALYSIS_APP_SERVICE",
fresh_analysis_query_service,
raising=False,
)
monkeypatch.setattr(
research_routes,
"_DEFAULT_RESEARCH_ANALYSIS_APP_SERVICE",
None,
raising=False,
)
try:
yield SimpleNamespace(
temp_dir=temp_dir,
store=fresh_store,
store_path=store_path,
analysis_run_store=fresh_analysis_run_store,
analysis_runs_path=analysis_runs_path,
)
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
@pytest.fixture()
def reload_web_api_app():
"""Reload ``apps.web_api.app`` and always unregister any atexit cleanup handler."""
loaded_modules = []
def _reload():
from apps.web_api.runtime.engine_registry import ENGINE_RUNTIME_REGISTRY
from apps.web_api.runtime.process_registry import PROCESS_RUNTIME_REGISTRY, build_default_process_table
from apps.web_api.runtime.system_state import SYSTEM_STATE_REGISTRY
from apps.web_api.runtime.task_runtime_store import TASK_RUNTIME_STORE
ENGINE_RUNTIME_REGISTRY.reset()
PROCESS_RUNTIME_REGISTRY.table.clear()
PROCESS_RUNTIME_REGISTRY.table.update(build_default_process_table())
SYSTEM_STATE_REGISTRY.set_state(started=False, starting=False)
SYSTEM_STATE_REGISTRY.clear_shutdown_request()
for task_id in list(TASK_RUNTIME_STORE.snapshot_all()):
TASK_RUNTIME_STORE.clear_task(task_id)
module = importlib.import_module("apps.web_api.app")
module = importlib.reload(module)
module.app.config.update(TESTING=True)
loaded_modules.append(module)
return module
yield _reload
for module in reversed(loaded_modules):
app_module = getattr(module, "_APP_MODULE", None)
cleanup_handler = getattr(app_module, "cleanup_handler", None)
if cleanup_handler is None:
continue
try:
atexit.unregister(cleanup_handler)
except Exception:
pass