operating-room-monitor-server/tests/test_app_integration.py

"""集成测试：通过真实的 ``create_app()`` + ``TestClient`` 走通 start/pending/resolve/end/result 全链路。

与 ``tests/test_api_contract.py`` 不同，这里不用 ``dependency_overrides`` 替换整个
``SurgeryPipeline``；而是通过 ``app.state.container`` 注入一个「stubbed session manager」，
其余 pipeline/voice/repository 组件都保持真实实现，并使用 in-memory SQLite 作为会话工厂。

这样可以覆盖：
1. `create_app()` 的 CORS / demo_orchestrator 路径挂载、lifespan 启动/关闭流程。
2. API → Pipeline → Session Registry → Repository 的真实调用链。
3. durable fallback 目录被 ``ArchivePersister`` 写入/清理的真实路径。

所有会与外界交互的边界（RTSP/海康/MinIO/百度）通过容器中的 stub 对象隔离。
"""

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator
from datetime import datetime, timezone
from typing import Any

import pytest
import pytest_asyncio
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

import app.db.models  # noqa: F401  # register ORM tables on Base.metadata
import main as main_module
from app.baked import pipeline as bp
from app.db.base import Base
from app.dependencies import AppContainer, build_container
from app.domain.consumption import SurgeryConsumptionStored
from app.services.video.session_registry import (
    PendingConsumableConfirmation,
    RunningSurgery,
    SurgerySessionState,
)


class _StubCameraSessionManager:
    """仅实现测试链路必需方法；其余方法委托 ``_registry`` / ``_archive``（由真实组件提供）。

    这样 ``SurgeryPipeline`` / ``VoiceConfirmationService`` 读到的接口与真实 CameraSessionManager
    等价，不需要真实 RTSP 或推理流水线。
    """

    def __init__(self, real: Any) -> None:
        self._real = real
        self._registry = real._registry
        self._archive = real._archive

    async def start_surgery(
        self,
        surgery_id: str,
        camera_ids: list[str],
        candidate_consumables: list[str],
    ) -> None:
        if self._registry.has_active(surgery_id):
            from app.surgery_errors import SurgeryPipelineError

            raise SurgeryPipelineError(
                "RECORDING_CANNOT_START",
                "该手术已在录制中，请勿重复开始。",
            )
        state = SurgerySessionState(
            candidate_consumables=list(candidate_consumables),
            name_to_code={},
        )
        state.ready.set()
        run = RunningSurgery(
            stop_event=asyncio.Event(), state=state, tasks=[]
        )
        await self._registry.register(surgery_id, run)

    def set_voice_terminal_id(self, surgery_id: str, terminal_id: str | None) -> None:
        self._real.set_voice_terminal_id(surgery_id, terminal_id)

    async def stop_surgery(
        self, surgery_id: str, *, require_active: bool = True
    ) -> str | None:
        run = await self._registry.unregister(surgery_id)
        if run is None:
            if require_active:
                from app.surgery_errors import SurgeryPipelineError

                raise SurgeryPipelineError(
                    "RECORDING_NOT_STOPPED",
                    "停录未能完成：当前没有该手术的活跃录制会话。",
                )
            return None
        voice_tid = run.state.voice_terminal_id
        details = list(run.state.details)
        await self._archive.persist_or_archive(surgery_id, details)
        return voice_tid

    def __getattr__(self, name: str) -> Any:
        return getattr(self._real, name)


class _StubVoiceService:
    """屏蔽 MinIO/百度调用；保留 ``synthesize_prompt_to_mp3`` 与 ``resolve_from_wav`` 的最小语义。"""

    def __init__(self, real: Any) -> None:
        self._real = real
        self._sessions = real._sessions

    def synthesize_prompt_to_mp3(self, prompt_text: str) -> bytes:
        return b"MP3-FAKE-" + prompt_text.encode("utf-8", errors="replace")

    async def resolve_from_wav(
        self,
        *,
        surgery_id: str,
        confirmation_id: str,
        wav_bytes: bytes,
        filename: str,
        content_type: str | None,
    ) -> Any:
        from app.services.voice_resolution import VoiceResolveResult
        from app.surgery_errors import SurgeryPipelineError

        pending = self._sessions.get_pending_confirmation_by_id(
            surgery_id, confirmation_id
        )
        if pending is None:
            raise SurgeryPipelineError(
                "CONFIRMATION_NOT_FOUND",
                "未找到该待确认项或已处理。",
            )
        label = (pending.options[0][0] if pending.options else None) or (
            pending.model_top1_label
        )
        await self._sessions.resolve_pending_confirmation(
            surgery_id,
            confirmation_id,
            chosen_label=label,
            rejected=False,
        )
        return VoiceResolveResult(
            resolved_label=label,
            rejected=False,
            asr_text="第一个",
            audio_object_key=f"stub/{surgery_id}/{confirmation_id}.wav",
            message="确认成功（stub）",
        )


