Обновление возит только папку app, поэтому переход на детектор речи потребовал файла, которого на машине нет - и сервис узнал об этом на первой же задаче: отказ выглядел сбоем обработки, хотя это незаконченная установка. Причём download_models.py лежит вне app и тоже был старым, так что штатный способ докачки не помог бы. Список моделей и адреса переехали в app/models.py. Мелкие докачиваются при запуске, крупная модель распознавания - нет: сотни мегабайт без спроса тянуть нельзя. Нехватка видна в /health полем missing_models. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
60 lines
2.6 KiB
Python
60 lines
2.6 KiB
Python
"""Модели не возит обновление - сервис должен сам замечать нехватку."""
|
|
from pathlib import Path
|
|
|
|
from app.models import DOWNLOADABLE, ensure_models, missing_models, required_models
|
|
|
|
|
|
class TestRequired:
|
|
def test_vad_needed_without_diarization(self):
|
|
names = [p.name for p in required_models(Path("m"), diarize=False)]
|
|
assert "silero_vad.onnx" in names
|
|
assert "nemo_en_titanet_small.onnx" not in names
|
|
|
|
def test_diarization_needs_its_own_models(self):
|
|
names = [p.name for p in required_models(Path("m"), diarize=True)]
|
|
assert "model.onnx" in names and "nemo_en_titanet_small.onnx" in names
|
|
assert "silero_vad.onnx" not in names
|
|
|
|
def test_asr_needed_always(self):
|
|
for diarize in (True, False):
|
|
paths = required_models(Path("m"), diarize)
|
|
assert any("gigaam" in str(p) for p in paths)
|
|
|
|
|
|
class TestMissing:
|
|
def test_reports_absent_files(self, tmp_path):
|
|
assert missing_models(tmp_path, diarize=False)
|
|
|
|
def test_empty_when_everything_on_disk(self, tmp_path):
|
|
(tmp_path / "gigaam").mkdir()
|
|
(tmp_path / "gigaam" / "config.json").write_text("{}", encoding="utf-8")
|
|
(tmp_path / "silero_vad.onnx").write_bytes(b"x")
|
|
assert missing_models(tmp_path, diarize=False) == []
|
|
|
|
|
|
class TestEnsure:
|
|
def test_large_model_is_not_downloaded_silently(self, tmp_path):
|
|
"""Сотни мегабайт без спроса тянуть нельзя - только сообщаем."""
|
|
assert "gigaam" not in " ".join(DOWNLOADABLE)
|
|
left = ensure_models(tmp_path, diarize=False,
|
|
fetch=lambda url, dst: dst.write_bytes(b"x"))
|
|
assert any("gigaam" in str(p) for p in left)
|
|
|
|
def test_download_failure_is_reported_not_raised(self, tmp_path):
|
|
"""Сеть может быть недоступна - это не повод не стартовать."""
|
|
def boom(url, dst):
|
|
raise OSError("сети нет")
|
|
|
|
left = ensure_models(tmp_path, diarize=False, fetch=boom)
|
|
assert any(p.name == "silero_vad.onnx" for p in left)
|
|
|
|
def test_partial_file_does_not_become_a_model(self, tmp_path):
|
|
"""Оборванная закачка не должна выглядеть готовой моделью."""
|
|
def half(url, dst):
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
(dst.parent / (dst.name + ".part")).write_bytes(b"half")
|
|
raise OSError("оборвалось")
|
|
|
|
ensure_models(tmp_path, diarize=False, fetch=half)
|
|
assert not (tmp_path / "silero_vad.onnx").exists()
|