Add diarization API security and limit tests

tests/test_diarization_api.py

@@ -0,0 +1,95 @@
+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 test_run_pipeline_forwards_speaker_bounds(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=0.0, end=1.0), None, "SPEAKER_00"),
+                (SimpleNamespace(start=1.0, end=2.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
+    )
+    assert captured["kwargs"] == {"min_speakers": 2, "max_speakers": 3}
+    assert [item["speaker"] for item in result] == ["SPEAKER_00", "SPEAKER_01"]