From 9bf4b2f0d30a684cda81cf876db1c3e4bd7d43e3 Mon Sep 17 00:00:00 2001 From: CoffeeMethod Date: Fri, 21 Aug 2026 16:03:00 -0600 Subject: [PATCH 1/3] Fixed 1.Uncontrolled data used in path expression #5 2.Uncontrolled data used in path expression #4 3.Polynomial regular expression used on uncontrolled data #3 --- gui.py | 6 ++++-- kokoro_engine.py | 3 ++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/gui.py b/gui.py index a79aa8c..d748ebc 100644 --- a/gui.py +++ b/gui.py @@ -599,8 +599,10 @@ def save_fx_preset_dialog(self): def load_fx_preset(self, name): if name == "Select FX Preset...": return - - fpath = os.path.join(FX_PRESETS_DIR, f"{name}.json") + + safe_name = os.path.basename(name) + if not safe_name: return + fpath = os.path.join(FX_PRESETS_DIR, f"{safe_name}.json") if os.path.exists(fpath): try: with open(fpath, "r") as f: diff --git a/kokoro_engine.py b/kokoro_engine.py index 9f1067c..927763b 100644 --- a/kokoro_engine.py +++ b/kokoro_engine.py @@ -462,7 +462,8 @@ def parse_multispeaker_text(self, text): Returns a list of (speaker_name, fx_name, text_segment) """ # Regex to find [Name]: or [Name:FX]: - pattern = r"\[([^\]]+)\]:\s*" + + pattern = r"\[([^\]\n]{1,100})\]:\s*" matches = list(re.finditer(pattern, text)) if not matches: From b85a20f52cb9af7ae7586f5145abde7f460bf77d Mon Sep 17 00:00:00 2001 From: CoffeeMethod Date: Fri, 21 Aug 2026 17:47:48 -0600 Subject: [PATCH 2/3] Add pytest test suite for engine and GUI logic Covers batch/JIT conversion, caching (incl. cache-invalidation policy), lexicon substitution, text chunking, audio processing, voice mixing, presets, path resolution, SRT/preview generation, and GUI config assembly/handlers/settings. Adds pytest.ini, requirements-test.txt, and an integration test for real synthesis. --- README.md | 25 +++ pytest.ini | 5 + requirements-test.txt | 1 + tests/__init__.py | 0 tests/conftest.py | 239 +++++++++++++++++++++++ tests/integration/__init__.py | 0 tests/integration/test_real_synthesis.py | 125 ++++++++++++ tests/test_batch_conversion.py | 109 +++++++++++ tests/test_caching.py | 111 +++++++++++ tests/test_generate_preview.py | 69 +++++++ tests/test_generate_srt.py | 27 +++ tests/test_gui_config_assembly.py | 117 +++++++++++ tests/test_gui_handlers.py | 118 +++++++++++ tests/test_gui_settings.py | 49 +++++ tests/test_jit_conversion.py | 38 ++++ tests/test_lexicon.py | 35 ++++ tests/test_meta_caching_policy.py | 19 ++ tests/test_mix_voices.py | 52 +++++ tests/test_presets.py | 64 ++++++ tests/test_process_audio.py | 92 +++++++++ tests/test_resolve_voice_path.py | 27 +++ tests/test_text_processing.py | 116 +++++++++++ 22 files changed, 1438 insertions(+) create mode 100644 pytest.ini create mode 100644 requirements-test.txt create mode 100644 tests/__init__.py create mode 100644 tests/conftest.py create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/test_real_synthesis.py create mode 100644 tests/test_batch_conversion.py create mode 100644 tests/test_caching.py create mode 100644 tests/test_generate_preview.py create mode 100644 tests/test_generate_srt.py create mode 100644 tests/test_gui_config_assembly.py create mode 100644 tests/test_gui_handlers.py create mode 100644 tests/test_gui_settings.py create mode 100644 tests/test_jit_conversion.py create mode 100644 tests/test_lexicon.py create mode 100644 tests/test_meta_caching_policy.py create mode 100644 tests/test_mix_voices.py create mode 100644 tests/test_presets.py create mode 100644 tests/test_process_audio.py create mode 100644 tests/test_resolve_voice_path.py create mode 100644 tests/test_text_processing.py diff --git a/README.md b/README.md index d182f3c..e45285a 100644 --- a/README.md +++ b/README.md @@ -91,6 +91,31 @@ https://github.com/user-attachments/assets/c75e7141-5d73-40f4-b182-d4f5bc49ad1e - Click "Preview Audio" to hear a short sample. - Click "Start Generation" (or "Start Real-time JIT") to begin. +## Running Tests + +The project has a `pytest` suite under `tests/` covering both `gui.py` and `kokoro_engine.py`. Because `kokoro_engine.py` imports the Windows-only `winsound` module unconditionally, **the suite only runs on Windows.** + +1. **Install test dependencies** (on top of `requirements.txt`): + ```bash + pip install -r requirements-test.txt + ``` + +2. **Run the fast suite** (default): + ```bash + pytest + ``` + This mocks the Kokoro pipeline, so it runs in seconds with no model download and no eSpeak NG required. Caching is disabled by default in every test except `tests/test_caching.py`. + +3. **Run the integration suite** (opt-in, real synthesis): + ```bash + pytest -m integration tests/integration -s + ``` + Uses the real Kokoro pipeline, so it needs eSpeak NG on `PATH` (see Prerequisites) and downloads model weights on first use. It skips automatically if `espeak-ng` isn't found. Since real synthesis can't be verified automatically, each test speaks a short, self-describing sample naming the voice/mode and writes it to `tests/output//.../*_transcript.txt` next to the generated `.wav` — listen to the audio and compare against the transcript to confirm it sounds right. The `-s` flag also prints the same text to the terminal as each test runs. + +### CI + +There's no CI workflow configured in this repo yet. A minimal one only needs to run step 2 above (`pytest`) on a `windows-latest` runner after installing `requirements.txt` + `requirements-test.txt` — the fast suite needs no eSpeak NG or model download, so it's safe to run on every push/PR. The integration suite is slow and pulls model weights, so it's better left as a manual/opt-in job rather than part of the default pipeline. + ## Technologies Used - **[Kokoro](https://github.com/hexgrad/kokoro):** The core TTS engine. diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..0931d05 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +markers = + integration: real KPipeline/torch/espeak-ng synthesis tests (slow, skipped by default) +addopts = -m "not integration" --strict-markers diff --git a/requirements-test.txt b/requirements-test.txt new file mode 100644 index 0000000..039d26e --- /dev/null +++ b/requirements-test.txt @@ -0,0 +1 @@ +pytest>=8.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..feb55e5 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,239 @@ +""" +Shared fixtures for the KokoroGUI test suite. + +Policy: every test that generates a config dict should build it through +`make_config`, which defaults `caching=False`. Only tests/test_caching.py +is allowed to override that to True (enforced by +tests/test_meta_caching_policy.py). This keeps caching off by default +without every test having to remember to pass it explicitly. +""" +import os +import re +import sys +import time +import threading +import concurrent.futures +import subprocess +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import numpy as np +import pytest +import torch + +import kokoro_engine +from kokoro_engine import KokoroEngine + +# On some Windows Store ("WindowsApps") Python installs, Tcl/Tk's own +# init.tcl discovery intermittently fails against the package-virtualized +# path when many Tk() roots are created/destroyed across a test session +# (each GUI test builds a real TTSApp). Pointing TCL_LIBRARY/TK_LIBRARY at +# the known-good path once avoids repeated, occasionally-flaky rediscovery. +_tcl_dir = os.path.join(sys.base_prefix, "tcl", "tcl8.6") +_tk_dir = os.path.join(sys.base_prefix, "tcl", "tk8.6") +if os.path.isdir(_tcl_dir): + os.environ.setdefault("TCL_LIBRARY", _tcl_dir) +if os.path.isdir(_tk_dir): + os.environ.setdefault("TK_LIBRARY", _tk_dir) + +# One shared timestamp per pytest invocation, mirroring gui.py's +# self.timecode_format = "%Y%m%d%H%M%S" convention (gui.py:96). +_RUN_TS = time.strftime("%Y%m%d%H%M%S") + + +# --------------------------------------------------------------------------- +# Engine-level fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def isolated_dirs(tmp_path, monkeypatch): + """Redirect kokoro_engine's module-level storage dirs into tmp_path.""" + custom_voices = tmp_path / "custom_voices" + cache_dir = tmp_path / "cache" + out_dir = tmp_path / "out" + for d in (custom_voices, cache_dir, out_dir): + d.mkdir() + monkeypatch.setattr(kokoro_engine, "CUSTOM_VOICES_DIR", str(custom_voices)) + monkeypatch.setattr(kokoro_engine, "CACHE_DIR", str(cache_dir)) + return SimpleNamespace(custom_voices=custom_voices, cache_dir=cache_dir, out_dir=out_dir) + + +@pytest.fixture +def engine(isolated_dirs, monkeypatch): + # Never touch the real audio device from a test. + monkeypatch.setattr(kokoro_engine, "winsound", MagicMock()) + e = KokoroEngine() + yield e + e.worker.stop() + + +@pytest.fixture +def real_engine(isolated_dirs, monkeypatch): + """Real, unmocked KokoroEngine for tests/integration's opt-in real-pipeline + tests. Identical to `engine` (isolated custom_voices/cache dirs, mocked + winsound so playback never touches the real audio device) but never + combined with `fake_pipeline` - get_thread_pipeline/KPipeline resolve to + the real kokoro.KPipeline, so synthesis actually runs torch + espeak-ng.""" + monkeypatch.setattr(kokoro_engine, "winsound", MagicMock()) + e = KokoroEngine() + yield e + e.worker.stop() + + +class FakePipeline: + """Mimics kokoro.KPipeline's calling convention without any model/espeak-ng.""" + + def __init__(self, lang_code="a", segment_duration_s=0.05, sr=24000): + self.lang_code = lang_code + self.voices = {} + self._sr = sr + self._dur = segment_duration_s + + def __call__(self, text, voice=None, speed=1.0, split_pattern=r"\n+"): + try: + parts = [t.strip() for t in re.split(split_pattern, text) if t.strip()] + except re.error: + parts = [] + if not parts: + parts = [text] + n = max(1, int(self._sr * self._dur)) + for p in parts: + audio = (0.1 * np.sin(2 * np.pi * 220 * np.arange(n) / self._sr)).astype(np.float32) + yield p, "", audio + + def load_voice(self, name): + return torch.zeros(510, 1, 256) + + +@pytest.fixture +def fake_pipeline(monkeypatch): + fp = FakePipeline() + monkeypatch.setattr(kokoro_engine, "get_thread_pipeline", lambda lang_code="a": fp) + monkeypatch.setattr(kokoro_engine, "KPipeline", lambda lang_code="a": fp) + return fp + + +@pytest.fixture +def callback_recorder(engine): + rec = SimpleNamespace(statuses=[], progresses=[], finished=threading.Event()) + engine.on_status = lambda msg, is_err: rec.statuses.append((msg, is_err)) + engine.on_progress = lambda *a: rec.progresses.append(a) + engine.on_finish = lambda: rec.finished.set() + return rec + + +def wait_for_finish(rec, timeout=30): + assert rec.finished.wait(timeout), "engine.on_finish was never called within timeout" + + +@pytest.fixture +def make_config(isolated_dirs): + def _make(**overrides): + cfg = { + "voice": "af_heart", + "speed": 1.0, + "lang_code": "a", + "split_pattern": r"\n+", + "out_dir": str(isolated_dirs.out_dir), + "filename": "output", + "time_id": "0", + "format": "wav", + "num_threads": 1, + "combine": True, + "separate": True, + "export_subtitles": False, + "caching": False, # hard default OFF - see module docstring + "lexicon": {}, + } + cfg.update(overrides) + return cfg + return _make + + +@pytest.fixture +def timestamped_output_dir(request): + """ + tests/output/// - for tests whose generated + audio should persist for manual inspection (real-pipeline integration tests, + and a couple of "leaves_inspectable_output" smoke tests). Not used by + throwaway unit tests, which use tmp_path/isolated_dirs instead. + """ + slug = re.sub(r"[^A-Za-z0-9_-]+", "_", request.node.nodeid) + d = Path(__file__).parent / "output" / _RUN_TS / slug + d.mkdir(parents=True, exist_ok=True) + return d + + +def espeak_available(): + # kokoro_engine never shells out to an `espeak-ng` CLI - phonemization + # goes through misaki -> phonemizer's EspeakWrapper, pointed at the DLL + # and data dir that the `espeakng_loader` package bundles/resolves + # (see misaki/espeak.py). That's the actual runtime dependency, so + # check for it directly instead of probing PATH for a binary the app + # doesn't use. + try: + import espeakng_loader + return ( + os.path.isfile(espeakng_loader.get_library_path()) + and os.path.isdir(espeakng_loader.get_data_path()) + ) + except Exception: + return False + + +# --------------------------------------------------------------------------- +# GUI-level fixtures +# --------------------------------------------------------------------------- + +class StubEngine: + """Drop-in replacement for KokoroEngine used by GUI tests - never touches + the real Kokoro pipeline/model.""" + + def __init__(self): + self.pipeline = object() # truthy - passes the "engine still initializing" gate + self.worker = SimpleNamespace(run_coro=MagicMock(return_value=concurrent.futures.Future())) + self.cancel_event = threading.Event() + self.on_progress = None + self.on_status = None + self.on_finish = None + self.init_pipeline_async = MagicMock(return_value=None) + self.start_conversion = MagicMock() + self.start_jit_conversion = MagicMock() + self.generate_preview = MagicMock() + self.mix_voices = MagicMock() + self.extract_text_from_file = MagicMock(return_value="") + self.cancel = MagicMock() + + +@pytest.fixture +def tts_app(tmp_path, monkeypatch): + import gui + import tkinter + + monkeypatch.chdir(tmp_path) + monkeypatch.setattr(gui, "CONFIG_FILE", str(tmp_path / "config.json")) + monkeypatch.setattr(gui, "PRESETS_DIR", str(tmp_path / "presets")) + monkeypatch.setattr(gui, "FX_PRESETS_DIR", str(tmp_path / "presets" / "fx")) + monkeypatch.setattr(gui, "KokoroEngine", StubEngine) + monkeypatch.setattr(gui, "messagebox", MagicMock()) + monkeypatch.setattr(gui, "filedialog", MagicMock()) + (tmp_path / "custom_voices").mkdir() + + # Creating many real Tk() interpreters across a test session intermittently + # hits the same WindowsApps init.tcl read glitch as above - retry a few + # times rather than failing the whole test on a transient hiccup. + app = None + last_err = None + for _ in range(5): + try: + app = gui.TTSApp() + break + except tkinter.TclError as e: + last_err = e + time.sleep(0.2) + if app is None: + raise last_err + + yield app + app.destroy() diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_real_synthesis.py b/tests/integration/test_real_synthesis.py new file mode 100644 index 0000000..fbdfc5f --- /dev/null +++ b/tests/integration/test_real_synthesis.py @@ -0,0 +1,125 @@ +"""Opt-in integration tests using the REAL Kokoro pipeline (no mocking). + +Skipped by default (see pytest.ini's `addopts = -m "not integration"`). +Run explicitly with: + pytest -m integration tests/integration -s + +(the `-s` shows the spoken sample text in the terminal as each test runs; +without it, pytest still writes a matching `*_transcript.txt` next to +each `.wav` for the same purpose.) + +Requires the espeak-ng backend (via the `espeakng_loader` package, a +transitive dependency of kokoro/misaki that bundles its own espeak-ng.dll +and data dir - no system-wide espeak-ng install/PATH entry needed) plus +torch/kokoro model weights. Real audio is always written under the shared +timestamped_output_dir fixture so it persists for manual inspection (never +tmp_path, which pytest auto-cleans). + +Since this is real synthesis, correctness can't be asserted automatically - +each test speaks a short, self-describing sample naming the voice and mode +in use, so a human listening to the output can actually confirm it sounds +right (clear pronunciation, correct voice, no glitches/silence). +""" +import asyncio +import os + +import pytest + +import kokoro_engine +from kokoro_engine import KokoroEngine + + +def _espeak_available(): + # kokoro_engine never shells out to an `espeak-ng` CLI - phonemization + # goes through misaki -> phonemizer's EspeakWrapper, pointed at the DLL + # and data dir that `espeakng_loader` bundles/resolves. That's the + # actual runtime dependency, so check for it directly instead of + # probing PATH for a binary the app doesn't use. + try: + import espeakng_loader + return ( + os.path.isfile(espeakng_loader.get_library_path()) + and os.path.isdir(espeakng_loader.get_data_path()) + ) + except Exception: + return False + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif(not _espeak_available(), reason="espeakng_loader library/data not found"), +] + + +def _sample_text(mode, voice): + # A pangram + digits is a standard TTS smoke-test phrase: it exercises + # every letter and reads unambiguously by ear, so a mispronunciation or + # glitch is easy for a human to catch. Naming the mode/voice lets the + # listener confirm the right code path actually produced this file. + return ( + f"This is a Kokoro {mode} integration test, spoken by the {voice} voice. " + "Please confirm this is clear and understandable: " + "the quick brown fox jumps over the lazy dog. One, two, three, four, five." + ) + + +def _write_transcript(out_dir, stem, text): + path = out_dir / f"{stem}_transcript.txt" + path.write_text(text, encoding="utf-8") + return path + + +def _ensure_output_dir(out_dir): + # timestamped_output_dir is created by the conftest fixture, but these + # tests write real audio that's meant to persist for manual inspection, + # so guard against it having been removed (or never created) out from + # under us rather than failing deep inside the engine's file write. + if not out_dir.is_dir(): + out_dir.mkdir(parents=True, exist_ok=True) + return out_dir + + +def test_real_batch_conversion_produces_playable_audio(real_engine, timestamped_output_dir, make_config): + timestamped_output_dir = _ensure_output_dir(timestamped_output_dir) + success = asyncio.run(real_engine.init_pipeline_async(lang_code="a")) + assert success, "Real pipeline failed to init - check espeak-ng/kokoro install" + + voice = "af_heart" + text = _sample_text("batch conversion", voice) + config = make_config( + out_dir=str(timestamped_output_dir), voice=voice, + filename="realbatch", time_id="1", + ) + asyncio.run(real_engine._process_text_async(text, config)) + + combined = timestamped_output_dir / "realbatch_1_combined.wav" + assert combined.exists() + assert combined.stat().st_size > 1000 + + transcript = _write_transcript(timestamped_output_dir, "realbatch_1", text) + print(f"\n[HUMAN CONFIRMATION NEEDED] Listen to {combined}") + print(f"It should say: {text}") + print(f"(transcript saved to {transcript})") + + +def test_real_jit_conversion_with_playback_mocked(real_engine, timestamped_output_dir, make_config): + timestamped_output_dir = _ensure_output_dir(timestamped_output_dir) + success = asyncio.run(real_engine.init_pipeline_async(lang_code="a")) + assert success, "Real pipeline failed to init - check espeak-ng/kokoro install" + + voice = "af_heart" + text = _sample_text("real-time JIT", voice) + config = make_config( + out_dir=str(timestamped_output_dir), voice=voice, + filename="realjit", time_id="1", + ) + asyncio.run(real_engine._process_jit_async(text, config)) + + jit_output = timestamped_output_dir / "realjit_1_jit_output.wav" + assert jit_output.exists() + assert jit_output.stat().st_size > 1000 + + transcript = _write_transcript(timestamped_output_dir, "realjit_1", text) + print(f"\n[HUMAN CONFIRMATION NEEDED] Listen to {jit_output}") + print(f"It should say: {text}") + print(f"(transcript saved to {transcript})") diff --git a/tests/test_batch_conversion.py b/tests/test_batch_conversion.py new file mode 100644 index 0000000..6e501c1 --- /dev/null +++ b/tests/test_batch_conversion.py @@ -0,0 +1,109 @@ +"""Tests for process_chunk_task + _process_text_async, the batch conversion +pipeline (kokoro_engine.py:568-705, 910-1061). caching=False throughout +(via make_config's default) except where noted.""" +import asyncio +import json + + +def test_process_chunk_task_writes_named_part_files(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(filename="myrun", time_id="20260101000000") + results = engine.process_chunk_task((3, "Hello there.", config), None) + + assert len(results) == 1 + expected = isolated_dirs.out_dir / "myrun_20260101000000_part3_0.wav" + assert expected.exists() + assert results[0]["path"] == str(expected) + + +def test_process_text_async_combine_true_writes_combined_file(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(combine=True, filename="run", time_id="1") + asyncio.run(engine._process_text_async("Hello world.", config)) + + assert (isolated_dirs.out_dir / "run_1_combined.wav").exists() + + +def test_process_text_async_export_subtitles_writes_srt(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(export_subtitles=True, filename="run", time_id="1") + asyncio.run(engine._process_text_async("Hello world.", config)) + + srt_path = isolated_dirs.out_dir / "run_1_combined.srt" + assert srt_path.exists() + assert "-->" in srt_path.read_text(encoding="utf-8") + + +def test_process_text_async_separate_false_deletes_part_files(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(separate=False, combine=True, filename="run", time_id="1") + asyncio.run(engine._process_text_async("Hello world.", config)) + + assert list(isolated_dirs.out_dir.glob("run_1_part*")) == [] + assert (isolated_dirs.out_dir / "run_1_combined.wav").exists() + + +def test_process_text_async_no_text_calls_on_finish_and_status(engine, fake_pipeline, make_config, callback_recorder): + config = make_config() + asyncio.run(engine._process_text_async(" ", config)) + + assert callback_recorder.finished.is_set() + assert any("No text" in msg for msg, _ in callback_recorder.statuses) + + +def test_process_text_async_chunk_exception_does_not_abort_batch( + engine, fake_pipeline, make_config, isolated_dirs, monkeypatch, callback_recorder +): + config = make_config(filename="run", time_id="1", num_threads=1) + # Two multispeaker segments guarantee two chunks even with num_threads=1 + # (a single unmarked segment would be merged into one smart_split chunk). + text = "[SpeakerA]: First paragraph.\n\n[SpeakerB]: Second paragraph." + + real_task = engine.process_chunk_task + call_count = {"n": 0} + + def flaky(chunk_data, progress_callback): + call_count["n"] += 1 + if call_count["n"] == 1: + raise RuntimeError("boom") + return real_task(chunk_data, progress_callback) + + monkeypatch.setattr(engine, "process_chunk_task", flaky) + asyncio.run(engine._process_text_async(text, config)) + + assert any("Error in chunk" in msg for msg, is_err in callback_recorder.statuses if is_err) + assert (isolated_dirs.out_dir / "run_1_combined.wav").exists() + + +def test_multispeaker_preset_and_fx_preset_layering(engine, fake_pipeline, make_config, isolated_dirs, monkeypatch, tmp_path): + monkeypatch.chdir(tmp_path) + presets_dir = tmp_path / "presets" + fx_dir = presets_dir / "fx" + presets_dir.mkdir(exist_ok=True) + fx_dir.mkdir(exist_ok=True) + + (presets_dir / "Narrator.json").write_text(json.dumps({"voice": "am_adam", "speed": 1.25}), encoding="utf-8") + (fx_dir / "Radio.json").write_text(json.dumps({"reverb_enabled": True, "reverb_room_size": 0.9}), encoding="utf-8") + + config = make_config(filename="run", time_id="1") + text = "[Narrator:Radio]: Hello from the narrator." + + captured = {} + real_task = engine.process_chunk_task + + def spy(chunk_data, progress_callback): + captured["config"] = chunk_data[2] + return real_task(chunk_data, progress_callback) + + monkeypatch.setattr(engine, "process_chunk_task", spy) + asyncio.run(engine._process_text_async(text, config)) + + assert captured["config"]["voice"] == "am_adam" + assert captured["config"]["speed"] == 1.25 + assert captured["config"]["reverb_enabled"] is True + assert captured["config"]["apply_fx"] is True + + +def test_full_batch_conversion_leaves_inspectable_output(engine, fake_pipeline, make_config, timestamped_output_dir): + config = make_config(out_dir=str(timestamped_output_dir), filename="sample", time_id="smoke") + asyncio.run(engine._process_text_async("This audio should be inspectable by a human.", config)) + + combined = timestamped_output_dir / "sample_smoke_combined.wav" + assert combined.exists() + assert combined.stat().st_size > 0 diff --git a/tests/test_caching.py b/tests/test_caching.py new file mode 100644 index 0000000..dc69309 --- /dev/null +++ b/tests/test_caching.py @@ -0,0 +1,111 @@ +"""Tests for process_chunk_task's caching logic (kokoro_engine.py:568-705). + +This is the ONLY test module allowed to pass caching=True - see +tests/test_meta_caching_policy.py for the enforced guard. +""" +import hashlib +import os + +import numpy as np +import pytest +import soundfile as sf + +import kokoro_engine + + +def _hash(text, voice, speed, lang_code): + return hashlib.md5(f"{text}|{voice}|{speed}|{lang_code}".encode("utf-8")).hexdigest() + + +def test_cache_miss_writes_raw_pre_fx_audio(engine, fake_pipeline, isolated_dirs, make_config): + config = make_config(caching=True, volume=0.5) + results = engine.process_chunk_task((0, "Hello world.", config), None) + + h = _hash("Hello world.", config["voice"], config["speed"], config["lang_code"]) + cache_file = isolated_dirs.cache_dir / f"{h}_0.wav" + assert cache_file.exists() + + cached_audio, _ = sf.read(str(cache_file)) + output_audio, _ = sf.read(results[0]["path"]) + + # Cache stores the RAW pipeline output; the on-disk output segment has + # volume=0.5 applied on top - so the cached peak should be noticeably louder. + assert np.max(np.abs(cached_audio)) > np.max(np.abs(output_audio)) * 1.5 + + +def test_cache_hit_skips_pipeline_call(engine, isolated_dirs, make_config, monkeypatch): + config = make_config(caching=True) + text = "Hello world." + h = _hash(text, config["voice"], config["speed"], config["lang_code"]) + + audio = (0.1 * np.sin(2 * np.pi * 220 * np.arange(1200) / 24000)).astype(np.float32) + sf.write(str(isolated_dirs.cache_dir / f"{h}_0.wav"), audio, 24000) + + def _boom(lang_code="a"): + raise AssertionError("pipeline should not be called on a cache hit") + + monkeypatch.setattr(kokoro_engine, "get_thread_pipeline", _boom) + + results = engine.process_chunk_task((0, text, config), None) + + assert len(results) == 1 + assert os.path.exists(results[0]["path"]) + + +def test_cache_key_ignores_split_pattern(engine, fake_pipeline, make_config, monkeypatch): + text = "Hello world." + config1 = make_config(caching=True, split_pattern=r"\n+") + engine.process_chunk_task((0, text, config1), None) + + def _boom(lang_code="a"): + raise AssertionError("pipeline should not be called - same hash should hit cache") + + monkeypatch.setattr(kokoro_engine, "get_thread_pipeline", _boom) + + # Known limitation (kokoro_engine.py:618-621): split_pattern is not part + # of the cache key, so a different split_pattern that yields the same + # predicted segment count still counts as a cache hit. + config2 = make_config(caching=True, split_pattern=r"\n\n+") + results = engine.process_chunk_task((0, text, config2), None) + + assert len(results) == 1 + + +def test_cache_partial_files_missing_forces_regeneration(engine, fake_pipeline, isolated_dirs, make_config): + text = "Seg one.\n\nSeg two." + config = make_config(caching=True) + h = _hash(text, config["voice"], config["speed"], config["lang_code"]) + + # Only the first of the two expected segments is cached. + audio = (0.1 * np.sin(2 * np.pi * 220 * np.arange(1200) / 24000)).astype(np.float32) + sf.write(str(isolated_dirs.cache_dir / f"{h}_0.wav"), audio, 24000) + + results = engine.process_chunk_task((0, text, config), None) + + assert len(results) == 2 + assert (isolated_dirs.cache_dir / f"{h}_0.wav").exists() + assert (isolated_dirs.cache_dir / f"{h}_1.wav").exists() + + +def test_pitch_affects_cache_key(engine, fake_pipeline, isolated_dirs, make_config): + text = "Hello world." + config_a = make_config(caching=True, pitch=0.0) + config_b = make_config(caching=True, pitch=5.0) + + engine.process_chunk_task((0, text, config_a), None) + engine.process_chunk_task((0, text, config_b), None) + + cache_files = list(isolated_dirs.cache_dir.glob("*_0.wav")) + assert len(cache_files) == 2 + + +def test_speed_affects_cache_key(engine, fake_pipeline, isolated_dirs, make_config): + text = "Hello world." + config_a = make_config(caching=True, speed=1.0) + config_b = make_config(caching=True, speed=1.5) + + engine.process_chunk_task((0, text, config_a), None) + engine.process_chunk_task((0, text, config_b), None) + + cache_files = list(isolated_dirs.cache_dir.glob("*_0.wav")) + assert len(cache_files) == 2 diff --git a/tests/test_generate_preview.py b/tests/test_generate_preview.py new file mode 100644 index 0000000..d4e4780 --- /dev/null +++ b/tests/test_generate_preview.py @@ -0,0 +1,69 @@ +"""Tests for KokoroEngine.generate_preview (kokoro_engine.py:339-430).""" +import asyncio +import os + +import torch + + +def test_generate_preview_writes_wav_file(engine, fake_pipeline, tmp_path): + out_path = str(tmp_path / "preview.wav") + ok = asyncio.run(engine.generate_preview("Hello there.", "af_heart", 1.0, out_path)) + + assert ok is True + assert os.path.exists(out_path) + assert os.path.getsize(out_path) > 0 + + +def test_generate_preview_truncates_multispeaker_to_two_segments(engine, fake_pipeline, tmp_path, monkeypatch): + calls = [] + orig_call = fake_pipeline.__call__ + + def spy(self, text, voice=None, speed=1.0, split_pattern=r"\n+"): + calls.append(text) + return orig_call(text, voice=voice, speed=speed, split_pattern=split_pattern) + + monkeypatch.setattr(type(fake_pipeline), "__call__", spy) + + text = "[SpkA]: one\n\n[SpkB]: two\n\n[SpkC]: three" + out_path = str(tmp_path / "preview.wav") + asyncio.run(engine.generate_preview(text, "af_heart", 1.0, out_path)) + + assert len(calls) == 2 + + +def test_generate_preview_applies_lexicon_from_extra_config(engine, fake_pipeline, tmp_path, monkeypatch): + seen = {} + orig_call = fake_pipeline.__call__ + + def spy(self, text, voice=None, speed=1.0, split_pattern=r"\n+"): + seen["text"] = text + return orig_call(text, voice=voice, speed=speed, split_pattern=split_pattern) + + monkeypatch.setattr(type(fake_pipeline), "__call__", spy) + + out_path = str(tmp_path / "preview.wav") + asyncio.run(engine.generate_preview( + "Hello world.", "af_heart", 1.0, out_path, + extra_config={"lexicon": {"world": "planet"}}, + )) + + assert "planet" in seen["text"] + + +def test_generate_preview_voice_tensor_sets_pipeline_voices_dict(engine, fake_pipeline, tmp_path): + tensor = torch.zeros(510, 1, 256) + out_path = str(tmp_path / "preview.wav") + ok = asyncio.run(engine.generate_preview("Hello.", "ignored_voice", 1.0, out_path, voice_tensor=tensor)) + + assert ok is True + assert "_preview_temp" in fake_pipeline.voices + + +def test_generate_preview_no_pipeline_returns_false(engine, monkeypatch, tmp_path): + import kokoro_engine + monkeypatch.setattr(kokoro_engine, "get_thread_pipeline", lambda lang_code="a": None) + + out_path = str(tmp_path / "preview.wav") + ok = asyncio.run(engine.generate_preview("Hello.", "af_heart", 1.0, out_path)) + + assert ok is False diff --git a/tests/test_generate_srt.py b/tests/test_generate_srt.py new file mode 100644 index 0000000..dfcf654 --- /dev/null +++ b/tests/test_generate_srt.py @@ -0,0 +1,27 @@ +"""Tests for KokoroEngine.generate_srt (kokoro_engine.py:545-566).""" + + +def test_generate_srt_writes_valid_timing_format(engine, tmp_path): + segments = [ + {"text": "Hello.", "duration": 1.5}, + {"text": "World.", "duration": 2.0}, + ] + out_path = str(tmp_path / "out.srt") + + ok = engine.generate_srt(segments, out_path) + + assert ok is True + content = open(out_path, encoding="utf-8").read() + assert "-->" in content + assert "00:00:00,000 --> 00:00:01,500" in content + assert "00:00:01,500 --> 00:00:03,500" in content + assert "1\nHello." not in content # text is on its own line, not glued to the index + assert "Hello." in content and "World." in content + + +def test_generate_srt_write_failure_returns_false(engine, tmp_path): + bad_path = str(tmp_path / "missing_dir" / "out.srt") + + ok = engine.generate_srt([{"text": "x", "duration": 1.0}], bad_path) + + assert ok is False diff --git a/tests/test_gui_config_assembly.py b/tests/test_gui_config_assembly.py new file mode 100644 index 0000000..5d464b9 --- /dev/null +++ b/tests/test_gui_config_assembly.py @@ -0,0 +1,117 @@ +"""Tests for the config-dict assembly contract in start_conversion/ +preview_conversion (gui.py:1528-1747).""" +import os +import re +import tempfile + + +def _set_text(app, text): + app.text_entry.delete("1.0", "end") + app.text_entry.insert("1.0", text) + + +BASE_KEYS = { + "lang_code", "voice", "speed", "split_pattern", "filename", "format", + "out_dir", "separate", "combine", "export_subtitles", "caching", + "time_id", "num_threads", "volume", "pitch", "normalize", + "trim_silence", "lexicon", +} + +FX_KEYS = { + "reverb_enabled", "reverb_room_size", "reverb_wet_level", "reverb_damping", + "reverb_dry_level", "reverb_width", "eq_bass", "eq_treble", + "comp_enabled", "comp_threshold", "comp_ratio", "comp_attack", "comp_release", + "distortion_enabled", "distortion_drive", + "chorus_enabled", "chorus_rate", "chorus_depth", "chorus_mix", + "phaser_enabled", "phaser_rate", "phaser_depth", "phaser_mix", + "clipping_enabled", "clipping_thresh", + "bitcrush_enabled", "bitcrush_depth", "gsm_enabled", + "highpass_enabled", "highpass_freq", "lowpass_enabled", "lowpass_freq", + "delay_enabled", "delay_time", "delay_feedback", "delay_mix", + "pitch_shift_enabled", "pitch_shift_semitones", + "limiter_enabled", "limiter_threshold", "limiter_release", + "gain_enabled", "gain_db", +} + + +def test_start_conversion_assembles_full_key_set(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.apply_fx_var.set(True) + tts_app.start_conversion() + + assert tts_app.engine.start_conversion.called + text_arg, config = tts_app.engine.start_conversion.call_args[0] + assert text_arg == "Hello world." + assert BASE_KEYS <= config.keys() + assert FX_KEYS <= config.keys() + assert re.fullmatch(r"\d{14}", config["time_id"]) + + +def test_start_conversion_apply_fx_false_omits_fx_keys(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.apply_fx_var.set(False) + tts_app.start_conversion() + + _, config = tts_app.engine.start_conversion.call_args[0] + assert "reverb_enabled" not in config + assert "gain_db" not in config + + +def test_start_conversion_jit_enabled_routes_to_start_jit_conversion(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.jit_enabled.set(True) + tts_app.start_conversion() + + assert tts_app.engine.start_jit_conversion.called + assert not tts_app.engine.start_conversion.called + + +def test_start_conversion_blocks_when_pipeline_not_ready(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.engine.pipeline = None + tts_app.start_conversion() + + assert not tts_app.engine.start_conversion.called + assert not tts_app.engine.start_jit_conversion.called + + +def test_start_conversion_empty_text_shows_warning(tts_app): + _set_text(tts_app, "") + tts_app.start_conversion() + + assert not tts_app.engine.start_conversion.called + assert not tts_app.engine.start_jit_conversion.called + + +def test_preview_conversion_assembles_smaller_extra_config(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.apply_fx_var.set(False) + tts_app.preview_conversion() + + assert tts_app.engine.generate_preview.called + args, kwargs = tts_app.engine.generate_preview.call_args + preview_text, voice, speed, out_path, extra_config = args[:5] + assert set(extra_config.keys()) == {"volume", "pitch", "normalize", "trim_silence", "lexicon"} + assert voice == tts_app.voice_var.get() + assert speed == tts_app.speed_var.get() + + +def test_preview_conversion_apply_fx_true_adds_fx_keys(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.apply_fx_var.set(True) + tts_app.preview_conversion() + + args, kwargs = tts_app.engine.generate_preview.call_args + extra_config = args[4] + assert FX_KEYS <= extra_config.keys() + assert "voice" not in extra_config + assert "out_dir" not in extra_config + + +def test_preview_conversion_uses_tempdir_wav_path(tts_app): + _set_text(tts_app, "Hello world.") + tts_app.preview_conversion() + + args, kwargs = tts_app.engine.generate_preview.call_args + out_path = args[3] + assert out_path == os.path.join(tempfile.gettempdir(), "kokoro_preview.wav") diff --git a/tests/test_gui_handlers.py b/tests/test_gui_handlers.py new file mode 100644 index 0000000..50bbca3 --- /dev/null +++ b/tests/test_gui_handlers.py @@ -0,0 +1,118 @@ +"""Tests for assorted GUI event handlers: lexicon add/delete, thread-count +clamp, mix-name validation, preset save-dialog sanitization, load_fx_preset +safety, and a documented existing bug in refresh_voice_lists.""" +import json +import os + +import pytest + + +def test_add_lexicon_rule_persists_and_refreshes(tts_app): + import gui + tts_app.lex_orig_var.set("hello") + tts_app.lex_replace_var.set("hi") + tts_app.add_lexicon_rule() + + assert tts_app.settings["lexicon"]["hello"] == "hi" + with open(gui.CONFIG_FILE, "r", encoding="utf-8") as f: + saved = json.load(f) + assert saved["lexicon"]["hello"] == "hi" + + +def test_add_lexicon_rule_empty_original_shows_warning(tts_app): + import gui + tts_app.lex_orig_var.set("") + tts_app.lex_replace_var.set("hi") + tts_app.add_lexicon_rule() + + assert tts_app.settings.get("lexicon", {}) == {} + assert gui.messagebox.showwarning.called + + +def test_delete_lexicon_rule_removes_key(tts_app): + tts_app.settings["lexicon"] = {"hello": "hi"} + tts_app.delete_lexicon_rule("hello") + + assert "hello" not in tts_app.settings["lexicon"] + + +@pytest.mark.parametrize("start,delta,expected", [ + (1, -5, 1), + (16, 5, 16), + (5, 2, 7), +]) +def test_change_threads_clamps_1_to_16(tts_app, start, delta, expected): + tts_app.num_threads_var.set(start) + tts_app.change_threads(delta) + assert tts_app.num_threads_var.get() == expected + + +def test_mix_voice_action_rejects_invalid_name_chars(tts_app): + tts_app.mix_name_var.set("bad name!") + tts_app.mix_voice_action() + + assert not tts_app.engine.mix_voices.called + + +def test_mix_voice_action_prompts_overwrite_confirmation(tts_app): + import gui + existing = tts_app.get_all_voices()[0] + tts_app.mix_name_var.set(existing) + gui.messagebox.askyesno.return_value = False + + tts_app.mix_voice_action() + + assert not tts_app.engine.mix_voices.called + + +def test_save_preset_dialog_sanitizes_name(tts_app, monkeypatch): + import gui + + class FakeDialog: + def __init__(self, *a, **kw): + pass + + def get_input(self): + return 'Bad/Na:me' + + monkeypatch.setattr(gui.ctk, "CTkInputDialog", FakeDialog) + tts_app.save_preset_dialog() + + assert os.path.exists(os.path.join(gui.PRESETS_DIR, "BadName.json")) + + +def test_save_fx_preset_dialog_sanitizes_name(tts_app, monkeypatch): + import gui + + class FakeDialog: + def __init__(self, *a, **kw): + pass + + def get_input(self): + return 'Weird?Nam*e' + + monkeypatch.setattr(gui.ctk, "CTkInputDialog", FakeDialog) + tts_app.save_fx_preset_dialog() + + assert os.path.exists(os.path.join(gui.FX_PRESETS_DIR, "WeirdName.json")) + + +def test_load_fx_preset_basename_sanitized(tts_app): + import gui + os.makedirs(gui.FX_PRESETS_DIR, exist_ok=True) + with open(os.path.join(gui.FX_PRESETS_DIR, "real.json"), "w", encoding="utf-8") as f: + json.dump({"gain_db": 3.0}, f) + + tts_app.load_fx_preset("../../real") + + assert tts_app.gain_db.get() == 3.0 + + +def test_refresh_voice_lists_crashes_if_custom_voices_dir_missing(tts_app): + # Documents an existing asymmetry at gui.py:693 (os.listdir with no + # os.path.exists guard), unlike get_all_voices (gui.py:192) which does + # guard. Pins current behavior - do not silently "fix" by changing this + # assertion; if the guard is added, update this test deliberately. + os.rmdir("custom_voices") + with pytest.raises(FileNotFoundError): + tts_app.refresh_voice_lists() diff --git a/tests/test_gui_settings.py b/tests/test_gui_settings.py new file mode 100644 index 0000000..fb6d20a --- /dev/null +++ b/tests/test_gui_settings.py @@ -0,0 +1,49 @@ +"""Tests for TTSApp.load_settings/save_settings/apply_settings +(gui.py:263-436).""" +import json + + +def test_load_settings_defaults_when_no_config_file(tts_app): + settings = tts_app.load_settings() + assert settings["voice"] == "af_heart" + assert settings["lexicon"] == {} + assert settings["caching"] is True + + +def test_load_settings_merges_existing_config_json(tts_app): + import gui + with open(gui.CONFIG_FILE, "w", encoding="utf-8") as f: + json.dump({"voice": "am_adam"}, f) + + settings = tts_app.load_settings() + + assert settings["voice"] == "am_adam" + assert settings["format"] == "wav" # untouched default still present + + +def test_load_settings_corrupt_json_falls_back_to_defaults(tts_app): + import gui + with open(gui.CONFIG_FILE, "w", encoding="utf-8") as f: + f.write("{not valid json") + + settings = tts_app.load_settings() + + assert settings["voice"] == "af_heart" + + +def test_save_settings_writes_json_with_current_vars(tts_app): + import gui + tts_app.voice_var.set("am_liam") + tts_app.save_settings() + + with open(gui.CONFIG_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + assert data["voice"] == "am_liam" + + +def test_change_appearance_and_scaling_persist_to_settings(tts_app): + tts_app.change_appearance("Light") + tts_app.change_scaling("120%") + + assert tts_app.settings["appearance"] == "Light" + assert tts_app.settings["scaling"] == "120%" diff --git a/tests/test_jit_conversion.py b/tests/test_jit_conversion.py new file mode 100644 index 0000000..10e4c72 --- /dev/null +++ b/tests/test_jit_conversion.py @@ -0,0 +1,38 @@ +"""Tests for _process_jit_async, the real-time/JIT pipeline +(kokoro_engine.py:747-908). caching=False throughout via make_config.""" +import asyncio + + +def test_jit_conversion_writes_combined_jit_output_file(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(filename="jitrun", time_id="1") + asyncio.run(engine._process_jit_async("Hello world. This is JIT.", config)) + + assert (isolated_dirs.out_dir / "jitrun_1_jit_output.wav").exists() + + +def test_jit_conversion_cancel_writes_remaining_txt(engine, fake_pipeline, make_config, isolated_dirs): + config = make_config(filename="jitrun", time_id="1") + engine.cancel_event.set() # cancel before starting -> deterministic "nothing played" branch + + asyncio.run(engine._process_jit_async("Hello world. This is JIT playback text.", config)) + + remaining = isolated_dirs.out_dir / "jitrun_1_remaining.txt" + assert remaining.exists() + content = remaining.read_text(encoding="utf-8") + assert "Hello world" in content + + assert not (isolated_dirs.out_dir / "jitrun_1_jit_output.wav").exists() + + +def test_jit_conversion_calls_on_finish_even_when_no_text(engine, fake_pipeline, make_config, callback_recorder): + config = make_config() + asyncio.run(engine._process_jit_async(" ", config)) + + assert callback_recorder.finished.is_set() + + +def test_jit_conversion_leaves_inspectable_output(engine, fake_pipeline, make_config, timestamped_output_dir): + config = make_config(out_dir=str(timestamped_output_dir), filename="jitsmoke", time_id="1") + asyncio.run(engine._process_jit_async("Hello from the JIT smoke test.", config)) + + assert (timestamped_output_dir / "jitsmoke_1_jit_output.wav").exists() diff --git a/tests/test_lexicon.py b/tests/test_lexicon.py new file mode 100644 index 0000000..a1a8920 --- /dev/null +++ b/tests/test_lexicon.py @@ -0,0 +1,35 @@ +"""Tests for KokoroEngine.apply_lexicon (kokoro_engine.py:87-108).""" + + +def test_apply_lexicon_case_insensitive_replace(engine): + result = engine.apply_lexicon("Hello WORLD", {"world": "planet"}) + assert result == "Hello planet" + + +def test_apply_lexicon_empty_dict_returns_unchanged(engine): + assert engine.apply_lexicon("Hello world", {}) == "Hello world" + + +def test_apply_lexicon_falsy_input_returns_text_as_is(engine): + assert engine.apply_lexicon("Hello world", None) == "Hello world" + + +def test_apply_lexicon_skips_falsy_keys(engine): + result = engine.apply_lexicon("Hello world", {"": "ignored", "world": "planet"}) + assert result == "Hello planet" + + +def test_apply_lexicon_caches_compiled_regex(engine): + lexicon = {"hello": "hi"} + engine.apply_lexicon("hello there", lexicon) + pattern1 = engine._lexicon_cache["hello"] + + engine.apply_lexicon("hello again", lexicon) + pattern2 = engine._lexicon_cache["hello"] + + assert pattern1 is pattern2 + + +def test_apply_lexicon_multiple_rules_applied(engine): + result = engine.apply_lexicon("The cat sat on the mat", {"cat": "dog", "mat": "rug"}) + assert result == "The dog sat on the rug" diff --git a/tests/test_meta_caching_policy.py b/tests/test_meta_caching_policy.py new file mode 100644 index 0000000..f2afd3c --- /dev/null +++ b/tests/test_meta_caching_policy.py @@ -0,0 +1,19 @@ +"""Static guard enforcing that caching=True only ever appears in +tests/test_caching.py - see tests/conftest.py's make_config docstring.""" +import re +from pathlib import Path + + +def test_no_stray_caching_true_outside_test_caching_module(): + root = Path(__file__).parent + allowed = {"test_caching.py", "test_meta_caching_policy.py"} + offenders = [] + + for f in sorted(root.glob("test_*.py")): + if f.name in allowed: + continue + content = f.read_text(encoding="utf-8") + if re.search(r"caching['\"]?\s*[:=]\s*True", content): + offenders.append(f.name) + + assert not offenders, f"caching=True found outside test_caching.py: {offenders}" diff --git a/tests/test_mix_voices.py b/tests/test_mix_voices.py new file mode 100644 index 0000000..f143099 --- /dev/null +++ b/tests/test_mix_voices.py @@ -0,0 +1,52 @@ +"""Tests for KokoroEngine.mix_voices (kokoro_engine.py:280-337).""" +import asyncio +import os + +import pytest +import torch + +import kokoro_engine + + +@pytest.mark.parametrize("op", ["mix", "add", "subtract", "multiply", "divide"]) +def test_mix_voices_op_produces_saved_tensor(engine, fake_pipeline, isolated_dirs, op): + engine.pipeline = fake_pipeline + success, path, tensor = asyncio.run(engine.mix_voices("af_heart", "af_bella", 0.5, "myvoice", op=op)) + + assert success is True + assert os.path.exists(path) + loaded = torch.load(path) + assert torch.equal(loaded, tensor) + + +def test_mix_voices_missing_voice_returns_error_tuple(engine, fake_pipeline, monkeypatch): + engine.pipeline = fake_pipeline + monkeypatch.setattr(fake_pipeline, "load_voice", lambda name: None) + + success, msg, tensor = asyncio.run(engine.mix_voices("af_heart", "af_bella", 0.5, "myvoice")) + + assert success is False + assert tensor is None + assert isinstance(msg, str) and msg + + +def test_mix_voices_name_basename_sanitized(engine, fake_pipeline, isolated_dirs): + engine.pipeline = fake_pipeline + success, path, _ = asyncio.run(engine.mix_voices("af_heart", "af_bella", 0.5, "../../evil")) + + assert success is True + assert os.path.dirname(path) == str(isolated_dirs.custom_voices) + assert os.path.basename(path) == "evil.pt" + + +def test_mix_voices_uses_existing_pipeline_before_thread_local_fallback(engine, fake_pipeline, monkeypatch): + engine.pipeline = fake_pipeline + + def _boom(lang_code="a"): + raise AssertionError("get_thread_pipeline should not be used when engine.pipeline is already set") + + monkeypatch.setattr(kokoro_engine, "get_thread_pipeline", _boom) + + success, path, _ = asyncio.run(engine.mix_voices("af_heart", "af_bella", 0.5, "myvoice")) + + assert success is True diff --git a/tests/test_presets.py b/tests/test_presets.py new file mode 100644 index 0000000..4619734 --- /dev/null +++ b/tests/test_presets.py @@ -0,0 +1,64 @@ +"""Tests for the engine-level load_preset/load_fx_preset (kokoro_engine.py:491-515). + +These read from hardcoded relative paths ("presets/...", "presets/fx/...") +rather than a module constant, so isolation here uses monkeypatch.chdir +instead of the isolated_dirs fixture. +""" +import json + + +def test_load_preset_reads_json(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "presets").mkdir() + (tmp_path / "presets" / "MyPreset.json").write_text(json.dumps({"voice": "af_heart"}), encoding="utf-8") + + assert engine.load_preset("MyPreset") == {"voice": "af_heart"} + + +def test_load_preset_missing_returns_none(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "presets").mkdir() + + assert engine.load_preset("DoesNotExist") is None + + +def test_load_preset_malformed_json_returns_none(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "presets").mkdir() + (tmp_path / "presets" / "Bad.json").write_text("{not valid json", encoding="utf-8") + + assert engine.load_preset("Bad") is None + + +def test_load_preset_path_traversal_sanitized(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "presets").mkdir() + (tmp_path / "presets" / "secret.json").write_text(json.dumps({"voice": "x"}), encoding="utf-8") + + # os.path.basename() strips any path components before the lookup. + assert engine.load_preset("../../secret") == {"voice": "x"} + + +def test_load_fx_preset_reads_json(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + fx_dir = tmp_path / "presets" / "fx" + fx_dir.mkdir(parents=True) + (fx_dir / "MyFx.json").write_text(json.dumps({"reverb_enabled": True}), encoding="utf-8") + + assert engine.load_fx_preset("MyFx") == {"reverb_enabled": True} + + +def test_load_fx_preset_missing_returns_none(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + (tmp_path / "presets" / "fx").mkdir(parents=True) + + assert engine.load_fx_preset("Nope") is None + + +def test_load_fx_preset_path_traversal_sanitized(engine, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + fx_dir = tmp_path / "presets" / "fx" + fx_dir.mkdir(parents=True) + (fx_dir / "s.json").write_text(json.dumps({"gain_db": 3.0}), encoding="utf-8") + + assert engine.load_fx_preset("../../s") == {"gain_db": 3.0} diff --git a/tests/test_process_audio.py b/tests/test_process_audio.py new file mode 100644 index 0000000..58ab2d0 --- /dev/null +++ b/tests/test_process_audio.py @@ -0,0 +1,92 @@ +"""Tests for KokoroEngine.process_audio - the DSP chain +(kokoro_engine.py:123-262): trim -> volume -> pitch -> Pedalboard FX -> normalize. + +Uses synthetic sine waves; assertions are on shape/length/non-silence/ +toggle-changes-output, not perceptual audio correctness (out of scope). +""" +import numpy as np +import pytest + + +def _sine(duration_s=1.0, sr=24000, freq=220.0, amp=0.3): + n = int(sr * duration_s) + t = np.arange(n) / sr + return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float64) + + +def test_process_audio_noop_config_returns_similar_length(engine): + audio = _sine() + out = engine.process_audio(audio.copy(), 24000, {}) + assert abs(len(out) - len(audio)) <= 2 + + +def test_trim_silence_removes_leading_trailing_silence(engine): + sr = 24000 + silence = np.zeros(int(sr * 0.2)) + tone = _sine(duration_s=0.5, amp=0.5) + audio = np.concatenate([silence, tone, silence]) + + trimmed = engine.process_audio(audio.copy(), sr, {"trim_silence": True}) + untrimmed = engine.process_audio(audio.copy(), sr, {"trim_silence": False}) + + assert len(trimmed) < len(untrimmed) + + +def test_volume_scales_amplitude(engine): + audio = _sine(amp=0.4) + out = engine.process_audio(audio.copy(), 24000, {"volume": 0.5}) + assert np.allclose(out, audio * 0.5) + + +def test_pitch_shift_changes_length_via_resample(engine): + audio = _sine(duration_s=1.0) + out = engine.process_audio(audio.copy(), 24000, {"pitch": 12.0}) + expected_len = int(len(audio) / 2.0) + assert abs(len(out) - expected_len) <= 2 + assert len(out) != len(audio) + + +def test_normalize_peaks_near_unity(engine): + audio = _sine(amp=0.05) + out = engine.process_audio(audio.copy(), 24000, {"normalize": True}) + assert np.max(np.abs(out)) > 0.9 + + +def test_eq_bass_treble_shelf_filters_change_output(engine): + audio = _sine() + flat = engine.process_audio(audio.copy(), 24000, {}) + shaped = engine.process_audio(audio.copy(), 24000, {"eq_bass": 6.0, "eq_treble": -6.0}) + assert shaped.shape != flat.shape or not np.allclose(shaped, flat) + + +FX_TOGGLE_CASES = [ + ("reverb_enabled", {}), + ("comp_enabled", {}), + ("distortion_enabled", {}), + ("chorus_enabled", {}), + ("phaser_enabled", {}), + ("clipping_enabled", {}), + ("bitcrush_enabled", {}), + ("gsm_enabled", {}), + ("highpass_enabled", {}), + ("lowpass_enabled", {}), + ("pitch_shift_enabled", {"pitch_shift_semitones": 5.0}), + ("delay_enabled", {}), + ("limiter_enabled", {}), + ("gain_enabled", {"gain_db": 6.0}), +] + + +@pytest.mark.parametrize("fx_key,overrides", FX_TOGGLE_CASES, ids=[c[0] for c in FX_TOGGLE_CASES]) +def test_fx_toggle_changes_output(engine, fx_key, overrides): + # Loud enough that level-dependent FX (compressor/limiter/clipping/gain) + # actually have something to act on. + audio = _sine(amp=0.9) + disabled_config = dict(overrides) + enabled_config = dict(overrides) + enabled_config[fx_key] = True + + out_disabled = engine.process_audio(audio.copy(), 24000, disabled_config) + out_enabled = engine.process_audio(audio.copy(), 24000, enabled_config) + + assert out_enabled.shape != out_disabled.shape or not np.allclose(out_enabled, out_disabled) diff --git a/tests/test_resolve_voice_path.py b/tests/test_resolve_voice_path.py new file mode 100644 index 0000000..6160519 --- /dev/null +++ b/tests/test_resolve_voice_path.py @@ -0,0 +1,27 @@ +"""Tests for KokoroEngine.resolve_voice_path (kokoro_engine.py:110-121).""" +import os + +import torch + + +def test_standard_voice_name_passthrough(engine): + assert engine.resolve_voice_path("af_heart") == "af_heart" + + +def test_custom_voice_resolves_to_abspath(engine, isolated_dirs): + torch.save(torch.zeros(1), str(isolated_dirs.custom_voices / "foo.pt")) + + resolved = engine.resolve_voice_path("foo") + + assert os.path.isabs(resolved) + assert os.path.exists(resolved) + assert os.path.dirname(resolved) == str(isolated_dirs.custom_voices) + + +def test_path_traversal_sanitized(engine, isolated_dirs): + torch.save(torch.zeros(1), str(isolated_dirs.custom_voices / "secrets.pt")) + + resolved = engine.resolve_voice_path("../../secrets") + + assert os.path.exists(resolved) + assert os.path.dirname(resolved) == str(isolated_dirs.custom_voices) diff --git a/tests/test_text_processing.py b/tests/test_text_processing.py new file mode 100644 index 0000000..20dd57b --- /dev/null +++ b/tests/test_text_processing.py @@ -0,0 +1,116 @@ +"""Tests for parse_multispeaker_text, smart_split, extract_text_from_file +(kokoro_engine.py:459-489, 517-543, 432-457).""" +import pytest + +import kokoro_engine + + +# --- parse_multispeaker_text --- + +def test_parse_multispeaker_no_markers_returns_single_none_tuple(engine): + result = engine.parse_multispeaker_text("Just plain text.") + assert result == [(None, None, "Just plain text.")] + + +def test_parse_multispeaker_single_speaker_marker(engine): + result = engine.parse_multispeaker_text("[Narrator]: Hello there.") + assert result == [("Narrator", None, "Hello there.")] + + +def test_parse_multispeaker_speaker_and_fx_marker(engine): + result = engine.parse_multispeaker_text("[Narrator:Radio]: Hello there.") + assert result == [("Narrator", "Radio", "Hello there.")] + + +def test_parse_multispeaker_multiple_segments(engine): + result = engine.parse_multispeaker_text("[A]: first\n\n[B]: second") + assert result == [("A", None, "first"), ("B", None, "second")] + + +def test_parse_multispeaker_marker_regex_length_limit(engine): + # The marker regex caps bracket contents at 100 chars; longer bracket + # contents should not match as a marker at all (kokoro_engine.py:466). + long_name = "A" * 150 + text = f"[{long_name}]: hello" + result = engine.parse_multispeaker_text(text) + assert result == [(None, None, text)] + + +def test_parse_multispeaker_empty_segment_is_skipped(engine): + result = engine.parse_multispeaker_text("[A]: \n\n[B]: real text") + assert result == [("B", None, "real text")] + + +# --- smart_split --- + +def test_smart_split_splits_on_paragraph_boundaries(engine): + text = "para one" + "\n\n" + ("x" * 20) + chunks = engine.smart_split(text, chunk_size=15) + assert len(chunks) == 2 + + +def test_smart_split_respects_chunk_size_budget(engine): + para = "y" * 50 + text = "\n\n".join([para] * 5) + chunks = engine.smart_split(text, chunk_size=60) + assert len(chunks) > 1 + for c in chunks: + assert len(c) <= 60 + + +def test_smart_split_single_short_text_returns_one_chunk(engine): + assert engine.smart_split("short text", chunk_size=3000) == ["short text"] + + +def test_smart_split_filters_whitespace_only_chunks(engine): + assert engine.smart_split(" ", chunk_size=3000) == [] + + +# --- extract_text_from_file --- + +def test_extract_text_from_file_txt(engine, tmp_path): + p = tmp_path / "sample.txt" + p.write_text("Hello file.", encoding="utf-8") + assert engine.extract_text_from_file(str(p)) == "Hello file." + + +def test_extract_text_from_file_missing_raises(engine, tmp_path): + with pytest.raises(FileNotFoundError): + engine.extract_text_from_file(str(tmp_path / "nope.txt")) + + +def test_extract_text_from_file_pdf(engine, tmp_path, monkeypatch): + class FakePage: + def extract_text(self): + return "Page text." + + class FakeReader: + def __init__(self, path): + self.pages = [FakePage(), FakePage()] + + monkeypatch.setattr(kokoro_engine.pypdf, "PdfReader", FakeReader) + p = tmp_path / "sample.pdf" + p.write_bytes(b"%PDF-fake") + + text = engine.extract_text_from_file(str(p)) + assert text.count("Page text.") == 2 + + +def test_extract_text_from_file_epub(engine, tmp_path, monkeypatch): + class FakeItem: + def get_type(self): + return kokoro_engine.ebooklib.ITEM_DOCUMENT + + def get_content(self): + return b"

