190 lines
7.6 KiB
Python
190 lines
7.6 KiB
Python
"""Тесты для загрузчика конфига плагинов и factory."""
|
||
|
||
import sys
|
||
import types
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from core.plugins.config import PluginsConfig, TranscriberConfig, load_plugins_config
|
||
from core.plugins.factory import UnknownProviderError, create_summarizer, create_transcriber
|
||
from core.plugins.faster_whisper import FasterWhisperCPU, FasterWhisperGPU
|
||
from core.plugins.null import NullSummarizer
|
||
|
||
PLUGINS_YAML = Path(__file__).parent.parent.parent / "config" / "plugins.yaml"
|
||
|
||
|
||
def test_load_plugins_config_from_yaml() -> None:
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
|
||
assert isinstance(config, PluginsConfig)
|
||
assert config.transcriber.enabled is True
|
||
assert config.transcriber.provider == "faster_whisper_cpu"
|
||
assert config.transcriber.model == "small"
|
||
assert config.transcriber.language == "ru"
|
||
assert config.summarizer.provider == "null"
|
||
assert config.summarizer.chunk_minutes == 20
|
||
assert config.chat.enabled is True
|
||
|
||
|
||
def test_factory_instantiates_faster_whisper_transcriber_from_config() -> None:
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
|
||
transcriber = create_transcriber(config.transcriber)
|
||
|
||
assert isinstance(transcriber, FasterWhisperCPU)
|
||
assert transcriber.model_name == "small"
|
||
assert transcriber.language == "ru"
|
||
# Путь — КОНТРАКТ с backend/services/ai_tiers.py (WHISPER_MODELS_ROOT,
|
||
# ADR-004): модель уровня хранится в подкаталоге по имени модели,
|
||
# не в плоском корне тома (см. комментарий в config/plugins.yaml).
|
||
assert transcriber.download_root == "/models/whisper/small"
|
||
|
||
|
||
def test_factory_instantiates_null_summarizer() -> None:
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
|
||
summarizer = create_summarizer(config.summarizer)
|
||
|
||
assert isinstance(summarizer, NullSummarizer)
|
||
assert summarizer.summarize("some transcript") == ""
|
||
|
||
|
||
def test_factory_unknown_transcriber_provider_raises() -> None:
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
config.transcriber.provider = "does-not-exist"
|
||
|
||
with pytest.raises(UnknownProviderError):
|
||
create_transcriber(config.transcriber)
|
||
|
||
|
||
def test_factory_unknown_summarizer_provider_raises() -> None:
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
config.summarizer.provider = "does-not-exist"
|
||
|
||
with pytest.raises(UnknownProviderError):
|
||
create_summarizer(config.summarizer)
|
||
|
||
|
||
def test_factory_creates_faster_whisper_without_importing_it(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""`create_transcriber` для `faster_whisper_cpu` не импортирует пакет `faster_whisper`.
|
||
|
||
Ленивость: тяжёлая зависимость грузится только внутри `transcribe()`, а не
|
||
при инстанцировании плагина фабрикой (важно для API-процесса).
|
||
"""
|
||
monkeypatch.delitem(sys.modules, "faster_whisper", raising=False)
|
||
config = load_plugins_config(PLUGINS_YAML)
|
||
|
||
transcriber = create_transcriber(config.transcriber)
|
||
|
||
assert isinstance(transcriber, FasterWhisperCPU)
|
||
assert "faster_whisper" not in sys.modules
|
||
|
||
|
||
@dataclass
|
||
class _FakeRawSegment:
|
||
"""Сегмент, возвращаемый заглушкой `WhisperModel.transcribe`."""
|
||
|
||
start: float
|
||
end: float
|
||
text: str
|
||
|
||
|
||
class _FakeWhisperModel:
|
||
"""Заглушка `faster_whisper.WhisperModel`, фиксирующая переданные аргументы."""
|
||
|
||
last_init_kwargs: dict[str, Any] | None = None
|
||
last_transcribe_kwargs: dict[str, Any] | None = None
|
||
|
||
def __init__(self, model_size_or_path: str, **kwargs: Any) -> None:
|
||
self.model_size_or_path = model_size_or_path
|
||
_FakeWhisperModel.last_init_kwargs = kwargs
|
||
|
||
def transcribe(self, audio_path: str, **kwargs: Any) -> tuple[list[_FakeRawSegment], object]:
|
||
_FakeWhisperModel.last_transcribe_kwargs = kwargs
|
||
segments = [
|
||
_FakeRawSegment(0.0, 2.0, "длинный сегмент"),
|
||
_FakeRawSegment(2.0, 2.2, "коротыш"), # 0.2с < 0.3с — отбрасывается
|
||
_FakeRawSegment(2.2, 2.6, "ровно на границе"), # 0.4с — остаётся
|
||
]
|
||
return segments, object()
|
||
|
||
|
||
def test_faster_whisper_transcribe_forwards_params_and_filters_short_segments(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""Юнит `FasterWhisperCPU` с замоканным `WhisperModel`: прокидывание
|
||
language/vad_filter/vad_parameters и отбрасывание сегментов короче 0.3с.
|
||
"""
|
||
fake_module = types.ModuleType("faster_whisper")
|
||
fake_module.WhisperModel = _FakeWhisperModel # type: ignore[attr-defined]
|
||
monkeypatch.setitem(sys.modules, "faster_whisper", fake_module)
|
||
monkeypatch.setattr(FasterWhisperCPU, "_model", None)
|
||
|
||
transcriber = FasterWhisperCPU(model="small", language="ru", download_root="/models/whisper")
|
||
result = transcriber.transcribe("audio.ogg", language="ru")
|
||
|
||
assert [segment.text for segment in result] == ["длинный сегмент", "ровно на границе"]
|
||
assert _FakeWhisperModel.last_init_kwargs == {
|
||
"device": "cpu",
|
||
"compute_type": "int8",
|
||
"download_root": "/models/whisper",
|
||
}
|
||
assert _FakeWhisperModel.last_transcribe_kwargs == {
|
||
"language": "ru",
|
||
"vad_filter": True,
|
||
"vad_parameters": {"min_silence_duration_ms": 500},
|
||
}
|
||
|
||
|
||
def test_factory_instantiates_faster_whisper_gpu_transcriber() -> None:
|
||
"""`create_transcriber` для `faster_whisper_gpu` (уровни `medium`/`max`, ADR-004)."""
|
||
config = TranscriberConfig(
|
||
provider="faster_whisper_gpu",
|
||
model="large-v3",
|
||
options={"download_root": "/models/whisper/large-v3", "compute_type": "float16"},
|
||
)
|
||
|
||
transcriber = create_transcriber(config)
|
||
|
||
assert isinstance(transcriber, FasterWhisperGPU)
|
||
assert transcriber.model_name == "large-v3"
|
||
assert transcriber.download_root == "/models/whisper/large-v3"
|
||
assert transcriber.compute_type == "float16"
|
||
assert transcriber.device == "cuda"
|
||
|
||
|
||
def test_faster_whisper_gpu_transcribe_forwards_cuda_device_and_compute_type(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""GPU-плагин прокидывает `device="cuda"`/`compute_type` в `WhisperModel`."""
|
||
fake_module = types.ModuleType("faster_whisper")
|
||
fake_module.WhisperModel = _FakeWhisperModel # type: ignore[attr-defined]
|
||
monkeypatch.setitem(sys.modules, "faster_whisper", fake_module)
|
||
monkeypatch.setattr(FasterWhisperGPU, "_model", None)
|
||
|
||
transcriber = FasterWhisperGPU(model="large-v3", download_root="/models/whisper/large-v3")
|
||
result = transcriber.transcribe("audio.ogg", language="ru")
|
||
|
||
assert [segment.text for segment in result] == ["длинный сегмент", "ровно на границе"]
|
||
assert _FakeWhisperModel.last_init_kwargs == {
|
||
"device": "cuda",
|
||
"compute_type": "float16",
|
||
"download_root": "/models/whisper/large-v3",
|
||
}
|
||
|
||
|
||
def test_cpu_and_gpu_transcribers_have_independent_model_singletons(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""CPU- и GPU-плагины не делят синглтон `_model` — у каждого подкласса свой атрибут."""
|
||
monkeypatch.setattr(FasterWhisperCPU, "_model", "cpu-sentinel")
|
||
monkeypatch.setattr(FasterWhisperGPU, "_model", None)
|
||
|
||
assert FasterWhisperCPU._model == "cpu-sentinel"
|
||
assert FasterWhisperGPU._model is None
|