124 lines
4.6 KiB
Python
124 lines
4.6 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)])
|
|
|
|
|
|
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()
|