wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""Tests for the voice (TTS/STT) service layer.
|
|
|
|
Covers Markdown cleaning, the OpenAI-compatible adapters' wire shape, the
|
|
OpenRouter base64-JSON STT branch, Azure auth headers, and catalog-driven
|
|
config resolution.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from deeptutor.services.config.provider_runtime import (
|
|
resolve_stt_runtime_config,
|
|
resolve_tts_runtime_config,
|
|
)
|
|
from deeptutor.services.voice import synthesize_speech, transcribe_audio
|
|
from deeptutor.services.voice.adapters.openai_compat import (
|
|
OpenAICompatSTTAdapter,
|
|
OpenAICompatTTSAdapter,
|
|
OpenRouterTTSAdapter,
|
|
)
|
|
from deeptutor.services.voice.base import (
|
|
build_auth_headers,
|
|
join_audio_path,
|
|
strip_markdown_for_speech,
|
|
)
|
|
from deeptutor.services.voice.config import STTConfig, TTSConfig
|
|
|
|
|
|
def _capture_post(monkeypatch: pytest.MonkeyPatch, response: httpx.Response) -> dict[str, Any]:
|
|
"""Patch ``httpx.AsyncClient.post`` to record args and return ``response``."""
|
|
captured: dict[str, Any] = {}
|
|
|
|
async def fake_post(self: httpx.AsyncClient, url: str, **kwargs: Any) -> httpx.Response:
|
|
captured["url"] = url
|
|
captured["json"] = kwargs.get("json")
|
|
captured["data"] = kwargs.get("data")
|
|
captured["files"] = kwargs.get("files")
|
|
captured["headers"] = kwargs.get("headers")
|
|
response.request = httpx.Request("POST", url)
|
|
return response
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
|
|
return captured
|
|
|
|
|
|
# ── text cleaning ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_strip_markdown_drops_code_and_unwraps_links() -> None:
|
|
md = "# Title\n\nHello **world**, read [the docs](http://x).\n\n```py\nprint(1)\n```\n- one\n- two"
|
|
out = strip_markdown_for_speech(md)
|
|
assert "Title" in out and "Hello world" in out and "the docs" in out
|
|
assert "print(1)" not in out # fenced code dropped
|
|
assert "**" not in out and "[" not in out and "#" not in out
|
|
|
|
|
|
def test_strip_markdown_truncates_on_boundary() -> None:
|
|
out = strip_markdown_for_speech("Sentence one. Sentence two. Sentence three.", max_chars=20)
|
|
assert len(out) <= 20
|
|
assert out.endswith(".")
|
|
|
|
|
|
def test_join_audio_path_appends_and_preserves_full_url() -> None:
|
|
assert join_audio_path("https://api.openai.com/v1", "audio/speech").endswith("/v1/audio/speech")
|
|
full = "https://r.azure.com/openai/deployments/tts/audio/speech?api-version=2025"
|
|
assert join_audio_path(full, "audio/speech") == full
|
|
|
|
|
|
def test_auth_headers_styles() -> None:
|
|
assert build_auth_headers("bearer", "k") == {"Authorization": "Bearer k"}
|
|
assert build_auth_headers("api_key_header", "k") == {"api-key": "k"}
|
|
assert build_auth_headers("token", "k") == {"Authorization": "Token k"}
|
|
assert build_auth_headers("bearer", "") == {}
|
|
|
|
|
|
# ── TTS adapter ───────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tts_adapter_posts_openai_shape(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, content=b"ID3audio-bytes", headers={"content-type": "audio/mpeg"})
|
|
captured = _capture_post(monkeypatch, resp)
|
|
config = TTSConfig(
|
|
model="gpt-4o-mini-tts",
|
|
base_url="https://api.openai.com/v1",
|
|
api_key="sk-test",
|
|
voice="alloy",
|
|
response_format="mp3",
|
|
)
|
|
audio, content_type = await OpenAICompatTTSAdapter().synthesize("hi there", config)
|
|
assert audio == b"ID3audio-bytes"
|
|
assert content_type == "audio/mpeg"
|
|
assert captured["url"] == "https://api.openai.com/v1/audio/speech"
|
|
assert captured["json"] == {
|
|
"model": "gpt-4o-mini-tts",
|
|
"input": "hi there",
|
|
"response_format": "mp3",
|
|
"voice": "alloy",
|
|
}
|
|
assert captured["headers"]["Authorization"] == "Bearer sk-test"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tts_adapter_azure_uses_api_key_header(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, content=b"x", headers={"content-type": "audio/mpeg"})
|
|
captured = _capture_post(monkeypatch, resp)
|
|
config = TTSConfig(
|
|
model="tts-1",
|
|
base_url="https://r.azure.com/openai/deployments/tts/audio/speech?api-version=2025-04-01",
|
|
api_key="azkey",
|
|
auth_style="api_key_header",
|
|
voice="alloy",
|
|
)
|
|
await OpenAICompatTTSAdapter().synthesize("hello", config)
|
|
assert captured["headers"]["api-key"] == "azkey"
|
|
assert "Authorization" not in captured["headers"]
|
|
# Full /audio/ URL is preserved verbatim.
|
|
assert captured["url"].endswith("api-version=2025-04-01")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tts_adapter_raises_on_http_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
from deeptutor.services.voice.base import VoiceProviderError
|
|
|
|
_capture_post(monkeypatch, httpx.Response(401, text="bad key"))
|
|
config = TTSConfig(model="m", base_url="https://x/v1", api_key="k", voice="alloy")
|
|
with pytest.raises(VoiceProviderError, match="401"):
|
|
await OpenAICompatTTSAdapter().synthesize("hi", config)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openrouter_tts_falls_back_to_chat_audio_stream(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
post_calls: list[dict[str, Any]] = []
|
|
|
|
async def fake_post(self: httpx.AsyncClient, url: str, **kwargs: Any) -> httpx.Response:
|
|
post_calls.append(
|
|
{
|
|
"url": url,
|
|
"json": kwargs.get("json"),
|
|
"headers": kwargs.get("headers"),
|
|
}
|
|
)
|
|
if len(post_calls) == 1:
|
|
response = httpx.Response(
|
|
500,
|
|
json={"error": {"message": "Internal Server Error"}},
|
|
)
|
|
else:
|
|
chunk = {
|
|
"choices": [
|
|
{
|
|
"delta": {
|
|
"audio": {
|
|
"data": base64.b64encode(b"pcm-audio").decode("ascii"),
|
|
"transcript": "hi",
|
|
}
|
|
}
|
|
}
|
|
]
|
|
}
|
|
response = httpx.Response(
|
|
200,
|
|
text=f"data: {json.dumps(chunk)}\n\ndata: [DONE]\n",
|
|
headers={"content-type": "text/event-stream"},
|
|
)
|
|
response.request = httpx.Request("POST", url)
|
|
return response
|
|
|
|
monkeypatch.setattr(httpx.AsyncClient, "post", fake_post)
|
|
|
|
config = TTSConfig(
|
|
model="openai/gpt-4o-mini-tts",
|
|
provider_name="openrouter",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
api_key="or-key",
|
|
voice="alloy",
|
|
response_format="pcm",
|
|
)
|
|
|
|
audio, content_type = await OpenRouterTTSAdapter().synthesize("hello", config)
|
|
|
|
assert audio == b"pcm-audio"
|
|
assert content_type == "audio/pcm"
|
|
assert post_calls[0]["url"] == "https://openrouter.ai/api/v1/audio/speech"
|
|
assert post_calls[1]["url"] == "https://openrouter.ai/api/v1/chat/completions"
|
|
assert post_calls[1]["json"]["modalities"] == ["text", "audio"]
|
|
assert post_calls[1]["json"]["audio"] == {"voice": "alloy", "format": "pcm16"}
|
|
assert post_calls[1]["headers"]["Authorization"] == "Bearer or-key"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_openrouter_gemini_tts_openai_voice_gets_clear_hint(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from deeptutor.services.voice.base import VoiceProviderError
|
|
|
|
_capture_post(
|
|
monkeypatch,
|
|
httpx.Response(500, json={"error": {"message": "Internal Server Error"}}),
|
|
)
|
|
config = TTSConfig(
|
|
model="google/gemini-3.1-flash-tts-preview",
|
|
provider_name="openrouter",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
api_key="or-key",
|
|
voice="alloy",
|
|
response_format="pcm",
|
|
)
|
|
with pytest.raises(VoiceProviderError, match="Kore"):
|
|
await OpenRouterTTSAdapter().synthesize("hello", config)
|
|
|
|
|
|
# ── STT adapter ───────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stt_adapter_multipart(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, json={"text": "hello world"})
|
|
captured = _capture_post(monkeypatch, resp)
|
|
config = STTConfig(model="whisper-1", base_url="https://api.openai.com/v1", api_key="sk")
|
|
text = await OpenAICompatSTTAdapter().transcribe(
|
|
b"RIFFxxxx", config, filename="a.wav", content_type="audio/wav"
|
|
)
|
|
assert text == "hello world"
|
|
assert captured["url"] == "https://api.openai.com/v1/audio/transcriptions"
|
|
assert captured["files"]["file"][0] == "a.wav"
|
|
assert captured["data"]["model"] == "whisper-1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stt_adapter_openrouter_base64(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, json={"text": "from base64"})
|
|
captured = _capture_post(monkeypatch, resp)
|
|
config = STTConfig(
|
|
model="openai/whisper-large-v3",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
api_key="sk",
|
|
request_style="base64_json",
|
|
)
|
|
text = await OpenAICompatSTTAdapter().transcribe(
|
|
b"audiobytes", config, filename="clip.webm", content_type="audio/webm"
|
|
)
|
|
assert text == "from base64"
|
|
assert captured["files"] is None # not multipart
|
|
assert captured["json"]["model"] == "openai/whisper-large-v3"
|
|
assert captured["json"]["input_audio"]["format"] == "webm"
|
|
assert captured["json"]["input_audio"]["data"] # base64 string present
|
|
|
|
|
|
# ── catalog resolution ────────────────────────────────────────────────────
|
|
|
|
|
|
def _voice_catalog() -> dict[str, Any]:
|
|
return {
|
|
"version": 1,
|
|
"services": {
|
|
"tts": {
|
|
"active_profile_id": "p1",
|
|
"active_model_id": "m1",
|
|
"profiles": [
|
|
{
|
|
"id": "p1",
|
|
"binding": "siliconflow",
|
|
"base_url": "",
|
|
"api_key": "sf-key",
|
|
"models": [
|
|
{
|
|
"id": "m1",
|
|
"model": "FunAudioLLM/CosyVoice2-0.5B",
|
|
"voice": "FunAudioLLM/CosyVoice2-0.5B:anna",
|
|
"response_format": "wav",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
},
|
|
"stt": {
|
|
"active_profile_id": "p2",
|
|
"active_model_id": "m2",
|
|
"profiles": [
|
|
{
|
|
"id": "p2",
|
|
"binding": "openrouter",
|
|
"base_url": "",
|
|
"api_key": "or-key",
|
|
"models": [{"id": "m2", "model": "openai/whisper-large-v3"}],
|
|
}
|
|
],
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def test_resolve_tts_config_uses_provider_default_base() -> None:
|
|
cfg = resolve_tts_runtime_config(catalog=_voice_catalog())
|
|
assert cfg.model == "FunAudioLLM/CosyVoice2-0.5B"
|
|
assert cfg.provider_name == "siliconflow"
|
|
assert cfg.base_url == "https://api.siliconflow.cn/v1" # filled from spec default
|
|
assert cfg.voice == "FunAudioLLM/CosyVoice2-0.5B:anna"
|
|
assert cfg.response_format == "wav"
|
|
assert cfg.api_key == "sf-key"
|
|
|
|
|
|
def test_resolve_stt_config_picks_openrouter_base64_style() -> None:
|
|
cfg = resolve_stt_runtime_config(catalog=_voice_catalog())
|
|
assert cfg.provider_name == "openrouter"
|
|
assert cfg.request_style == "base64_json"
|
|
assert cfg.base_url == "https://openrouter.ai/api/v1"
|
|
|
|
|
|
def test_resolve_tts_config_picks_openrouter_adapter() -> None:
|
|
catalog = _voice_catalog()
|
|
catalog["services"]["tts"]["profiles"][0]["binding"] = "openrouter"
|
|
catalog["services"]["tts"]["profiles"][0]["models"][0]["model"] = (
|
|
"google/gemini-3.1-flash-tts-preview"
|
|
)
|
|
cfg = resolve_tts_runtime_config(catalog=catalog)
|
|
assert cfg.provider_name == "openrouter"
|
|
assert cfg.adapter == "openrouter_tts"
|
|
|
|
|
|
def test_resolve_tts_config_raises_without_model() -> None:
|
|
catalog = {"version": 1, "services": {"tts": {"profiles": []}}}
|
|
with pytest.raises(ValueError, match="No active TTS model"):
|
|
resolve_tts_runtime_config(catalog=catalog)
|
|
|
|
|
|
# ── facade ────────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_synthesize_speech_facade_strips_markdown(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, content=b"audio", headers={"content-type": "audio/wav"})
|
|
captured = _capture_post(monkeypatch, resp)
|
|
audio, ctype = await synthesize_speech("# Hi\n\n**bold**", catalog=_voice_catalog())
|
|
assert audio == b"audio"
|
|
assert captured["json"]["input"] == "Hi\n\nbold" # markdown stripped
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transcribe_audio_facade(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
resp = httpx.Response(200, json={"text": "transcribed"})
|
|
_capture_post(monkeypatch, resp)
|
|
text = await transcribe_audio(b"bytes", catalog=_voice_catalog(), filename="x.webm")
|
|
assert text == "transcribed"
|