import importlib.util import pathlib import unittest from array import array MODULE_PATH = pathlib.Path(__file__).with_name("tts_gateway.py") SPEC = importlib.util.spec_from_file_location("tts_gateway", MODULE_PATH) gateway = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(gateway) class LanguageSegmentationTests(unittest.TestCase): def setUp(self): self.original_code_switch = gateway.CODE_SWITCH_ENABLED def tearDown(self): gateway.CODE_SWITCH_ENABLED = self.original_code_switch def test_german_only(self): self.assertEqual( gateway.segment_languages("Guten Abend, wie warm ist es heute?"), [("de", "Guten Abend, wie warm ist es heute?")], ) def test_english_only(self): text = "This is a short test and it is running on the local server." self.assertEqual(gateway.segment_languages(text), [("en", text)]) def test_mixed_compounds(self): gateway.CODE_SWITCH_ENABLED = True text = "Ich öffne das Unraid-Dashboard und prüfe die Docker-Container." self.assertEqual( gateway.segment_languages(text), [ ("de", "Ich öffne das "), ("en", "Unraid-Dashboard"), ("de", " und prüfe die "), ("en", "Docker-Container"), ("de", "."), ], ) def test_operational_vocabulary_is_not_read_as_german(self): gateway.CODE_SWITCH_ENABLED = True text = ( "31 von 60 Containern laufen, alle sind up und healthy. " "Die Health-Checks zeigen keine Crashes oder Restart-Loops. " "Immich meldet unsupported image format und premature end of JPEG." ) segments = gateway.segment_languages(text) english = [part.lower() for language, part in segments if language == "en"] self.assertIn("containern", english) self.assertIn("up", english) self.assertIn("healthy", english) self.assertIn("health-checks", english) self.assertIn("crashes", english) self.assertIn("restart-loops", english) self.assertIn("unsupported image format", english) self.assertIn("premature end of jpeg", english) prepared = gateway.prepare_segments(text) self.assertTrue(all(any(char.isalnum() for char in part) for _, part in prepared)) def test_long_text_is_cleaned_and_chunked(self): text = "✅ **Status**\n" + ("Ein langer deutscher Diagnosesatz. " * 30) segments = gateway.prepare_segments(text) self.assertTrue(segments) self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS for _, part in segments)) self.assertNotIn("✅", "".join(part for _, part in segments)) def test_inflected_container_has_english_pronunciation(self): gateway.CODE_SWITCH_ENABLED = True prepared = gateway.prepare_segments("Ich prüfe die Containern.") self.assertIn(("en", "containers."), prepared) def test_code_switch_is_disabled_by_default(self): gateway.CODE_SWITCH_ENABLED = False text = "Docker-Container laufen, die Health-Checks melden healthy." self.assertEqual(gateway.segment_languages(text), [("de", text)]) def test_weather_notation_is_spoken_naturally(self): text = ( "Rastatt (76437), Sonntag 23.08.: 22° / 10°, " "0 % Regen. Quellen: wetter.com und wetteronline.de" ) spoken = gateway.clean_for_speech(text) self.assertIn("Postleitzahl sieben sechs vier drei sieben", spoken) self.assertIn("23. August", spoken) self.assertIn("Höchstwert 22 Grad, Tiefstwert 10 Grad", spoken) self.assertIn("0 Prozent Regen", spoken) self.assertIn("wetter Punkt C O M", spoken) self.assertIn("wetteronline Punkt de", spoken) def test_full_date_and_raw_url_are_normalized(self): spoken = gateway.clean_for_speech( "Am 23.08.2026 steht es auf https://www.example.org/path?q=1." ) self.assertIn("23. August 2026", spoken) self.assertIn("example Punkt O R G", spoken) self.assertNotIn("https", spoken) def test_wind_speed_is_spoken_as_a_unit(self): for text in ( "Böen bis etwa 29 km/h.", "Böen bis etwa 29 KM / H.", "Wind mit 29 kmh.", ): with self.subTest(text=text): spoken = gateway.clean_for_speech(text) self.assertIn("29 Kilometer pro Stunde", spoken) self.assertNotIn("km", spoken.lower()) def test_qwen_normalizes_ipv4_time_date_and_count(self): spoken = gateway.prepare_for_qwen_speech( "full_kiosk (192.168.1.5): 2× Timeout um 02:14 seit 01.09." ) self.assertIn("full kiosk", spoken) self.assertIn("192 Punkt 168 Punkt 1 Punkt 5", spoken) self.assertIn("2 mal Timeout", spoken) self.assertIn("2 Uhr 14", spoken) self.assertIn("1. September", spoken) def test_qwen_speaks_temperature_and_time_ranges_with_bis(self): spoken = gateway.prepare_for_qwen_speech( "15°–23 °C, 93 %, ca. 12 mm, vor allem um 05–06 Uhr." ) self.assertIn("15 bis 23 Grad", spoken) self.assertIn("93 Prozent", spoken) self.assertIn("5 bis 6 Uhr", spoken) def test_qwen_keeps_prosody_punctuation(self): spoken = gateway.prepare_for_qwen_speech( "Ist das gut? Ja! SarahTV: erreichbar." ) self.assertEqual(spoken, "Ist das gut? Ja! SarahTV: erreichbar.") def test_visual_punctuation_becomes_natural_pauses(self): spoken = gateway.clean_for_speech( "Status: stabil – keine Fehler; Docker-Container laufen." ) self.assertEqual( spoken, "Status, stabil, keine Fehler, Docker Container laufen.", ) self.assertNotRegex(spoken, r"[:;\-‐‑‒–—−]") def test_question_and_exclamation_marks_are_not_sent_to_xtts(self): spoken = gateway.clean_for_speech("Wie geht es dir? Wirklich gut!") self.assertEqual(spoken, "Wie geht es dir. Wirklich gut.") self.assertNotRegex(spoken, r"[!?]") def test_punctuation_only_segments_are_never_synthesized(self): segments = gateway.prepare_segments("Status: – alles läuft.") self.assertTrue(segments) self.assertTrue(all(any(char.isalnum() for char in part) for _, part in segments)) def test_weather_summary_keeps_complete_sentences_together(self): text = ( "Heute in Rastatt: teils sonnig, trocken, 10 bis 22 Grad. " "Abends wolkiger, 16 bis 21 Grad. Böen bis 29 km/h. " "Kein Regen erwartet. Quelle: wetter.com, wetteronline.de" ) segments = gateway.prepare_segments(text) spoken = " ".join(part for _, part in segments) self.assertEqual(len(segments), 1) self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS for _, part in segments)) self.assertIn("29 Kilometer pro Stunde", spoken) self.assertIn("Kein Regen erwartet", spoken) self.assertIn("wetteronline Punkt de", spoken) def test_short_followup_sentence_shares_the_same_xtts_request(self): text = ( "Ehrlich gesagt habe ich keine echten Gefühle wie Menschen, aber " "ich bin wach, aufmerksam und motiviert, dir zu helfen. " "Klingt das gut?" ) segments = gateway.prepare_segments(text) self.assertEqual(len(segments), 1) self.assertIn("dir zu helfen. Klingt das gut.", segments[0][1]) def test_xtts_receives_a_stable_terminal_stop_cue(self): self.assertEqual( gateway._stabilize_xtts_ending("Heute bleibt es trocken."), "Heute bleibt es trocken;", ) self.assertEqual( gateway._stabilize_xtts_ending("Klingt das gut."), "Klingt das gut;", ) def test_normal_sentence_is_not_broken_into_word_sized_requests(self): text = ( "Die automatische Komprimierung ist in OpenClaw standardmäßig " "aktiviert und lässt sich über die Konfigurationsdatei steuern." ) self.assertEqual(gateway.prepare_segments(text), [("de", text)]) def test_overlong_sentence_prefers_clause_boundaries(self): clause = "dieser natürlich gesprochene Teilsatz bleibt zusammen," text = " ".join([clause] * 12) + " und endet hier." segments = gateway.prepare_segments(text) self.assertGreater(len(segments), 1) self.assertTrue(all(len(part) <= gateway.CHUNK_CHARS for _, part in segments)) self.assertTrue(all(len(part.split()) > 4 for _, part in segments)) class FallbackTests(unittest.TestCase): def setUp(self): self.original_xtts = gateway.synthesize_xtts self.original_piper = gateway.synthesize_piper def tearDown(self): gateway.synthesize_xtts = self.original_xtts gateway.synthesize_piper = self.original_piper def test_piper_is_used_when_xtts_fails(self): def fail(*_args): raise RuntimeError("synthetic XTTS failure") gateway.synthesize_xtts = fail gateway.synthesize_piper = lambda *_args: (b"piper", "audio/wav") self.assertEqual( gateway.synthesize("synthetic test", "wav", 1.0), (b"piper", "audio/wav"), ) class AudioJoinTests(unittest.TestCase): @staticmethod def pcm(values): samples = array("h", values) return samples.tobytes() def test_trim_removes_long_silent_edges(self): pcm = self.pcm(([0] * 1000) + ([500] * 100) + ([0] * 1000)) trimmed = gateway._trim_pcm(pcm) self.assertLess(len(trimmed), len(pcm)) self.assertGreater(len(trimmed), 100 * 2) def test_join_inserts_pause_only_after_sentence(self): spoken = self.pcm([500] * 1000) inline = gateway._join_pcm([("Docker", spoken), ("Container", spoken)]) sentence = gateway._join_pcm([("Fertig.", spoken), ("Weiter", spoken)]) self.assertGreater(len(sentence), len(inline)) def test_long_internal_silence_is_shortened(self): spoken = self.pcm([900] * 1000) long_silence = self.pcm([0] * int(24000 * 1.5)) result = gateway._compress_internal_silence(spoken + long_silence + spoken) expected_keep = int(24000 * gateway.INTERNAL_SILENCE_KEEP_MS / 1000) * 2 self.assertEqual(len(result), len(spoken) * 2 + expected_keep) def test_natural_short_pause_is_preserved(self): spoken = self.pcm([900] * 1000) short_silence = self.pcm([0] * int(24000 * 0.3)) original = spoken + short_silence + spoken self.assertEqual(gateway._compress_internal_silence(original), original) if __name__ == "__main__": unittest.main()