@pytest_asyncio.fixture
async def sqlite_factory() -> AsyncGenerator[async_sessionmaker[AsyncSession], None]:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    factory = async_sessionmaker(
        engine,
        class_=AsyncSession,
        expire_on_commit=False,
        autoflush=False,
        autobegin=False,
    )
    yield factory
    await engine.dispose()


@pytest.fixture
def integration_client(
    monkeypatch: pytest.MonkeyPatch,
    sqlite_factory: async_sessionmaker[AsyncSession],
    tmp_path,
) -> TestClient:
    async def _noop() -> None:
        return None

    monkeypatch.setattr(main_module, "check_database", _noop)

    class _FakeEngine:
        async def dispose(self) -> None:
            return None

    monkeypatch.setattr(main_module, "engine", _FakeEngine())

    from app.config import settings as real_settings

    monkeypatch.setattr(
        bp,
        "ARCHIVE_PERSIST_DURABLE_FALLBACK_DIR",
        str(tmp_path / "pending_archive"),
    )

    def _stubbed_build_container(*args, **kwargs) -> AppContainer:
        container = build_container(real_settings, session_factory=sqlite_factory)
        container.camera_session_manager = _StubCameraSessionManager(
            container.camera_session_manager
        )
        container.surgery_pipeline._sessions = container.camera_session_manager
        container.voice_confirmation_service._sessions = (
            container.camera_session_manager._registry
        )
        container.surgery_pipeline._voice = _StubVoiceService(
            container.surgery_pipeline
        )
        return container

    monkeypatch.setattr(main_module, "build_container", _stubbed_build_container)

    async def _instant_sleep(_d: float) -> None:
        return None

    monkeypatch.setattr("app.api.asyncio.sleep", _instant_sleep)

    app = main_module.create_app()
    with TestClient(app) as client:
        yield client


def _enqueue_pending(
    client: TestClient, *, surgery_id: str
) -> str:
    container: AppContainer = client.app.state.container
    run = container.camera_session_manager._registry.get_running(surgery_id)
    assert run is not None
    cid = "cid-integration-1"
    pending = PendingConsumableConfirmation(
        id=cid,
        status="pending",
        options=[("纱布", 0.42)],
        prompt_text="请确认：是否为纱布",
        created_at=datetime.now(timezone.utc),
        model_top1_label="纱布",
        model_top1_confidence=0.42,
    )
    run.state.pending_fifo.append(cid)
    run.state.pending_by_id[cid] = pending
    return cid


def test_full_flow_start_pending_resolve_end_result(
    integration_client: TestClient,
) -> None:
    client = integration_client
    surgery_id = "100001"

    r = client.post(
        "/client/surgeries/start",
        json={
            "surgery_id": surgery_id,
            "camera_ids": ["cam1"],
            "candidate_consumables": ["纱布"],
        },
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "accepted"

    r2 = client.get(f"/client/surgeries/{surgery_id}/pending-confirmation")
    assert r2.status_code == 404, r2.text

    cid = _enqueue_pending(client, surgery_id=surgery_id)

    r3 = client.get(f"/client/surgeries/{surgery_id}/pending-confirmation")
    assert r3.status_code == 200, r3.text
    body3 = r3.json()
    assert body3["confirmation_id"] == cid
    import base64

    decoded = base64.b64decode(body3["prompt_audio_mp3_base64"].encode("ascii"))
    assert decoded.startswith(b"MP3-FAKE-")

    r4 = client.post(
        f"/client/surgeries/{surgery_id}/pending-confirmation/{cid}/resolve",
        files={"audio": ("voice.wav", b"RIFFxxxx", "audio/wav")},
    )
    assert r4.status_code == 200, r4.text
    body4 = r4.json()
    assert body4["resolved_label"] == "纱布"
    assert body4["rejected"] is False

    r5 = client.get(f"/client/surgeries/{surgery_id}/result")
    assert r5.status_code == 200, r5.text
    body5 = r5.json()
    assert body5["surgery_id"] == surgery_id
    assert len(body5["details"]) == 1
    row = body5["details"][0]
    assert row["item_name"] == "纱布"
    assert row["qty"] == 1
    assert row["doctor_id"] == "voice"

    r6 = client.post(
        "/client/surgeries/end", json={"surgery_id": surgery_id}
    )
    assert r6.status_code == 200, r6.text

    r7 = client.get(f"/client/surgeries/{surgery_id}/result")
    assert r7.status_code == 200, r7.text
    body7 = r7.json()
    assert len(body7["details"]) == 1
    assert body7["details"][0]["item_name"] == "纱布"


def test_result_not_ready_before_start(integration_client: TestClient) -> None:
    r = integration_client.get("/client/surgeries/999999/result")
    assert r.status_code == 503
    assert r.json()["detail"]["code"] == "RESULT_NOT_READY"


def test_health_endpoint_ok_via_real_app(integration_client: TestClient) -> None:
    r = integration_client.get("/health")
    assert r.status_code == 200
    body = r.json()
    assert body["status"] == "ok"
    assert body["database"] == "connected"