148 lines
3.1 KiB
Python
148 lines
3.1 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from voice_assistant.tts.voxcpm import VoxCPMTTSAdapter
|
|
|
|
|
|
class MockHTTPStream:
|
|
def __init__(
|
|
self,
|
|
chunks: list[bytes],
|
|
status_code: int = 200,
|
|
) -> None:
|
|
self.chunks = chunks
|
|
self.headers = {"content-type": "audio/pcm"}
|
|
self.status_code = status_code
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise httpx.HTTPStatusError(
|
|
"mock error",
|
|
request=httpx.Request("POST", "http://test"),
|
|
response=httpx.Response(
|
|
status_code=self.status_code,
|
|
),
|
|
)
|
|
|
|
async def aiter_bytes(self):
|
|
for chunk in self.chunks:
|
|
yield chunk
|
|
|
|
|
|
class MockHTTPClient:
|
|
last_request = None
|
|
|
|
def __init__(self, *args, **kwargs) -> None:
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def stream(self, method, url, json):
|
|
self.last_request = {
|
|
"method": method,
|
|
"url": url,
|
|
"json": json,
|
|
}
|
|
|
|
MockHTTPClient.last_request = self.last_request
|
|
|
|
return MockHTTPStream(
|
|
[
|
|
b"\x01\x02\x03\x04",
|
|
b"\x05\x06\x07\x08",
|
|
]
|
|
)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voxcpm_tts_streams_audio(monkeypatch):
|
|
monkeypatch.setattr(
|
|
httpx,
|
|
"AsyncClient",
|
|
MockHTTPClient,
|
|
)
|
|
|
|
tts = VoxCPMTTSAdapter(
|
|
server="http://127.0.0.1:8000",
|
|
model="openbmb/VoxCPM2",
|
|
voice="default",
|
|
)
|
|
|
|
chunks = [
|
|
chunk
|
|
async for chunk in tts.synthesize(
|
|
"Olá Alex, tudo bem?"
|
|
)
|
|
]
|
|
|
|
assert chunks == [
|
|
b"\x01\x02\x03\x04",
|
|
b"\x05\x06\x07\x08",
|
|
]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voxcpm_tts_sends_expected_payload(monkeypatch):
|
|
monkeypatch.setattr(
|
|
httpx,
|
|
"AsyncClient",
|
|
MockHTTPClient,
|
|
)
|
|
|
|
tts = VoxCPMTTSAdapter(
|
|
server="http://127.0.0.1:8000",
|
|
model="openbmb/VoxCPM2",
|
|
voice="default",
|
|
)
|
|
|
|
chunks = [
|
|
chunk
|
|
async for chunk in tts.synthesize("Olá")
|
|
]
|
|
|
|
assert chunks
|
|
|
|
request = MockHTTPClient.last_request
|
|
|
|
assert request is not None
|
|
|
|
assert request["method"] == "POST"
|
|
assert request["url"] == (
|
|
"http://127.0.0.1:8000/v1/audio/speech"
|
|
)
|
|
assert request["json"] == {
|
|
"input": "Olá",
|
|
"model": "openbmb/VoxCPM2",
|
|
"voice": "default",
|
|
"response_format": "pcm",
|
|
"stream": True,
|
|
"stream_format": "audio",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_voxcpm_tts_rejects_empty_text():
|
|
tts = VoxCPMTTSAdapter(
|
|
server="http://127.0.0.1:8000",
|
|
model="openbmb/VoxCPM2",
|
|
voice="default",
|
|
)
|
|
|
|
with pytest.raises(
|
|
ValueError,
|
|
match="Text cannot be empty",
|
|
):
|
|
chunks = [
|
|
chunk
|
|
async for chunk in tts.synthesize("")
|
|
] |