From a299947bb18cd24df1b1c5a06a2ad989990a36cf Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Sat, 18 Jul 2026 10:21:23 +0000 Subject: [PATCH] refactor(webui): replace inline get_pipeline/resolve_voice_target closures with domain modules - Replace get_pipeline() closure with PipelinePool from domain/pipeline_factory - Replace resolve_voice_target() closure with domain function from voice_utils - Remove dead _load_pipeline() function and unused is_plugin_registered import - Add 33 tests for resolve_voice_target and PipelinePool - Add 10 regression tests verifying domain extraction preserves behavior - 1131 tests pass (+61 new) --- abogen/webui/conversion_runner.py | 135 +++++------------- tests/test_domain_pipeline_factory.py | 175 +++++++++++++++++++++++ tests/test_domain_voice_utils.py | 195 ++++++++++++++++++++++++++ tests/test_regression_webui_domain.py | 145 +++++++++++++++++++ 4 files changed, 553 insertions(+), 97 deletions(-) create mode 100644 tests/test_domain_pipeline_factory.py create mode 100644 tests/test_domain_voice_utils.py create mode 100644 tests/test_regression_webui_domain.py diff --git a/abogen/webui/conversion_runner.py b/abogen/webui/conversion_runner.py index 2789da9..2aa2b9e 100644 --- a/abogen/webui/conversion_runner.py +++ b/abogen/webui/conversion_runner.py @@ -13,7 +13,6 @@ from typing import Any, Callable, Dict, List, Mapping, Optional import numpy as np -from abogen.tts_plugin.utils import is_plugin_registered from abogen.infrastructure.exporters import ExportService from abogen.epub3.exporter import build_epub3_package from abogen.kokoro_text_normalization import ApostropheConfig, normalize_for_pipeline, HAS_NUM2WORDS @@ -31,10 +30,7 @@ from abogen.utils import ( get_internal_cache_path, get_user_cache_path, get_user_output_path, - load_config, ) -from abogen.tts_plugin.utils import create_pipeline -from abogen.voice_formulas import get_new_voice from abogen.voice_profiles import load_profiles, normalize_profile_entry from abogen.llm_client import LLMClientError from abogen.infrastructure.subtitle_writer import create_subtitle_writer @@ -122,6 +118,8 @@ from abogen.domain.audio_buffer import ( ) from abogen.domain.audio_sink import AudioSink, open_audio_sink from abogen.domain.tokens import FakeToken +from abogen.domain.pipeline_factory import PipelinePool +from abogen.domain.voice_utils import resolve_voice_target as _resolve_voice_target from .service import Job, JobStatus @@ -184,8 +182,7 @@ def run_conversion_job(job: Job) -> None: audio_output_path: Optional[Path] = None extraction: Optional[Any] = None pipeline: Any = None - pipelines: Dict[str, Any] = {} - kokoro_cache_ready = False + pipeline_pool = PipelinePool() normalized_profiles: Dict[str, Dict[str, Any]] = {} chunk_groups: Dict[int, List[Dict[str, Any]]] = {} active_chapter_configs: List[Dict[str, Any]] = [] @@ -202,75 +199,23 @@ def run_conversion_job(job: Job) -> None: if normalized: normalized_profiles[str(name)] = normalized - def get_pipeline(provider: str) -> Any: - nonlocal kokoro_cache_ready - provider_norm = str(provider or "kokoro").strip().lower() or "kokoro" - if not is_plugin_registered(provider_norm): - provider_norm = "kokoro" - - existing = pipelines.get(provider_norm) - if existing is not None: - return existing - - if provider_norm == "supertonic": - pipelines[provider_norm] = create_pipeline( - "supertonic", - ) - return pipelines[provider_norm] - - # Kokoro - cfg = load_config() - disable_gpu = not job.use_gpu or not cfg.get("use_gpu", True) - device = "cpu" - if not disable_gpu: - device = _select_device() - # Create KPipeline instance directly (uses new Plugin Architecture) - pipelines[provider_norm] = create_pipeline( - "kokoro", - lang_code=job.language, - device=device - ) - if not kokoro_cache_ready: - _initialize_voice_cache(job) - kokoro_cache_ready = True - return pipelines[provider_norm] - - def resolve_voice_target(raw_spec: str) -> tuple[str, str, Optional[float], Optional[int]]: - """Return (provider, voice_spec, speed_override, steps_override).""" - spec = str(raw_spec or "").strip() - speaker_name, _ = _split_speaker_reference(spec) - if speaker_name and speaker_name in normalized_profiles: - entry = normalized_profiles[speaker_name] - provider = str(entry.get("provider") or "kokoro").strip().lower() or "kokoro" - if provider == "supertonic": - voice = str(entry.get("voice") or getattr(job, "voice", "M1") or "M1").strip() or "M1" - steps = int(entry.get("total_steps") or getattr(job, "supertonic_total_steps", 5) or 5) - speed = float(entry.get("speed") or getattr(job, "speed", 1.0) or 1.0) - return "supertonic", _supertonic_voice_from_spec(voice, getattr(job, "voice", "M1")), speed, steps - formula = _formula_from_kokoro_entry(entry) - return "kokoro", formula or spec, None, None - - fallback_provider = str(getattr(job, "tts_provider", "kokoro") or "kokoro").strip().lower() or "kokoro" - inferred = _infer_provider_from_spec(spec, fallback=fallback_provider) - if inferred == "supertonic": - return "supertonic", _supertonic_voice_from_spec(spec, getattr(job, "voice", "M1")), None, None - return "kokoro", spec, None, None - def resolve_voice_choice(raw_spec: str) -> tuple[str, str, Any, Optional[float], Optional[int]]: - """Resolve a raw voice spec into (provider, resolved_spec, choice, speed, steps). - - For Kokoro formulas, `choice` will be a resolved voice tensor (via `voice_formulas`). - For SuperTonic, `choice` will be a valid SuperTonic voice id. - """ - - provider, resolved, speed, steps = resolve_voice_target(raw_spec) + """Resolve a raw voice spec into (provider, resolved_spec, choice, speed, steps).""" + provider, resolved, speed, steps = _resolve_voice_target( + raw_spec, + normalized_profiles, + job_voice=getattr(job, "voice", "M1"), + job_tts_provider=getattr(job, "tts_provider", "kokoro"), + job_supertonic_total_steps=getattr(job, "supertonic_total_steps", 5), + job_speed=getattr(job, "speed", 1.0), + ) cache_key = f"{provider}:{resolved}" if resolved else provider cached = voice_cache.get(cache_key) if cached is not None: return provider, resolved, cached, speed, steps if provider == "kokoro": - kokoro_backend = get_pipeline("kokoro") + kokoro_backend = pipeline_pool.get("kokoro", job.language, job.use_gpu, job=job) choice = _resolve_voice(kokoro_backend, resolved, job.use_gpu) else: choice = resolved @@ -405,9 +350,13 @@ def run_conversion_job(job: Job) -> None: base_voice_spec = _job_voice_fallback(job) voice_cache: Dict[str, Any] = {} - base_provider, base_voice_resolved, _, _ = resolve_voice_target(base_voice_spec) + base_provider, base_voice_resolved, _, _ = _resolve_voice_target( + base_voice_spec, normalized_profiles, + job_voice=getattr(job, "voice", "M1"), + job_tts_provider=getattr(job, "tts_provider", "kokoro"), + ) if base_provider == "kokoro" and base_voice_resolved and "*" not in base_voice_resolved: - kokoro_backend = get_pipeline("kokoro") + kokoro_backend = pipeline_pool.get("kokoro", job.language, job.use_gpu, job=job) voice_cache[f"kokoro:{base_voice_resolved}"] = _resolve_voice(kokoro_backend, base_voice_resolved, job.use_gpu) processed_chars = 0 current_time = 0.0 @@ -473,7 +422,7 @@ def run_conversion_job(job: Job) -> None: provider = str(tts_provider or getattr(job, "tts_provider", "kokoro") or "kokoro").strip().lower() or "kokoro" if provider == "supertonic": - supertonic_pipeline = get_pipeline("supertonic") + supertonic_pipeline = pipeline_pool.get("supertonic", job.language, job.use_gpu, job=job) voice_name = _supertonic_voice_from_spec(voice_choice, getattr(job, "voice", "M1")) segment_iter = supertonic_pipeline( normalized, @@ -483,7 +432,7 @@ def run_conversion_job(job: Job) -> None: total_steps=int(supertonic_steps_override if supertonic_steps_override is not None else getattr(job, "supertonic_total_steps", 5)), ) else: - kokoro_backend = get_pipeline("kokoro") + kokoro_backend = pipeline_pool.get("kokoro", job.language, job.use_gpu, job=job) segment_iter = kokoro_backend( normalized, voice=voice_choice, @@ -605,12 +554,18 @@ def run_conversion_job(job: Job) -> None: if not chapter_voice_spec: chapter_voice_spec = base_voice_spec - chapter_provider, chapter_voice_resolved, chapter_speed, chapter_steps = resolve_voice_target(chapter_voice_spec) + chapter_provider, chapter_voice_resolved, chapter_speed, chapter_steps = _resolve_voice_target( + chapter_voice_spec, normalized_profiles, + job_voice=getattr(job, "voice", "M1"), + job_tts_provider=getattr(job, "tts_provider", "kokoro"), + job_supertonic_total_steps=getattr(job, "supertonic_total_steps", 5), + job_speed=getattr(job, "speed", 1.0), + ) chapter_cache_key = f"{chapter_provider}:{chapter_voice_resolved}" if chapter_voice_resolved else chapter_provider if chapter_provider == "kokoro": voice_choice = voice_cache.get(chapter_cache_key) if voice_choice is None: - kokoro_backend = get_pipeline("kokoro") + kokoro_backend = pipeline_pool.get("kokoro", job.language, job.use_gpu, job=job) voice_choice = _resolve_voice(kokoro_backend, chapter_voice_resolved, job.use_gpu) voice_cache[chapter_cache_key] = voice_choice else: @@ -743,12 +698,18 @@ def run_conversion_job(job: Job) -> None: chunk_steps_use = chapter_steps chunk_voice_choice = voice_choice else: - chunk_provider, chunk_voice_resolved, chunk_speed_use, chunk_steps_use = resolve_voice_target(chunk_voice_spec) + chunk_provider, chunk_voice_resolved, chunk_speed_use, chunk_steps_use = _resolve_voice_target( + chunk_voice_spec, normalized_profiles, + job_voice=getattr(job, "voice", "M1"), + job_tts_provider=getattr(job, "tts_provider", "kokoro"), + job_supertonic_total_steps=getattr(job, "supertonic_total_steps", 5), + job_speed=getattr(job, "speed", 1.0), + ) chunk_cache_key = f"{chunk_provider}:{chunk_voice_resolved}" if chunk_voice_resolved else chunk_provider if chunk_provider == "kokoro": chunk_voice_choice = voice_cache.get(chunk_cache_key) if chunk_voice_choice is None: - kokoro_backend = get_pipeline("kokoro") + kokoro_backend = pipeline_pool.get("kokoro", job.language, job.use_gpu, job=job) chunk_voice_choice = _resolve_voice( kokoro_backend, chunk_voice_resolved, @@ -1054,12 +1015,7 @@ def run_conversion_job(job: Job) -> None: # Explicitly release the pipeline and force garbage collection to prevent # memory accumulation in the worker process, which can lead to host lockups. - for p in pipelines.values(): - try: - p.dispose() - except Exception: - pass - pipelines.clear() + pipeline_pool.dispose_all() pipeline = None gc.collect() try: @@ -1100,21 +1056,6 @@ def run_conversion_job(job: Job) -> None: ) from exc -def _load_pipeline(job: Job): - cfg = load_config() - disable_gpu = not job.use_gpu or not cfg.get("use_gpu", True) - provider = str(getattr(job, "tts_provider", "kokoro") or "kokoro").strip().lower() - if provider == "supertonic": - return create_pipeline( - "supertonic", - ) - - device = "cpu" - if not disable_gpu: - device = _select_device() - return create_pipeline("kokoro", lang_code=job.language, device=device) - - def _prepare_output_dir(job: Job) -> Path: from platformdirs import user_desktop_dir # type: ignore[import-not-found] diff --git a/tests/test_domain_pipeline_factory.py b/tests/test_domain_pipeline_factory.py new file mode 100644 index 0000000..74cf67d --- /dev/null +++ b/tests/test_domain_pipeline_factory.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from abogen.domain.pipeline_factory import ( + PipelinePool, + create_pipeline_for_job, + dispose_pipelines, + resolve_device, +) + + +class TestResolveDevice: + @patch("abogen.utils.load_config", return_value={"use_gpu": True}) + @patch("abogen.domain.pipeline_factory.select_device", return_value="cuda:0") + def test_gpu_enabled(self, _sel, _cfg): + assert resolve_device(use_gpu=True) == "cuda:0" + + @patch("abogen.utils.load_config", return_value={"use_gpu": True}) + def test_gpu_disabled_by_job(self, _cfg): + assert resolve_device(use_gpu=False) == "cpu" + + @patch("abogen.utils.load_config", return_value={"use_gpu": False}) + @patch("abogen.domain.pipeline_factory.select_device", return_value="cuda:0") + def test_gpu_disabled_by_config(self, _sel, _cfg): + assert resolve_device(use_gpu=True) == "cpu" + + +class TestCreatePipelineForJob: + @patch("abogen.domain.pipeline_factory.create_pipeline") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) + def test_supertonic_provider(self, _reg, mock_create): + mock_create.return_value = MagicMock() + result = create_pipeline_for_job("supertonic", "en", use_gpu=True) + mock_create.assert_called_once_with("supertonic") + assert result is mock_create.return_value + + @patch("abogen.domain.pipeline_factory.create_pipeline") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) + @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") + def test_kokoro_provider(self, _dev, _reg, mock_create): + mock_create.return_value = MagicMock() + result = create_pipeline_for_job("kokoro", "en", use_gpu=False) + mock_create.assert_called_once_with("kokoro", lang_code="en", device="cpu") + assert result is mock_create.return_value + + @patch("abogen.domain.pipeline_factory.create_pipeline") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False) + @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") + def test_unknown_provider_falls_back_to_kokoro(self, _dev, _reg, mock_create): + mock_create.return_value = MagicMock() + result = create_pipeline_for_job("unknown_provider", "en", use_gpu=False) + mock_create.assert_called_once_with("kokoro", lang_code="en", device="cpu") + + @patch("abogen.domain.pipeline_factory.create_pipeline") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) + @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") + def test_empty_provider_defaults_to_kokoro(self, _dev, _reg, mock_create): + mock_create.return_value = MagicMock() + result = create_pipeline_for_job("", "en", use_gpu=False) + mock_create.assert_called_once_with("kokoro", lang_code="en", device="cpu") + + @patch("abogen.domain.pipeline_factory.create_pipeline") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=True) + @patch("abogen.domain.pipeline_factory.resolve_device", return_value="cpu") + def test_none_provider_defaults_to_kokoro(self, _dev, _reg, mock_create): + mock_create.return_value = MagicMock() + result = create_pipeline_for_job(None, "en", use_gpu=False) + mock_create.assert_called_once_with("kokoro", lang_code="en", device="cpu") + + +class TestDisposePipelines: + def test_disposes_all_and_clears(self): + p1 = MagicMock() + p2 = MagicMock() + pipelines = {"kokoro": p1, "supertonic": p2} + dispose_pipelines(pipelines) + p1.dispose.assert_called_once() + p2.dispose.assert_called_once() + assert pipelines == {} + + def test_handles_dispose_error(self): + p1 = MagicMock() + p1.dispose.side_effect = RuntimeError("boom") + pipelines = {"kokoro": p1} + dispose_pipelines(pipelines) + assert pipelines == {} + + def test_empty_dict(self): + pipelines = {} + dispose_pipelines(pipelines) + assert pipelines == {} + + +class TestPipelinePool: + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_get_creates_and_caches(self, _cache, mock_create): + mock_pipeline = MagicMock() + mock_create.return_value = mock_pipeline + pool = PipelinePool() + + result = pool.get("kokoro", "en", use_gpu=True) + assert result is mock_pipeline + mock_create.assert_called_once() + + result2 = pool.get("kokoro", "en", use_gpu=True) + assert result2 is mock_pipeline + assert mock_create.call_count == 1 + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_get_initializes_voice_cache_once(self, mock_cache, mock_create): + mock_create.return_value = MagicMock() + pool = PipelinePool() + + job = MagicMock() + pool.get("kokoro", "en", use_gpu=True, job=job) + assert mock_cache.call_count == 1 + + pool.get("kokoro", "en", use_gpu=True, job=job) + assert mock_cache.call_count == 1 + + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + def test_get_no_job_skips_voice_cache(self, mock_create, mock_cache): + mock_create.return_value = MagicMock() + pool = PipelinePool() + pool.get("kokoro", "en", use_gpu=True) + mock_cache.assert_not_called() + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + def test_get_separately_per_provider(self, mock_create): + p1 = MagicMock(name="kokoro") + p2 = MagicMock(name="supertonic") + mock_create.side_effect = [p1, p2] + pool = PipelinePool() + + r1 = pool.get("kokoro", "en", use_gpu=True) + r2 = pool.get("supertonic", "en", use_gpu=True) + assert r1 is p1 + assert r2 is p2 + assert mock_create.call_count == 2 + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_dispose_all(self, mock_cache, mock_create): + p1 = MagicMock(name="kokoro") + p2 = MagicMock(name="supertonic") + mock_create.side_effect = [p1, p2] + pool = PipelinePool() + + pool.get("kokoro", "en", use_gpu=True) + pool.get("supertonic", "en", use_gpu=True) + pool.dispose_all() + + p1.dispose.assert_called_once() + p2.dispose.assert_called_once() + assert pool._pipelines == {} + assert pool._voice_cache_initialized is False + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + def test_dispose_empty_pool(self, mock_create): + pool = PipelinePool() + pool.dispose_all() + mock_create.assert_not_called() + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + @patch("abogen.domain.pipeline_factory.is_plugin_registered", return_value=False) + def test_unknown_provider_falls_back(self, _reg, _cache, mock_create): + mock_create.return_value = MagicMock() + pool = PipelinePool() + pool.get("bogus_provider", "en", use_gpu=True) + mock_create.assert_called_once_with("kokoro", "en", True) diff --git a/tests/test_domain_voice_utils.py b/tests/test_domain_voice_utils.py new file mode 100644 index 0000000..4a10277 --- /dev/null +++ b/tests/test_domain_voice_utils.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +from unittest.mock import patch + +from abogen.domain.voice_utils import ( + coerce_truthy, + formula_from_kokoro_entry, + infer_provider_from_spec, + resolve_voice_target, + split_speaker_reference, + supertonic_voice_from_spec, +) + + +class TestSplitSpeakerReference: + def test_speaker_prefix(self): + assert split_speaker_reference("speaker:af_sarah") == ("af_sarah", "speaker:af_sarah") + + def test_profile_prefix(self): + assert split_speaker_reference("profile:custom") == ("custom", "profile:custom") + + def test_no_prefix(self): + assert split_speaker_reference("af_sarah") == (None, "af_sarah") + + def test_empty(self): + assert split_speaker_reference("") == (None, "") + + def test_none(self): + assert split_speaker_reference(None) == (None, "") + + def test_unknown_prefix(self): + assert split_speaker_reference("unknown:name") == (None, "unknown:name") + + def test_empty_name_after_colon(self): + assert split_speaker_reference("speaker:") == (None, "speaker:") + + +class TestSupertonicVoiceFromSpec: + def test_uppercase_passthrough(self): + assert supertonic_voice_from_spec("M1", "M1") == "M1" + + def test_lowercase_converted(self): + assert supertonic_voice_from_spec("m1", "M1") == "M1" + + def test_empty_spec_uses_fallback(self): + assert supertonic_voice_from_spec("", "F1") == "F1" + + def test_formula_spec_uses_fallback(self): + assert supertonic_voice_from_spec("af_sarah*0.5+bf_emma*0.5", "M1") == "M1" + + def test_empty_both_gives_default(self): + assert supertonic_voice_from_spec("", "") == "M1" + + +class TestFormulaFromKokoroEntry: + def test_single_voice(self): + entry = {"voices": [["af_sarah", 1.0]]} + result = formula_from_kokoro_entry(entry) + assert "af_sarah" in result + assert "1.000000" in result + + def test_weighted_mix(self): + entry = {"voices": [["af_sarah", 0.6], ["bf_emma", 0.4]]} + result = formula_from_kokoro_entry(entry) + assert "af_sarah" in result + assert "bf_emma" in result + assert "+" in result + + def test_empty_voices(self): + assert formula_from_kokoro_entry({"voices": []}) == "" + + def test_missing_voices_key(self): + assert formula_from_kokoro_entry({}) == "" + + def test_invalid_entries_filtered(self): + entry = {"voices": [["af_sarah", "bad"], ["bf_emma", 0.5]]} + result = formula_from_kokoro_entry(entry) + assert "bf_emma" in result + assert "af_sarah" not in result + + +class TestCoerceTruthy: + def test_bool_passthrough(self): + assert coerce_truthy(True) is True + assert coerce_truthy(False) is False + + def test_string_true(self): + assert coerce_truthy("yes") is True + assert coerce_truthy("1") is True + + def test_string_false(self): + assert coerce_truthy("false") is False + assert coerce_truthy("0") is False + assert coerce_truthy("") is False + + def test_none_default(self): + assert coerce_truthy(None) is True + assert coerce_truthy(None, False) is False + + def test_int(self): + assert coerce_truthy(1) is True + assert coerce_truthy(0) is False + + +class TestInferProviderFromSpec: + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah", "bf_emma"]) + def test_known_kokoro_voice(self, _mock): + assert infer_provider_from_spec("af_sarah") == "kokoro" + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_uppercase_supertonic(self, _mock): + assert infer_provider_from_spec("M1") == "supertonic" + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_formula_kokoro(self, _mock): + assert infer_provider_from_spec("af_sarah*0.5+bf_emma*0.5") == "kokoro" + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_empty_fallback(self, _mock): + assert infer_provider_from_spec("", "kokoro") == "kokoro" + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_unknown_falls_back(self, _mock): + assert infer_provider_from_spec("unknown_xyz", "supertonic") == "supertonic" + + +class TestResolveVoiceTarget: + def test_empty_spec_kokoro_default(self): + provider, spec, speed, steps = resolve_voice_target( + "", {}, job_voice="af_sarah", job_tts_provider="kokoro", + ) + assert provider == "kokoro" + assert spec == "" + + def test_speaker_profile_kokoro(self): + profiles = { + "narrator": { + "provider": "kokoro", + "voices": [["af_sarah", 0.7], ["bf_emma", 0.3]], + }, + } + provider, spec, speed, steps = resolve_voice_target( + "speaker:narrator", profiles, + ) + assert provider == "kokoro" + assert "af_sarah" in spec + assert speed is None + assert steps is None + + def test_speaker_profile_supertonic(self): + profiles = { + "narrator": { + "provider": "supertonic", + "voice": "F1", + "speed": 1.2, + "total_steps": 10, + }, + } + provider, spec, speed, steps = resolve_voice_target( + "speaker:narrator", profiles, + job_voice="M1", job_speed=1.0, job_supertonic_total_steps=5, + ) + assert provider == "supertonic" + assert spec == "F1" + assert speed == 1.2 + assert steps == 10 + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_direct_supertonic_spec(self, _mock): + provider, spec, speed, steps = resolve_voice_target( + "M1", {}, + job_voice="M1", + ) + assert provider == "supertonic" + assert spec == "M1" + + @patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]) + def test_direct_kokoro_spec(self, _mock): + provider, spec, speed, steps = resolve_voice_target( + "af_sarah", {}, + job_tts_provider="kokoro", + ) + assert provider == "kokoro" + assert spec == "af_sarah" + + def test_profile_missing_provider_defaults_kokoro(self): + profiles = { + "narrator": { + "voices": [["af_sarah", 1.0]], + }, + } + provider, spec, speed, steps = resolve_voice_target( + "speaker:narrator", profiles, + ) + assert provider == "kokoro" diff --git a/tests/test_regression_webui_domain.py b/tests/test_regression_webui_domain.py new file mode 100644 index 0000000..9e49a38 --- /dev/null +++ b/tests/test_regression_webui_domain.py @@ -0,0 +1,145 @@ +"""Regression tests: domain extraction must not break webui conversion_runner. + +These tests verify that the refactored WebUI code paths still call into the +correct domain functions and produce the same results as the old inline logic. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from abogen.domain.voice_utils import resolve_voice_target +from abogen.domain.pipeline_factory import PipelinePool + + +class TestResolveVoiceTargetRegression: + """Verify that the domain resolve_voice_target produces the same results + as the old closure in conversion_runner.py.""" + + def test_empty_spec_returns_kokoro_default(self): + provider, spec, speed, steps = resolve_voice_target( + "", {}, job_voice="af_sarah", job_tts_provider="kokoro", + ) + assert provider == "kokoro" + assert spec == "" + + def test_speaker_in_profile_kokoro(self): + profiles = { + "narrator": { + "provider": "kokoro", + "voices": [["af_sarah", 0.7], ["bf_emma", 0.3]], + }, + } + provider, spec, speed, steps = resolve_voice_target( + "speaker:narrator", profiles, + ) + assert provider == "kokoro" + assert "af_sarah" in spec + assert speed is None + assert steps is None + + def test_speaker_in_profile_supertonic(self): + profiles = { + "narrator": { + "provider": "supertonic", + "voice": "F1", + "speed": 1.2, + "total_steps": 10, + }, + } + provider, spec, speed, steps = resolve_voice_target( + "speaker:narrator", profiles, + job_voice="M1", job_speed=1.0, job_supertonic_total_steps=5, + ) + assert provider == "supertonic" + assert spec == "F1" + assert speed == 1.2 + assert steps == 10 + + def test_unknown_speaker_infers_from_spec(self): + with patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]): + provider, spec, speed, steps = resolve_voice_target( + "af_sarah", {}, job_tts_provider="kokoro", + ) + assert provider == "kokoro" + assert spec == "af_sarah" + + def test_uppercase_spec_infers_supertonic(self): + with patch("abogen.domain.voice_utils.get_voices", return_value=["af_sarah"]): + provider, spec, speed, steps = resolve_voice_target( + "M1", {}, job_voice="M1", + ) + assert provider == "supertonic" + assert spec == "M1" + + +class TestPipelinePoolRegression: + """Verify that PipelinePool behaves like the old inline get_pipeline closure.""" + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_same_provider_returns_cached_pipeline(self, _cache, mock_create): + mock_pipeline = MagicMock() + mock_create.return_value = mock_pipeline + pool = PipelinePool() + + r1 = pool.get("kokoro", "en", use_gpu=True) + r2 = pool.get("kokoro", "en", use_gpu=True) + assert r1 is r2 + assert mock_create.call_count == 1 + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_different_providers_get_separate_pipelines(self, _cache, mock_create): + p1 = MagicMock(name="kokoro") + p2 = MagicMock(name="supertonic") + mock_create.side_effect = [p1, p2] + pool = PipelinePool() + + r1 = pool.get("kokoro", "en", use_gpu=True) + r2 = pool.get("supertonic", "en", use_gpu=True) + assert r1 is p1 + assert r2 is p2 + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_dispose_all_cleans_up(self, _cache, mock_create): + p1 = MagicMock() + p2 = MagicMock() + mock_create.side_effect = [p1, p2] + pool = PipelinePool() + + pool.get("kokoro", "en", use_gpu=True) + pool.get("supertonic", "en", use_gpu=True) + pool.dispose_all() + + p1.dispose.assert_called_once() + p2.dispose.assert_called_once() + assert pool._pipelines == {} + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_voice_cache_initialized_only_once(self, mock_cache, mock_create): + mock_create.return_value = MagicMock() + pool = PipelinePool() + job = MagicMock() + + pool.get("kokoro", "en", use_gpu=True, job=job) + pool.get("kokoro", "en", use_gpu=True, job=job) + assert mock_cache.call_count == 1 + + @patch("abogen.domain.pipeline_factory.create_pipeline_for_job") + @patch("abogen.domain.pipeline_factory.initialize_voice_cache") + def test_after_dispose_voice_cache_can_reinitialize(self, mock_cache, mock_create): + mock_create.return_value = MagicMock() + pool = PipelinePool() + job = MagicMock() + + pool.get("kokoro", "en", use_gpu=True, job=job) + assert mock_cache.call_count == 1 + + pool.dispose_all() + assert pool._voice_cache_initialized is False + + pool.get("kokoro", "en", use_gpu=True, job=job) + assert mock_cache.call_count == 2