from __future__ import annotations from contextlib import contextmanager import importlib from pathlib import Path import sys import types import pytest FastAPI = pytest.importorskip("fastapi").FastAPI TestClient = pytest.importorskip("fastapi.testclient").TestClient @pytest.fixture(autouse=True) def _cleanup_question_router_module(): yield sys.modules.pop("deeptutor.api.routers.question", None) class _DummyProcessLogEvent: def __init__(self, **kwargs) -> None: self.data = {"type": "process_log", **kwargs} def to_dict(self): return self.data @contextmanager def _noop_context(*_args, **_kwargs): yield def _package(name: str) -> types.ModuleType: module = types.ModuleType(name) module.__path__ = [] return module def _load_question_router_module(monkeypatch: pytest.MonkeyPatch): sys.modules.pop("deeptutor.api.routers.question", None) fake_agents = _package("deeptutor.agents") fake_agents_question = types.ModuleType("deeptutor.agents.question") fake_agents_question.AgentCoordinator = object fake_agents.question = fake_agents_question monkeypatch.setitem(sys.modules, "deeptutor.agents", fake_agents) monkeypatch.setitem(sys.modules, "deeptutor.agents.question", fake_agents_question) fake_logging = _package("deeptutor.logging") fake_logging.ProcessLogEvent = _DummyProcessLogEvent fake_logging.bind_log_context = _noop_context fake_logging.capture_process_logs = _noop_context fake_logging.current_log_context = lambda: {} monkeypatch.setitem(sys.modules, "deeptutor.logging", fake_logging) fake_config = types.ModuleType("deeptutor.services.config") fake_config.PROJECT_ROOT = Path.cwd() fake_config.load_config_with_main = lambda *_args, **_kwargs: {} monkeypatch.setitem(sys.modules, "deeptutor.services.config", fake_config) fake_llm_package = _package("deeptutor.services.llm") fake_llm_config = types.ModuleType("deeptutor.services.llm.config") fake_llm_config.get_llm_config = lambda: None fake_llm_package.config = fake_llm_config monkeypatch.setitem(sys.modules, "deeptutor.services.llm", fake_llm_package) monkeypatch.setitem(sys.modules, "deeptutor.services.llm.config", fake_llm_config) fake_settings_package = _package("deeptutor.services.settings") fake_interface_settings = types.ModuleType("deeptutor.services.settings.interface_settings") fake_interface_settings.get_ui_language = lambda default="en": default fake_settings_package.interface_settings = fake_interface_settings monkeypatch.setitem(sys.modules, "deeptutor.services.settings", fake_settings_package) monkeypatch.setitem( sys.modules, "deeptutor.services.settings.interface_settings", fake_interface_settings, ) fake_tools = _package("deeptutor.tools") fake_tools_question = types.ModuleType("deeptutor.tools.question") async def _default_mimic_exam_questions(*_args, **_kwargs): return {"success": True} fake_tools_question.mimic_exam_questions = _default_mimic_exam_questions fake_tools.question = fake_tools_question monkeypatch.setitem(sys.modules, "deeptutor.tools", fake_tools) monkeypatch.setitem(sys.modules, "deeptutor.tools.question", fake_tools_question) return importlib.import_module("deeptutor.api.routers.question") def _build_app(router_module) -> FastAPI: app = FastAPI() app.include_router(router_module.router, prefix="/api/v1/question") return app def test_mimic_websocket_accepts_config_and_returns_messages( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: question_router_module = _load_question_router_module(monkeypatch) async def _fake_mimic_exam_questions(*_args, **_kwargs): return {"success": False, "error": "stub mimic failure"} monkeypatch.setattr(question_router_module, "mimic_exam_questions", _fake_mimic_exam_questions) # ``MIMIC_OUTPUT_DIR`` was a module-level constant resolved at import time # (which froze it to the admin path). It's now a per-call helper so the # path follows whichever user is running. Patch the helper instead. monkeypatch.setattr( question_router_module, "_mimic_output_dir", lambda: tmp_path / "mimic_papers" ) with TestClient(_build_app(question_router_module)) as client: with client.websocket_connect("/api/v1/question/mimic") as websocket: websocket.send_json( { "mode": "parsed", "paper_path": str(tmp_path / "paper"), "kb_name": "demo-kb", "max_questions": 3, } ) messages = [websocket.receive_json() for _ in range(3)] assert [message["type"] for message in messages] == ["status", "status", "error"] assert messages[0]["stage"] == "init" assert messages[1]["stage"] == "processing" assert messages[2]["content"] == "stub mimic failure"