Spaces:
Running
Running
| from __future__ import annotations | |
| import asyncio | |
| from pathlib import Path | |
| from types import SimpleNamespace | |
| import pytest | |
| from fastapi import HTTPException | |
| from starlette.requests import Request | |
| from app.api import diarization | |
| def _request(headers: dict[str, str] | None = None) -> Request: | |
| raw_headers = [(key.lower().encode(), value.encode()) for key, value in (headers or {}).items()] | |
| scope = {"type": "http", "method": "POST", "path": "/v1/diarization/diarize", "headers": raw_headers} | |
| return Request(scope) | |
| def test_authorize_rejects_missing_configuration(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.delenv("DIARIZATION_API_KEY", raising=False) | |
| with pytest.raises(HTTPException) as exc: | |
| diarization._authorize(_request()) | |
| assert exc.value.status_code == 503 | |
| def test_authorize_rejects_invalid_key(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_API_KEY", "expected") | |
| with pytest.raises(HTTPException) as exc: | |
| diarization._authorize(_request({"Authorization": "Bearer wrong"})) | |
| assert exc.value.status_code == 401 | |
| def test_authorize_accepts_valid_key(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_API_KEY", "expected") | |
| diarization._authorize(_request({"Authorization": "Bearer expected"})) | |
| def test_max_input_bytes_rejects_invalid_configuration(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_MAX_INPUT_BYTES", "not-a-number") | |
| with pytest.raises(HTTPException) as exc: | |
| diarization._max_input_bytes() | |
| assert exc.value.status_code == 500 | |
| def test_max_input_bytes_rejects_too_small_configuration(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_MAX_INPUT_BYTES", "1024") | |
| with pytest.raises(HTTPException) as exc: | |
| diarization._max_input_bytes() | |
| assert exc.value.status_code == 500 | |
| def test_max_response_spans_rejects_invalid_configuration(monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_MAX_RESPONSE_SPANS", "0") | |
| with pytest.raises(HTTPException) as exc: | |
| diarization._max_response_spans() | |
| assert exc.value.status_code == 500 | |
| def test_save_body_rejects_declared_oversize(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_MAX_INPUT_BYTES", "1048576") | |
| class StreamRequest: | |
| headers = {"content-length": "1048577"} | |
| def stream(self): | |
| raise AssertionError("body should not be read after declared-size rejection") | |
| with pytest.raises(HTTPException) as exc: | |
| asyncio.run(diarization._save_body(StreamRequest(), tmp_path / "input")) | |
| assert exc.value.status_code == 413 | |
| def _normalize(raw: list[dict]) -> list[dict]: | |
| return diarization._normalize_spans(raw, duration=10.0, max_spans=10, max_speakers=32) | |
| def test_normalize_spans_trims_speaker_label() -> None: | |
| result = _normalize([{"speaker": " SPEAKER_00 ", "start": 1, "end": 2}]) | |
| assert result == [{"speaker": "SPEAKER_00", "start": 1.0, "end": 2.0}] | |
| def test_normalize_spans_rejects_invalid_output(span: dict) -> None: | |
| with pytest.raises(HTTPException) as exc: | |
| _normalize([span]) | |
| assert exc.value.status_code == 422 | |
| def test_normalize_spans_clamps_small_duration_rounding_error() -> None: | |
| result = _normalize([{"speaker": "A", "start": 9.9, "end": 10.05}]) | |
| assert result[0]["end"] == 10.0 | |
| def test_normalize_spans_allows_overlapping_different_speakers() -> None: | |
| result = _normalize([ | |
| {"speaker": "A", "start": 0, "end": 2}, | |
| {"speaker": "B", "start": 1, "end": 3}, | |
| ]) | |
| assert [item["speaker"] for item in result] == ["A", "B"] | |
| def test_normalize_spans_rejects_excess_speakers() -> None: | |
| with pytest.raises(HTTPException, match="too many speakers"): | |
| diarization._normalize_spans( | |
| [ | |
| {"speaker": "A", "start": 0, "end": 1}, | |
| {"speaker": "B", "start": 1, "end": 2}, | |
| ], | |
| duration=10.0, | |
| max_spans=10, | |
| max_speakers=1, | |
| ) | |
| def test_run_pipeline_forwards_speaker_bounds_and_normalizes(tmp_path: Path) -> None: | |
| captured: dict[str, object] = {} | |
| class FakeDiarization: | |
| def itertracks(self, yield_label=False): | |
| assert yield_label is True | |
| return iter([ | |
| (SimpleNamespace(start=1.0, end=2.0), None, " SPEAKER_00 "), | |
| (SimpleNamespace(start=0.0, end=1.0), None, "SPEAKER_01"), | |
| ]) | |
| class FakePipeline: | |
| def __call__(self, path, **kwargs): | |
| captured["path"] = path | |
| captured["kwargs"] = kwargs | |
| return FakeDiarization() | |
| result = diarization._run_pipeline( | |
| FakePipeline(), tmp_path / "audio.wav", min_speakers=2, max_speakers=3, | |
| duration=2.0, max_spans=10, max_speakers_allowed=32, | |
| ) | |
| assert captured["kwargs"] == {"min_speakers": 2, "max_speakers": 3} | |
| assert [item["speaker"] for item in result] == ["SPEAKER_01", "SPEAKER_00"] | |
| def test_run_pipeline_rejects_too_many_spans(tmp_path: Path) -> None: | |
| class FakeDiarization: | |
| def itertracks(self, yield_label=False): | |
| return iter([ | |
| (SimpleNamespace(start=0.0, end=0.5), None, "A"), | |
| (SimpleNamespace(start=0.5, end=1.0), None, "A"), | |
| ]) | |
| class FakePipeline: | |
| def __call__(self, path, **kwargs): | |
| return FakeDiarization() | |
| with pytest.raises(HTTPException, match="too many spans"): | |
| diarization._run_pipeline( | |
| FakePipeline(), tmp_path / "audio.wav", None, None, | |
| duration=2.0, max_spans=1, max_speakers_allowed=32, | |
| ) | |