Spaces:
Running
Running
File size: 4,080 Bytes
64a7b73 68376b1 d772075 64a7b73 d772075 64a7b73 68376b1 64a7b73 d772075 64a7b73 68376b1 64a7b73 68376b1 d772075 68376b1 d772075 | 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 | 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())
|