Spaces:
Running
Running
| from __future__ import annotations | |
| import asyncio | |
| from pathlib import Path | |
| import httpx | |
| import pytest | |
| from app.core.exceptions import InputError, ProcessingError | |
| from app.services.diarization_service import DiarizationService | |
| def _service() -> DiarizationService: | |
| return DiarizationService(None) # type: ignore[arg-type] | |
| def _path(tmp_path: Path) -> Path: | |
| path = tmp_path / "sample.wav" | |
| path.write_bytes(b"sample") | |
| return path | |
| def test_speaker_bounds_validation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_URL", "http://localhost:9000/diarize") | |
| with pytest.raises(InputError, match="between 1 and 32"): | |
| asyncio.run(_service().diarize(_path(tmp_path), "audio/wav", "sample.wav", min_speakers=0)) | |
| def test_speaker_bounds_order_validation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: | |
| monkeypatch.setenv("DIARIZATION_URL", "http://localhost:9000/diarize") | |
| with pytest.raises(InputError, match="cannot exceed"): | |
| asyncio.run( | |
| _service().diarize( | |
| _path(tmp_path), "audio/wav", "sample.wav", min_speakers=4, max_speakers=2 | |
| ) | |
| ) | |
| def test_adapter_forwards_auth_and_speaker_headers( | |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch | |
| ) -> None: | |
| monkeypatch.setenv("DIARIZATION_URL", "https://example.test/v1/diarization/diarize") | |
| monkeypatch.setenv("DIARIZATION_API_KEY", "secret") | |
| captured: dict = {} | |
| async def handler(request: httpx.Request) -> httpx.Response: | |
| captured["headers"] = request.headers | |
| captured["body"] = await request.aread() | |
| return httpx.Response( | |
| 200, | |
| json={ | |
| "spans": [ | |
| {"speaker": "SPEAKER_00", "start": 0.0, "end": 1.25}, | |
| {"speaker": "SPEAKER_01", "start": 1.25, "end": 2.5}, | |
| ] | |
| }, | |
| ) | |
| async def run() -> list[dict]: | |
| service = _service() | |
| transport = httpx.MockTransport(handler) | |
| original = httpx.AsyncClient | |
| class Client: | |
| def __init__(self, *args, **kwargs): | |
| self._client = original(transport=transport, *args, **kwargs) | |
| async def __aenter__(self): | |
| return await self._client.__aenter__() | |
| async def __aexit__(self, *args): | |
| return await self._client.__aexit__(*args) | |
| monkeypatch.setattr(httpx, "AsyncClient", Client) | |
| return await service.diarize( | |
| _path(tmp_path), "audio/wav", "sample.wav", min_speakers=2, max_speakers=4 | |
| ) | |
| result = asyncio.run(run()) | |
| assert result[0]["speaker"] == "SPEAKER_00" | |
| assert captured["headers"]["authorization"] == "Bearer secret" | |
| assert captured["headers"]["x-min-speakers"] == "2" | |
| assert captured["headers"]["x-max-speakers"] == "4" | |
| assert captured["headers"]["x-filename"] == "sample.wav" | |
| assert captured["body"] == b"sample" | |
| def test_adapter_rejects_invalid_span( | |
| tmp_path: Path, monkeypatch: pytest.MonkeyPatch | |
| ) -> None: | |
| monkeypatch.setenv("DIARIZATION_URL", "https://example.test/v1/diarization/diarize") | |
| monkeypatch.setenv("DIARIZATION_API_KEY", "secret") | |
| async def handler(request: httpx.Request) -> httpx.Response: | |
| return httpx.Response(200, json={"spans": [{"speaker": "A", "start": 2, "end": 2}]}) | |
| async def run() -> None: | |
| transport = httpx.MockTransport(handler) | |
| original = httpx.AsyncClient | |
| class Client: | |
| def __init__(self, *args, **kwargs): | |
| self._client = original(transport=transport, *args, **kwargs) | |
| async def __aenter__(self): | |
| return await self._client.__aenter__() | |
| async def __aexit__(self, *args): | |
| return await self._client.__aexit__(*args) | |
| monkeypatch.setattr(httpx, "AsyncClient", Client) | |
| await _service().diarize(_path(tmp_path), "audio/wav", "sample.wav") | |
| with pytest.raises(ProcessingError, match="invalid span"): | |
| asyncio.run(run()) | |