Spaces:
Running
Running
File size: 6,198 Bytes
1e00a68 9a7e2d7 72f6b6a 9a7e2d7 1e00a68 9a7e2d7 1e00a68 9a7e2d7 1e00a68 9a7e2d7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | 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}]
@pytest.mark.parametrize(
"span",
[
{"speaker": "A", "start": -1, "end": 1},
{"speaker": "A", "start": 2, "end": 2},
{"speaker": "A", "start": 3, "end": 2},
{"speaker": "A", "start": float("nan"), "end": 2},
{"speaker": "A", "start": 1, "end": float("inf")},
{"speaker": "A", "start": 1, "end": 11},
{"speaker": "", "start": 1, "end": 2},
{"speaker": None, "start": 1, "end": 2},
],
)
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,
)
|