Vladimir BryzgalovandClaude Opus 5
9dc67bda5c
Локальный FastAPI-сервис поверх GigaAM v3 и sherpa-onnx: приём аудио, очередь задач, разделение по говорящим, постобработка терминов. Доставка на Windows - ZIP со встроенным Python, без установки чего-либо. Обновление кода при запуске тянется из релизов Gitea: меняется только папка app, десятки килобайт вместо всего пакета. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
125 lines
4.8 KiB
Python
125 lines
4.8 KiB
Python
"""Тесты HTTP-слоя. Пайплайн подменяется заглушкой: модели тут не нужны."""
|
|
import sys
|
|
import types
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
TOKEN = "test-token-123"
|
|
|
|
|
|
@pytest.fixture
|
|
def client(tmp_path, monkeypatch):
|
|
"""Поднимает приложение с временным каталогом и заглушкой моделей."""
|
|
config = tmp_path / "config.toml"
|
|
config.write_text(
|
|
f'[server]\nhost="127.0.0.1"\nport=8756\n'
|
|
f'[security]\ntoken="{TOKEN}"\nallow_ips=""\n'
|
|
f'[processing]\nthreads=1\nspeakers=2\nmax_upload_mb=1\n',
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.setenv("TALKSCORE_ASR_HOME", str(tmp_path))
|
|
|
|
# sherpa_onnx и onnx_asr в тестах не нужны - подменяем до импорта приложения
|
|
for name in ("sherpa_onnx", "onnx_asr", "onnxruntime"):
|
|
monkeypatch.setitem(sys.modules, name, types.ModuleType(name))
|
|
|
|
for mod in [m for m in list(sys.modules) if m.startswith("app.")]:
|
|
del sys.modules[mod]
|
|
|
|
import app.config as cfg
|
|
monkeypatch.setattr(cfg, "BASE_DIR", tmp_path)
|
|
import app.main as main
|
|
monkeypatch.setattr(main.pipeline, "ffmpeg", "/bin/true", raising=False)
|
|
main._state["ready"] = True
|
|
main._worker_stop.set() # воркер не должен разбирать очередь в тестах
|
|
|
|
with TestClient(main.app) as c:
|
|
c.headers.update({"Authorization": f"Bearer {TOKEN}"})
|
|
yield c
|
|
|
|
|
|
class TestAuth:
|
|
def test_health_needs_no_token(self, client):
|
|
client.headers.pop("Authorization")
|
|
assert client.get("/health").status_code == 200
|
|
|
|
def test_jobs_require_token(self, client):
|
|
client.headers.pop("Authorization")
|
|
r = client.post("/v1/jobs", files={"file": ("a.wav", b"x")})
|
|
assert r.status_code == 401
|
|
|
|
def test_wrong_token_rejected(self, client):
|
|
client.headers.update({"Authorization": "Bearer nope"})
|
|
r = client.get("/v1/jobs/whatever")
|
|
assert r.status_code == 401
|
|
|
|
def test_valid_token_passes(self, client):
|
|
assert client.get("/v1/jobs/nonexistent").status_code == 404
|
|
|
|
|
|
class TestUpload:
|
|
def test_creates_job(self, client):
|
|
r = client.post("/v1/jobs", files={"file": ("call.mp3", b"fake audio data")})
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["status"] == "queued"
|
|
assert len(body["job_id"]) == 32
|
|
|
|
def test_rejects_empty_file(self, client):
|
|
r = client.post("/v1/jobs", files={"file": ("empty.wav", b"")})
|
|
assert r.status_code == 400
|
|
|
|
def test_rejects_too_large_file(self, client):
|
|
"""Лимит в тестовом конфиге - 1 МБ."""
|
|
r = client.post("/v1/jobs", files={"file": ("big.wav", b"x" * (2 * 1024 * 1024))})
|
|
assert r.status_code == 413
|
|
|
|
def test_speakers_parameter_is_stored(self, client):
|
|
r = client.post("/v1/jobs?speakers=3", files={"file": ("a.wav", b"data")})
|
|
assert r.status_code == 200
|
|
|
|
def test_speakers_out_of_range_rejected(self, client):
|
|
r = client.post("/v1/jobs?speakers=99", files={"file": ("a.wav", b"data")})
|
|
assert r.status_code == 422
|
|
|
|
|
|
class TestJobStatus:
|
|
def test_unknown_job_is_404(self, client):
|
|
assert client.get("/v1/jobs/deadbeef").status_code == 404
|
|
|
|
def test_queued_job_reports_position(self, client):
|
|
job_id = client.post("/v1/jobs", files={"file": ("a.wav", b"data")}).json()["job_id"]
|
|
body = client.get(f"/v1/jobs/{job_id}").json()
|
|
assert body["status"] == "queued"
|
|
assert body["queue_position"] == 0
|
|
|
|
def test_second_job_is_behind_first(self, client):
|
|
client.post("/v1/jobs", files={"file": ("a.wav", b"data")})
|
|
second = client.post("/v1/jobs", files={"file": ("b.wav", b"data")}).json()["job_id"]
|
|
assert client.get(f"/v1/jobs/{second}").json()["queue_position"] == 1
|
|
|
|
def test_delete_removes_job(self, client):
|
|
job_id = client.post("/v1/jobs", files={"file": ("a.wav", b"data")}).json()["job_id"]
|
|
assert client.delete(f"/v1/jobs/{job_id}").status_code == 200
|
|
assert client.get(f"/v1/jobs/{job_id}").status_code == 404
|
|
|
|
|
|
class TestHealth:
|
|
def test_reports_queue_and_threads(self, client):
|
|
body = client.get("/health").json()
|
|
assert "queue" in body and "threads" in body
|
|
|
|
|
|
class TestSchemaExposure:
|
|
"""Схема API не требует токена, поэтому по умолчанию она закрыта."""
|
|
|
|
def test_docs_hidden_by_default(self, client):
|
|
assert client.get("/docs").status_code == 404
|
|
|
|
def test_openapi_hidden_by_default(self, client):
|
|
assert client.get("/openapi.json").status_code == 404
|
|
|
|
def test_redoc_hidden_by_default(self, client):
|
|
assert client.get("/redoc").status_code == 404
|