The remote had one commit this branch did not (03f8a85), which adds the
Apache-2.0 LICENSE, the NOTICE and SPDX headers to every .py and .sh
file. The merge was clean - the headers sit at the top of each file and
my changes are further down.
The new files this branch added (languages.py, translations.py, the
cluster package, the new tests and scripts) were written after the
merge base, so the licence commit never saw them. They now carry the
same header, so the convention holds across the whole tree.
Nothing else changed. 883 tests passed / 8 failed - the same 8
pre-existing failures.
156 lines
4.6 KiB
Python
156 lines
4.6 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Per-utterance language in the TTS call (spec 31 section 4.3 / section 9).
|
|
|
|
Two guarantees are pinned here:
|
|
|
|
1. **Refuse, never wrong.** A language Qwen3-TTS cannot speak is rejected before
|
|
any network call or filesystem write, so an unsupported tag leaves a visible
|
|
refusal instead of a silent English render.
|
|
2. **Byte-identical when varied.** With no language passed, the payload does not
|
|
change, so existing renders are untouched.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import wave
|
|
|
|
import pytest
|
|
|
|
from umbraaudio import languages as lang
|
|
from umbraaudio.voices.qwen_tts import QwenTTSEngine
|
|
|
|
|
|
def test_gap_language_refused_before_touching_the_network(tmp_path) -> None:
|
|
engine = QwenTTSEngine()
|
|
out = tmp_path / "utt_0001.wav"
|
|
with pytest.raises(ValueError) as exc:
|
|
engine.generate("Hallo wereld.", out, language="nl")
|
|
assert "Dutch" in str(exc.value)
|
|
assert not out.exists(), "refusal must happen before any output is written"
|
|
|
|
|
|
def test_unrecognised_language_is_refused(tmp_path) -> None:
|
|
engine = QwenTTSEngine()
|
|
out = tmp_path / "utt_0002.wav"
|
|
with pytest.raises(ValueError) as exc:
|
|
engine.generate("Whatever.", out, language="klingon")
|
|
assert "unrecognised" in str(exc.value)
|
|
assert not out.exists()
|
|
|
|
|
|
def test_voiceable_languages_pass_the_refusal_guard() -> None:
|
|
for code in ("en", "zh", "ja", "ko", "de", "es", "fr", "ru", "it", "pt"):
|
|
assert lang.refusal_reason(code) is None
|
|
|
|
|
|
def test_resolved_name_is_what_the_model_needs() -> None:
|
|
"""A code must become the model's own string, not the code itself."""
|
|
assert lang.tts_name("es") == "spanish"
|
|
assert lang.tts_name("de") == "german"
|
|
assert lang.tts_name("zh") == "chinese"
|
|
|
|
|
|
def _tiny_wav() -> bytes:
|
|
buf = io.BytesIO()
|
|
with wave.open(buf, "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(24000)
|
|
w.writeframes(b"\x00\x00" * 240)
|
|
return buf.getvalue()
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, content: bytes) -> None:
|
|
self.status_code = 200
|
|
self.content = content
|
|
self.headers = {"content-type": "audio/wav"}
|
|
self.text = ""
|
|
|
|
|
|
def test_remote_payload_sends_language_when_varied(tmp_path) -> None:
|
|
engine = QwenTTSEngine()
|
|
captured: dict = {}
|
|
|
|
class _Resp(_FakeResponse):
|
|
def __init__(self) -> None:
|
|
super().__init__(_tiny_wav())
|
|
|
|
import httpx
|
|
|
|
def fake_post(url, data=None, files=None, timeout=None): # noqa: ANN001
|
|
captured["url"] = url
|
|
captured["data"] = data
|
|
return _Resp()
|
|
|
|
original = httpx.post
|
|
httpx.post = fake_post # type: ignore[assignment]
|
|
try:
|
|
engine._remote_generate_at(
|
|
url="http://127.0.0.1:9/v1",
|
|
text="Hola, amigo.",
|
|
files=None,
|
|
instruct=None,
|
|
ref_text=None,
|
|
prompt_slug=None,
|
|
temperature=0.9,
|
|
do_sample=True,
|
|
seed=None,
|
|
language="spanish",
|
|
)
|
|
finally:
|
|
httpx.post = original # type: ignore[assignment]
|
|
|
|
assert captured["data"]["language"] == "spanish"
|
|
|
|
|
|
def test_remote_payload_omits_language_when_not_given(tmp_path) -> None:
|
|
"""No language passed -> payload unchanged, so existing renders are identical."""
|
|
engine = QwenTTSEngine()
|
|
captured: dict = {}
|
|
|
|
import httpx
|
|
|
|
class _Resp(_FakeResponse):
|
|
def __init__(self) -> None:
|
|
super().__init__(_tiny_wav())
|
|
|
|
def fake_post(url, data=None, files=None, timeout=None): # noqa: ANN001
|
|
captured["data"] = data
|
|
return _Resp()
|
|
|
|
original = httpx.post
|
|
httpx.post = fake_post # type: ignore[assignment]
|
|
try:
|
|
engine._remote_generate_at(
|
|
url="http://127.0.0.1:9/v1",
|
|
text="Plain English line.",
|
|
files=None,
|
|
instruct=None,
|
|
ref_text=None,
|
|
prompt_slug=None,
|
|
temperature=0.9,
|
|
do_sample=True,
|
|
seed=None,
|
|
language=None,
|
|
)
|
|
finally:
|
|
httpx.post = original # type: ignore[assignment]
|
|
|
|
assert "language" not in captured["data"]
|
|
|
|
|
|
def test_local_clone_defaults_to_english_when_not_given() -> None:
|
|
"""The local paths keep their historical default rather than becoming None."""
|
|
import inspect
|
|
|
|
from umbraaudio.voices import qwen_tts as mod
|
|
|
|
clone = inspect.signature(mod.QwenTTSEngine._generate_clone)
|
|
vd = inspect.signature(mod.QwenTTSEngine._generate_voicedesign)
|
|
assert clone.parameters["language"].default == "english"
|
|
assert vd.parameters["language"].default == "english"
|