820872e117
Port the original #27 detected-language pipeline onto the verified PR #39 prerequisite while preserving the current-main conversation instrument and host continuity changes. Keep voice selection server-side with no user selector or client voice field. Reuse 207c16ab only for its stricter exact-code trust boundary, omitting malformed or absent language so Piper defaults to Amy.
678 lines
23 KiB
Python
678 lines
23 KiB
Python
"""Voice-mode language routing: private Whisper STT decides the Piper voice.
|
||
|
||
Every assertion here runs without a GPU, a microphone or a cluster. The browser
|
||
contract is exercised by driving the real ``atlas-voice.js`` inside a stub DOM
|
||
(``testing/tests/data/atlas_voice_language_probe.js``), and the two server-side
|
||
trust boundaries are exercised by applying the real image patch to fixtures that
|
||
carry the exact upstream anchors and then importing the patched result.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from testing.tests.test_hermes_chat_support import HERMES, ROOT
|
||
|
||
DOCKERFILES = ROOT / "dockerfiles"
|
||
ATLAS_PATCH = DOCKERFILES / "hermes-webui-atlas-patch.py"
|
||
VOICE_SCRIPT = DOCKERFILES / "hermes-webui-atlas-voice.js"
|
||
VOICE_PROBE = ROOT / "testing" / "tests" / "data" / "atlas_voice_language_probe.js"
|
||
WEBUI_FIXTURE = ROOT / "testing" / "fixtures" / "hermes-webui-0.52.181"
|
||
ATLAS_TTS_URL = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech"
|
||
|
||
# Voices baked by the multilingual Piper work (PR #26): en=amy, ru=irina,
|
||
# es=claude. Anything outside this set must resolve to English.
|
||
SUPPORTED = ("en", "ru", "es")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Module loaders
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _load_stt_server(monkeypatch):
|
||
"""Import the Jetson Whisper service without CUDA, torch or whisper."""
|
||
path = DOCKERFILES / "hermes-jetson-stt-server.py"
|
||
spec = importlib.util.spec_from_file_location("hermes_jetson_stt_server", path)
|
||
assert spec and spec.loader
|
||
module = importlib.util.module_from_spec(spec)
|
||
monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace())
|
||
monkeypatch.setitem(
|
||
sys.modules,
|
||
"torch",
|
||
SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False)),
|
||
)
|
||
monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace())
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
def _load_stt_client():
|
||
"""Import the local-command STT client that Hermes shells out to."""
|
||
path = HERMES / "scripts" / "hermes_stt_client.py"
|
||
spec = importlib.util.spec_from_file_location("hermes_stt_client", path)
|
||
assert spec and spec.loader
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Patch fixtures — each file carries the exact upstream fragment the image
|
||
# patch pins, so importing the patched result exercises the inserted code.
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
INDEX_FIXTURE = (
|
||
'<select id="settingsTtsEngine">'
|
||
'<option value="browser">Browser speech synthesis</option>'
|
||
'<option value="edge">Edge TTS (server)</option></select>\n'
|
||
'<script src="static/boot.js?v=__WEBUI_VERSION__" defer></script>\n'
|
||
)
|
||
|
||
UI_FIXTURE = """function _playEdgeTtsChunked(text, btn){
|
||
fetch('/api/tts',{method:'POST',body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch})});
|
||
}
|
||
function readAloud(clean, btn, engine){
|
||
if(engine==='edge'){
|
||
_playEdgeTtsChunked(clean, btn);
|
||
}
|
||
}
|
||
function autoRead(clean, engine){
|
||
if(engine==='edge'){
|
||
_playEdgeTtsChunked(clean, null);
|
||
}
|
||
}
|
||
"""
|
||
|
||
HELPERS_FIXTURE = '''"""Stand-in for the WebUI helper module the patched code imports."""
|
||
|
||
|
||
def bad(handler, message, status=400):
|
||
return {"status": status, "error": message}
|
||
'''
|
||
|
||
ROUTES_FIXTURE = '''"""Stand-in carrying the exact upstream anchors the Atlas TTS patch pins."""
|
||
|
||
import json
|
||
import os
|
||
from urllib.request import ProxyHandler, Request, build_opener
|
||
|
||
|
||
class _NoRedirectTtsHandler:
|
||
"""Placeholder for the upstream no-redirect opener handler."""
|
||
|
||
|
||
class _Logger:
|
||
def __init__(self):
|
||
self.failures = []
|
||
|
||
def exception(self, message):
|
||
self.failures.append(message)
|
||
|
||
|
||
logger = _Logger()
|
||
UPSTREAM_REQUESTS = []
|
||
|
||
|
||
class _Upstream:
|
||
def __init__(self, payload):
|
||
self._payload = payload
|
||
|
||
def read(self):
|
||
return self._payload
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *exc_info):
|
||
return False
|
||
|
||
|
||
def _buffer_tts_audio_response(response):
|
||
return response.read()
|
||
|
||
|
||
def _tts_open(req, *, timeout=30, opener_factory=None):
|
||
"""Thin network seam for the TTS upstream fetch so tests can intercept it."""
|
||
UPSTREAM_REQUESTS.append(json.loads(req.data.decode("utf-8")))
|
||
return _Upstream(b"RIFFsynthetic")
|
||
|
||
|
||
def _handle_tts(handler, data, text, rate_str, engine):
|
||
# ── ElevenLabs TTS ──────────────────────────────────────────────────
|
||
return None
|
||
'''
|
||
|
||
UPLOAD_FIXTURE = '''"""Stand-in carrying the exact upstream /api/transcribe response anchor."""
|
||
|
||
|
||
def j(handler, payload, status=200):
|
||
return {"status": status, "payload": payload}
|
||
|
||
|
||
def handle_transcribe(handler, result):
|
||
try:
|
||
transcript = str(result.get('transcript') or '').strip()
|
||
return j(handler, {'ok': True, 'transcript': transcript})
|
||
except ValueError as error:
|
||
return j(handler, {'error': str(error)}, status=400)
|
||
'''
|
||
|
||
TRANSCRIPTION_FIXTURE = '''"""Stand-in carrying the exact upstream local-command STT envelope anchor."""
|
||
|
||
import contextlib
|
||
from pathlib import Path
|
||
|
||
|
||
class _Logger:
|
||
def info(self, *args):
|
||
return None
|
||
|
||
|
||
logger = _Logger()
|
||
|
||
|
||
def _transcribe_local_command(file_path, normalized_model, output_dir):
|
||
try:
|
||
with contextlib.nullcontext(output_dir):
|
||
txt_files = sorted(Path(output_dir).glob("*.txt"))
|
||
transcript_text = txt_files[0].read_text(encoding="utf-8").strip()
|
||
logger.info(
|
||
"Transcribed %s via local STT command (%s, %d chars)",
|
||
Path(file_path).name,
|
||
normalized_model,
|
||
len(transcript_text),
|
||
)
|
||
return {"success": True, "transcript": transcript_text, "provider": "local_command"}
|
||
except OSError as error:
|
||
return {"success": False, "transcript": "", "error": str(error)}
|
||
'''
|
||
|
||
|
||
def _write(path: Path, text: str) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
|
||
@pytest.fixture
|
||
def patched_webui(tmp_path, monkeypatch):
|
||
"""Apply the real Atlas image patch to fixture trees and import the result."""
|
||
webui = tmp_path / "hermes-webui"
|
||
agent = tmp_path / "hermes"
|
||
# Start with the pinned full-surface fixture introduced by PR #39 so this
|
||
# test proves the language pipeline composes with its voice-selector removal
|
||
# and conversation instrument, not merely with the older #27 anchors.
|
||
shutil.copytree(WEBUI_FIXTURE, webui)
|
||
_write(webui / "api" / "__init__.py", "")
|
||
_write(webui / "api" / "helpers.py", HELPERS_FIXTURE)
|
||
_write(webui / "api" / "routes.py", ROUTES_FIXTURE)
|
||
_write(webui / "api" / "upload.py", UPLOAD_FIXTURE)
|
||
_write(agent / "tools" / "__init__.py", "")
|
||
_write(agent / "tools" / "transcription_tools.py", TRANSCRIPTION_FIXTURE)
|
||
|
||
environment = dict(os.environ)
|
||
environment["HERMES_WEBUI_PATCH_ROOT"] = str(webui)
|
||
environment["HERMES_AGENT_PATCH_ROOT"] = str(agent)
|
||
completed = subprocess.run(
|
||
[sys.executable, str(ATLAS_PATCH)],
|
||
env=environment,
|
||
capture_output=True,
|
||
text=True,
|
||
)
|
||
assert completed.returncode == 0, completed.stderr or completed.stdout
|
||
|
||
for name in ("api", "api.helpers", "api.routes", "api.upload", "tools",
|
||
"tools.transcription_tools"):
|
||
sys.modules.pop(name, None)
|
||
monkeypatch.syspath_prepend(str(agent))
|
||
monkeypatch.syspath_prepend(str(webui))
|
||
import api.routes as routes # noqa: PLC0415
|
||
import api.upload as upload # noqa: PLC0415
|
||
import tools.transcription_tools as transcription # noqa: PLC0415
|
||
|
||
yield SimpleNamespace(
|
||
webui=webui,
|
||
agent=agent,
|
||
routes=routes,
|
||
upload=upload,
|
||
transcription=transcription,
|
||
)
|
||
|
||
for name in ("api", "api.helpers", "api.routes", "api.upload", "tools",
|
||
"tools.transcription_tools"):
|
||
sys.modules.pop(name, None)
|
||
|
||
|
||
class _Handler:
|
||
"""Just enough BaseHTTPRequestHandler surface for the Atlas TTS branch."""
|
||
|
||
def __init__(self):
|
||
self.status = None
|
||
self.headers_sent = {}
|
||
self.wfile = SimpleNamespace(write=self._write)
|
||
self.body = b""
|
||
|
||
def send_response(self, status):
|
||
self.status = status
|
||
|
||
def send_header(self, name, value):
|
||
self.headers_sent[name] = value
|
||
|
||
def end_headers(self):
|
||
return None
|
||
|
||
def _write(self, payload):
|
||
self.body += payload
|
||
|
||
|
||
def _atlas_tts(patched, monkeypatch, data):
|
||
"""Run the patched Atlas branch and return the JSON it sent to hermes-tts."""
|
||
monkeypatch.setenv("HERMES_WEBUI_ATLAS_TTS_URL", ATLAS_TTS_URL)
|
||
patched.routes.UPSTREAM_REQUESTS.clear()
|
||
handler = _Handler()
|
||
result = patched.routes._handle_tts(handler, data, "Some reply.", "", "atlas")
|
||
assert result is True, "the Atlas branch must own the response"
|
||
assert handler.status == 200
|
||
assert len(patched.routes.UPSTREAM_REQUESTS) == 1
|
||
return patched.routes.UPSTREAM_REQUESTS[0]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1. Whisper service reports the language it actually decoded with
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_stt_response_carries_whisper_detected_language(monkeypatch):
|
||
module = _load_stt_server(monkeypatch)
|
||
payload = module._transcription_payload(
|
||
{
|
||
"language": "ru",
|
||
"segments": [
|
||
{"text": " Как дела?", "no_speech_prob": 0.1, "avg_logprob": -0.2}
|
||
],
|
||
}
|
||
)
|
||
assert payload["text"] == "Как дела?"
|
||
assert payload["language"] == "ru"
|
||
assert payload["model"] == module.MODEL_NAME
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("raw", "expected"),
|
||
[
|
||
("en", "en"),
|
||
("RU", "ru"),
|
||
(" es ", "es"),
|
||
("yue", "yue"),
|
||
("fr", "fr"),
|
||
("en-US", ""),
|
||
("en_US", ""),
|
||
("e", ""),
|
||
("english", ""),
|
||
("", ""),
|
||
("../en", ""),
|
||
("en\x00", ""),
|
||
("ru; rm -rf /", ""),
|
||
("рус", ""),
|
||
(None, ""),
|
||
(7, ""),
|
||
(["ru"], ""),
|
||
({"language": "ru"}, ""),
|
||
],
|
||
)
|
||
def test_stt_language_field_is_shape_validated(monkeypatch, raw, expected):
|
||
module = _load_stt_server(monkeypatch)
|
||
assert module._detected_language({"language": raw}) == expected
|
||
|
||
|
||
def test_stt_language_absent_when_whisper_omits_it(monkeypatch):
|
||
module = _load_stt_server(monkeypatch)
|
||
assert module._detected_language({}) == ""
|
||
assert module._detected_language("not a result") == ""
|
||
assert module._transcription_payload({"text": "hi"})["language"] == ""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. The local-command client carries the language without breaking the
|
||
# .txt contract Hermes reads the transcript from
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_stt_client_writes_language_sidecar_beside_the_txt_contract(tmp_path):
|
||
module = _load_stt_client()
|
||
module._write_result(tmp_path, "voice-input", "Как дела?", "ru")
|
||
assert (tmp_path / "voice-input.txt").read_text(encoding="utf-8") == "Как дела?"
|
||
assert (tmp_path / "voice-input.language").read_text(encoding="utf-8") == "ru"
|
||
# Hermes globs *.txt and reads the first match: the sidecar must not join it.
|
||
assert sorted(p.name for p in tmp_path.glob("*.txt")) == ["voice-input.txt"]
|
||
|
||
|
||
def test_stt_client_omits_the_sidecar_when_no_language_was_detected(tmp_path):
|
||
module = _load_stt_client()
|
||
module._write_result(tmp_path, "voice-input", "Hello.", "")
|
||
assert (tmp_path / "voice-input.txt").exists()
|
||
assert not (tmp_path / "voice-input.language").exists()
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("raw", "expected"),
|
||
[
|
||
("en", "en"),
|
||
("ES", "es"),
|
||
(" ru ", "ru"),
|
||
("en-US", ""),
|
||
("", ""),
|
||
("../../etc/passwd", ""),
|
||
("en\n", "en"),
|
||
("e", ""),
|
||
(None, ""),
|
||
(12, ""),
|
||
(["en"], ""),
|
||
],
|
||
)
|
||
def test_stt_client_normalises_the_service_language_field(raw, expected):
|
||
module = _load_stt_client()
|
||
assert module._normalize_language(raw) == expected
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. The patched agent envelope and /api/transcribe response carry it through
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_patched_local_command_envelope_carries_the_sidecar_language(patched_webui, tmp_path):
|
||
output = tmp_path / "stt-out"
|
||
output.mkdir()
|
||
(output / "voice-input.txt").write_text("Как дела?", encoding="utf-8")
|
||
(output / "voice-input.language").write_text("ru\n", encoding="utf-8")
|
||
result = patched_webui.transcription._transcribe_local_command(
|
||
"/tmp/voice-input.wav", "small", output
|
||
)
|
||
assert result == {
|
||
"success": True,
|
||
"transcript": "Как дела?",
|
||
"provider": "local_command",
|
||
"language": "ru",
|
||
}
|
||
|
||
|
||
def test_patched_local_command_envelope_defaults_to_no_language(patched_webui, tmp_path):
|
||
output = tmp_path / "stt-out"
|
||
output.mkdir()
|
||
(output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
|
||
result = patched_webui.transcription._transcribe_local_command(
|
||
"/tmp/voice-input.wav", "small", output
|
||
)
|
||
assert result["transcript"] == "Hello."
|
||
assert result["language"] == ""
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"hostile",
|
||
["en-US", "../../en", "en; rm -rf /", "e", "english", "", "\x00en", "e n"],
|
||
)
|
||
def test_patched_local_command_envelope_rejects_malformed_sidecars(
|
||
patched_webui, tmp_path, hostile
|
||
):
|
||
output = tmp_path / "stt-out"
|
||
output.mkdir()
|
||
(output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
|
||
(output / "voice-input.language").write_text(hostile, encoding="utf-8")
|
||
result = patched_webui.transcription._transcribe_local_command(
|
||
"/tmp/voice-input.wav", "small", output
|
||
)
|
||
assert result["language"] == ""
|
||
|
||
|
||
def test_patched_local_command_envelope_survives_an_undecodable_sidecar(
|
||
patched_webui, tmp_path
|
||
):
|
||
"""A corrupt sidecar must cost the language, never the transcript."""
|
||
output = tmp_path / "stt-out"
|
||
output.mkdir()
|
||
(output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
|
||
(output / "voice-input.language").write_bytes(b"\xff\xfe\x00ru")
|
||
result = patched_webui.transcription._transcribe_local_command(
|
||
"/tmp/voice-input.wav", "small", output
|
||
)
|
||
assert result["success"] is True
|
||
assert result["transcript"] == "Hello."
|
||
assert result["language"] == ""
|
||
|
||
|
||
def test_patched_transcribe_response_reports_the_language(patched_webui):
|
||
response = patched_webui.upload.handle_transcribe(
|
||
None, {"success": True, "transcript": " Как дела? ", "language": "ru"}
|
||
)
|
||
assert response["payload"] == {"ok": True, "transcript": "Как дела?", "language": "ru"}
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"hostile",
|
||
["", None, "en-US", "englishhh", "../en", 5, ["ru"], {"a": "b"}, "e"],
|
||
)
|
||
def test_patched_transcribe_response_blanks_untrusted_languages(patched_webui, hostile):
|
||
response = patched_webui.upload.handle_transcribe(
|
||
None, {"success": True, "transcript": "Hello.", "language": hostile}
|
||
)
|
||
assert response["payload"]["language"] == ""
|
||
assert response["payload"]["transcript"] == "Hello."
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. The /api/tts trust boundary: allow-list only, and never `voice`
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize("language", SUPPORTED)
|
||
def test_atlas_tts_forwards_allow_listed_languages(patched_webui, monkeypatch, language):
|
||
body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas", "language": language})
|
||
assert body["language"] == language
|
||
assert body["model"] == "piper"
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"hostile",
|
||
[
|
||
"fr",
|
||
"de",
|
||
"",
|
||
None,
|
||
"EN-GB",
|
||
" RU ",
|
||
"ru-RU",
|
||
"es_MX",
|
||
"../../ru_RU-irina-medium",
|
||
"ru; rm -rf /",
|
||
"ru\x00",
|
||
"ру",
|
||
5,
|
||
["ru"],
|
||
{"language": "ru"},
|
||
True,
|
||
"x" * 8192,
|
||
],
|
||
)
|
||
def test_atlas_tts_omits_untrusted_languages(
|
||
patched_webui, monkeypatch, hostile
|
||
):
|
||
body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas", "language": hostile})
|
||
assert "language" not in body
|
||
|
||
|
||
def test_atlas_tts_language_is_absent_when_the_client_sends_none(patched_webui, monkeypatch):
|
||
body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas"})
|
||
assert "language" not in body
|
||
|
||
|
||
def test_atlas_tts_voice_field_cannot_steer_synthesis(patched_webui, monkeypatch):
|
||
body = _atlas_tts(
|
||
patched_webui,
|
||
monkeypatch,
|
||
{"engine": "atlas", "voice": "ru_RU-irina-medium", "language": "en"},
|
||
)
|
||
assert body["language"] == "en"
|
||
body = _atlas_tts(
|
||
patched_webui,
|
||
monkeypatch,
|
||
{"engine": "atlas", "voice": "es_MX-claude-high"},
|
||
)
|
||
assert "language" not in body
|
||
|
||
|
||
def test_atlas_tts_language_helper_only_ever_returns_a_baked_voice_language(patched_webui):
|
||
resolve = patched_webui.routes._atlas_tts_language
|
||
hostile = [
|
||
None, 0, 1, -1, True, False, [], {}, set(), object(), b"ru",
|
||
"", " ", "\t\n", "en", "EN", "en-US", "en_us", "ru-RU", "es-MX",
|
||
"e", "eng", "english", "ru ru", "ru;es", "../ru", "ru\x00", "ру",
|
||
"x" * 65536, "en" * 4096,
|
||
]
|
||
for value in hostile:
|
||
expected = value if value in SUPPORTED else ""
|
||
assert resolve({"language": value}) == expected
|
||
for value in hostile:
|
||
assert resolve(value) == ""
|
||
assert resolve({"voice": "ru_RU-irina-medium"}) == ""
|
||
|
||
|
||
def test_manual_tts_button_body_still_carries_no_language(patched_webui):
|
||
"""The read-aloud button has no trusted STT signal, so it must stay Amy."""
|
||
ui = (patched_webui.webui / "static" / "ui.js").read_text(encoding="utf-8")
|
||
assert "engine:engineOverride||'edge'" in ui
|
||
assert "language" not in ui
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. Browser contract, driven through the real atlas-voice.js
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def voice_probe():
|
||
node = shutil.which("node")
|
||
if not node:
|
||
pytest.skip("node is required to drive the browser voice-mode contract")
|
||
completed = subprocess.run(
|
||
[node, str(VOICE_PROBE), str(VOICE_SCRIPT)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=180,
|
||
)
|
||
assert completed.returncode == 0, completed.stderr
|
||
return json.loads(completed.stdout)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("scenario", "expected"),
|
||
[
|
||
("english_turn_speaks_english", "en"),
|
||
("russian_turn_speaks_russian", "ru"),
|
||
("spanish_turn_speaks_spanish", "es"),
|
||
],
|
||
)
|
||
def test_voice_mode_speaks_the_language_whisper_detected(voice_probe, scenario, expected):
|
||
requests = voice_probe[scenario]["tts"]
|
||
assert requests, "voice mode never reached /api/tts"
|
||
for request in requests:
|
||
assert request["engine"] == "atlas"
|
||
assert request["language"] == expected
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"scenario", ["missing_language_falls_back", "unsupported_language_falls_back"]
|
||
)
|
||
def test_voice_mode_omits_language_without_a_trusted_signal(voice_probe, scenario):
|
||
requests = voice_probe[scenario]["tts"]
|
||
assert requests, "voice mode never reached /api/tts"
|
||
for request in requests:
|
||
assert "language" not in request
|
||
|
||
|
||
def test_voice_mode_drops_hostile_language_values(voice_probe):
|
||
for case in voice_probe["hostile_language_values_are_dropped"]["results"]:
|
||
for request in case["tts"]:
|
||
assert "language" not in request, case["sent"]
|
||
|
||
|
||
def test_voice_mode_never_sends_a_voice_field(voice_probe):
|
||
for request in voice_probe["voice_field_is_never_sent"]["tts"]:
|
||
assert set(request) <= {"text", "engine", "language"}
|
||
assert "voice" not in request
|
||
|
||
|
||
def test_voice_mode_does_not_reuse_a_previous_turn_language(voice_probe):
|
||
requests = voice_probe["language_does_not_leak_into_later_turn"]["tts"]
|
||
assert len(requests) == 3
|
||
assert requests[0]["language"] == "ru"
|
||
assert "language" not in requests[1]
|
||
assert requests[2]["language"] == "es"
|
||
|
||
|
||
def test_voice_mode_ignores_language_from_an_empty_transcript(voice_probe):
|
||
result = voice_probe["empty_transcript_does_not_arm_a_language"]
|
||
assert result["sendsAfterBlank"] == []
|
||
assert result["sends"] == ["Hello."]
|
||
assert result["tts"], "the follow-up turn should still be spoken"
|
||
for request in result["tts"]:
|
||
assert "language" not in request
|
||
|
||
|
||
def test_voice_mode_discards_language_when_the_session_changes(voice_probe):
|
||
result = voice_probe["session_change_discards_language"]
|
||
assert result["afterSwitch"] == []
|
||
for request in result["tts"]:
|
||
assert "language" not in request
|
||
|
||
|
||
def test_voice_mode_discards_language_when_voice_mode_is_turned_off(voice_probe):
|
||
result = voice_probe["deactivation_discards_language"]
|
||
assert result["afterDeactivate"] == []
|
||
for request in result["tts"]:
|
||
assert "language" not in request
|
||
|
||
|
||
def test_voice_mode_speaks_nothing_when_transcription_fails(voice_probe):
|
||
result = voice_probe["transcribe_error_speaks_nothing"]
|
||
assert result["tts"] == []
|
||
assert any("Whisper is down" in toast for toast in result["toasts"])
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. Build-time enforcement and documented semantics
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_image_build_verifies_every_language_routing_patch():
|
||
dockerfile = (DOCKERFILES / "Dockerfile.hermes-webui").read_text(encoding="utf-8")
|
||
assert "'language': detected" in dockerfile
|
||
assert "def _atlas_tts_language(body):" in dockerfile
|
||
assert 'request_payload["language"] = _atlas_language' in dockerfile
|
||
assert '"language": detected_language' in dockerfile
|
||
assert "takeSttLanguage(token)" in dockerfile
|
||
assert "/opt/hermes-webui/api/upload.py" in dockerfile
|
||
assert "/opt/hermes/tools/transcription_tools.py" in dockerfile
|
||
|
||
|
||
def test_atlas_patch_roots_are_overridable_for_offline_verification():
|
||
patch = ATLAS_PATCH.read_text(encoding="utf-8")
|
||
assert 'os.environ.get("HERMES_WEBUI_PATCH_ROOT", "/opt/hermes-webui")' in patch
|
||
assert 'os.environ.get("HERMES_AGENT_PATCH_ROOT", "/opt/hermes")' in patch
|
||
|
||
|
||
def test_notes_document_the_stt_driven_voice_selection_and_its_limits():
|
||
notes = (HERMES / "NOTES.md").read_text(encoding="utf-8")
|
||
assert "STT-detected language" in notes
|
||
for marker in ("hands-free", "Typed messages", "en_US-amy-medium"):
|
||
assert marker in notes
|