d3ro-voice/apps/desktop/sidecar/tests/test_sidecar_redteam_r3_1.py

"""
사이드카 회귀 테스트 (r3-1).

- models-dir 을 쓰는 설치본에서 받지 않은 모델을 /load 하면 HF 에서 몰래 내려받지 않고 404 로 거부하며,
  쓰던 기본 모델을 내리지 않는다.
- 클라이언트가 떠난 전사(cancel)는 세그먼트 반복을 멈춘다 — 끝까지 디코딩하며 다음 요청을 막지 않는다.

실행 (apps/desktop/sidecar 에서):
    .venv/Scripts/python.exe -m unittest discover -s tests -v
"""

from __future__ import annotations

import sys
import tempfile
import threading
import types
import unittest
from pathlib import Path
from typing import Iterator
from unittest import mock

SIDECAR_DIR = Path(__file__).resolve().parent.parent
if str(SIDECAR_DIR) not in sys.path:
    sys.path.insert(0, str(SIDECAR_DIR))

import numpy as np  # noqa: E402
from fastapi.testclient import TestClient  # noqa: E402

import main  # noqa: E402


class _Segment:
    def __init__(self, text: str) -> None:
        self.text = text
        self.start = 0.0
        self.end = 1.0
        self.avg_logprob = -0.1


class _Info:
    language = "ko"


class _FakeModel:
    created: list["_FakeModel"] = []

    def __init__(self, source: str, **_kwargs: object) -> None:
        self.source = source
        _FakeModel.created.append(self)

    def transcribe(self, _audio: np.ndarray, **_kwargs: object) -> tuple[Iterator[_Segment], _Info]:
        def gen() -> Iterator[_Segment]:
            for index in range(5):
                yield _Segment(f"segment {index}")

        return gen(), _Info()


class ImplicitDownloadGuardTest(unittest.TestCase):
    def setUp(self) -> None:
        _FakeModel.created = []
        self._tmp = tempfile.TemporaryDirectory()
        main._models_dir = Path(self._tmp.name)
        main._gpu_available = False
        main._gpu_choice = None
        main._aux_models.clear()
        main._model_devices.clear()
        self.previous = _FakeModel("turbo")
        main._model = self.previous  # type: ignore[assignment]
        main._model_id = "large-v3-turbo"

        fake_utils = types.ModuleType("faster_whisper.utils")

        def download_model(model_id: str, local_files_only: bool = False, **_kwargs: object) -> str:
            if local_files_only:
                raise FileNotFoundError(f"{model_id} not in cache")
            raise AssertionError("must never download implicitly")

        fake_utils.download_model = download_model  # type: ignore[attr-defined]
        fake_module = types.ModuleType("faster_whisper")
        fake_module.WhisperModel = _FakeModel  # type: ignore[attr-defined]
        fake_module.utils = fake_utils  # type: ignore[attr-defined]
        self._patch = mock.patch.dict(
            sys.modules, {"faster_whisper": fake_module, "faster_whisper.utils": fake_utils}
        )
        self._patch.start()
        self.client = TestClient(main.app)

    def tearDown(self) -> None:
        self._patch.stop()
        self._tmp.cleanup()
        main._models_dir = None
        main._model = None
        main._model_id = None
        main._aux_models.clear()
        main._model_devices.clear()

    def test_uninstalled_model_is_rejected_without_unloading_primary(self) -> None:
        res = self.client.post("/load", json={"model_id": "large-v3"})
        self.assertEqual(res.status_code, 404)
        self.assertEqual(res.json()["code"], "model_not_installed")
        self.assertIs(main._model, self.previous)
        self.assertEqual(main._model_id, "large-v3-turbo")
        self.assertEqual(len(_FakeModel.created), 1)

    def test_installed_model_in_models_dir_loads(self) -> None:
        model_dir = Path(self._tmp.name) / "small"
        model_dir.mkdir()
        (model_dir / "model.bin").write_bytes(b"x")
        with mock.patch.object(main, "_probe_model"):
            res = self.client.post("/load", json={"model_id": "small"})
        self.assertEqual(res.status_code, 200)
        self.assertEqual(main._model_id, "small")


class TranscriptionCancelTest(unittest.TestCase):
    def test_cancelled_transcription_stops_iterating_segments(self) -> None:
        cancel = threading.Event()
        cancel.set()
        with self.assertRaises(main.TranscriptionCancelled):
            main._run_transcription(_FakeModel("x"), np.zeros(16000, dtype=np.float32), {}, cancel)

    def test_uncancelled_transcription_consumes_all_segments(self) -> None:
        segments, text, language = main._run_transcription(
            _FakeModel("x"), np.zeros(16000, dtype=np.float32), {}, threading.Event()
        )
        self.assertEqual(len(segments), 5)
        self.assertEqual(language, "ko")
        self.assertTrue(text.startswith("segment 0"))


if __name__ == "__main__":
    unittest.main()