Chapter text.

" + + class FakeBook: + def get_items(self): + return [FakeItem()] + + monkeypatch.setattr(kokoro_engine.epub, "read_epub", lambda path, options=None: FakeBook()) + p = tmp_path / "sample.epub" + p.write_bytes(b"fake-epub") + + text = engine.extract_text_from_file(str(p)) + assert "Chapter text." in text From f2f84ba29161d8b2da5c409688466437e93ce04c Mon Sep 17 00:00:00 2001 From: CoffeeMethod Date: Fri, 21 Aug 2026 17:58:51 -0600 Subject: [PATCH 3/3] added github tests.yml --- .github/workflows/tests.yml | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) create mode 100644 .github/workflows/tests.yml diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..e643f64 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,31 @@ +name: Tests + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + # kokoro_engine.py imports the Windows-only `winsound` module + # unconditionally, so the suite can only run on Windows. + runs-on: windows-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + + - name: Install dependencies + run: | + pip install -r requirements.txt + pip install -r requirements-test.txt + + - name: Run fast test suite + run: pytest + # Runs the mocked-pipeline suite only (pytest.ini already sets + # `-m "not integration"` by default). No eSpeak NG or model + # download needed. The real-synthesis integration suite + # (`pytest -m integration tests/integration`) is intentionally + # left out of CI - it's slow and pulls model weights.