185 lines
7.3 KiB
Python
185 lines
7.3 KiB
Python
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_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_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_is_split_into_short_complete_chunks(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), 4)
|
||
self.assertTrue(all(len(part) <= 60 for _, part in segments))
|
||
self.assertIn("29 Kilometer pro Stunde", spoken)
|
||
self.assertIn("Kein Regen erwartet", spoken)
|
||
self.assertIn("wetteronline Punkt de", spoken)
|
||
|
||
|
||
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))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|