From a05357bab9d5fa18afecc89f03bdc7e2a1160cab Mon Sep 17 00:00:00 2001 From: Artem Akymenko Date: Thu, 9 Jul 2026 14:02:13 +0000 Subject: [PATCH] feat: add PluginManager, compat adapter, and consumer migration - Add PluginManager singleton for plugin discovery and engine caching - Add CompatBackend adapter wrapping Engine/EngineSession into old create_backend() API - Update tts_plugin/__init__.py with public exports - Migrate preview.py and its test to use compat.create_backend - Add integration and plugin manager contract tests --- abogen/tts_plugin/__init__.py | 295 +++++------ abogen/tts_plugin/compat.py | 137 +++++ abogen/tts_plugin/plugin_manager.py | 153 ++++++ abogen/webui/routes/utils/preview.py | 468 +++++++++--------- tests/contracts/test_integration_pr5.py | 400 +++++++++++++++ .../contracts/test_plugin_manager_contract.py | 264 ++++++++++ .../test_preview_applies_manual_overrides.py | 120 ++--- 7 files changed, 1404 insertions(+), 433 deletions(-) create mode 100644 abogen/tts_plugin/compat.py create mode 100644 abogen/tts_plugin/plugin_manager.py create mode 100644 tests/contracts/test_integration_pr5.py create mode 100644 tests/contracts/test_plugin_manager_contract.py diff --git a/abogen/tts_plugin/__init__.py b/abogen/tts_plugin/__init__.py index 051f71e..10c9154 100644 --- a/abogen/tts_plugin/__init__.py +++ b/abogen/tts_plugin/__init__.py @@ -1,139 +1,156 @@ -"""TTS Plugin Architecture - Public API. - -This package defines the frozen Plugin API for the TTS Plugin Architecture. -All public interfaces are fully defined but contain no business logic. - -Public modules: -- types: Core domain value objects (AudioFormat, Duration, VoiceSelection, etc.) -- errors: Error hierarchy (EngineError and subtypes) -- manifest: Plugin manifest types (PluginManifest, EngineManifest, etc.) -- engine: Engine and EngineSession protocols -- capabilities: Optional capability interfaces (VoiceLister, PreviewGenerator, etc.) -- host_context: HostContext dataclass -- plugin: Plugin contract (create_engine function signature) - -Usage: - from abogen.tts_plugin import ( - # Types - AudioFormat, - Duration, - VoiceSelection, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - EngineConfig, - # Errors - EngineError, - ModelNotFoundError, - ModelLoadError, - NetworkError, - InvalidInputError, - ConfigurationError, - CancelledError, - InternalError, - # Manifest - PluginManifest, - EngineManifest, - VoiceSourceManifest, - VoiceManifest, - ParameterManifest, - AudioFormatManifest, - EnumOption, - RequirementManifest, - GpuRequirement, - ModelManifest, - # Engine - Engine, - EngineSession, - # Capabilities - VoiceLister, - PreviewGenerator, - StreamingSynthesizer, - CancelableSession, - # Host Context - HostContext, - HttpClient, - ) -""" - -from abogen.tts_plugin.capabilities import ( - CancelableSession, - PreviewGenerator, - StreamingSynthesizer, - VoiceLister, -) -from abogen.tts_plugin.engine import Engine, EngineSession -from abogen.tts_plugin.errors import ( - CancelledError, - ConfigurationError, - EngineError, - InternalError, - InvalidInputError, - ModelLoadError, - ModelNotFoundError, - NetworkError, -) -from abogen.tts_plugin.host_context import HttpClient, HostContext -from abogen.tts_plugin.manifest import ( - AudioFormatManifest, - EngineManifest, - EnumOption, - GpuRequirement, - ModelManifest, - ParameterManifest, - PluginManifest, - RequirementManifest, - VoiceManifest, - VoiceSourceManifest, -) -from abogen.tts_plugin.types import ( - AudioFormat, - Duration, - EngineConfig, - ParameterValues, - SynthesisRequest, - SynthesizedAudio, - VoiceSelection, -) - -__all__ = [ - # Types - "AudioFormat", - "Duration", - "VoiceSelection", - "ParameterValues", - "SynthesisRequest", - "SynthesizedAudio", - "EngineConfig", - # Errors - "EngineError", - "ModelNotFoundError", - "ModelLoadError", - "NetworkError", - "InvalidInputError", - "ConfigurationError", - "CancelledError", - "InternalError", - # Manifest - "PluginManifest", - "EngineManifest", - "VoiceSourceManifest", - "VoiceManifest", - "ParameterManifest", - "AudioFormatManifest", - "EnumOption", - "RequirementManifest", - "GpuRequirement", - "ModelManifest", - # Engine - "Engine", - "EngineSession", - # Capabilities - "VoiceLister", - "PreviewGenerator", - "StreamingSynthesizer", - "CancelableSession", - # Host Context - "HostContext", - "HttpClient", -] +"""TTS Plugin Architecture - Public API. + +This package defines the frozen Plugin API for the TTS Plugin Architecture. +All public interfaces are fully defined but contain no business logic. + +Public modules: +- types: Core domain value objects (AudioFormat, Duration, VoiceSelection, etc.) +- errors: Error hierarchy (EngineError and subtypes) +- manifest: Plugin manifest types (PluginManifest, EngineManifest, etc.) +- engine: Engine and EngineSession protocols +- capabilities: Optional capability interfaces (VoiceLister, PreviewGenerator, etc.) +- host_context: HostContext dataclass +- plugin: Plugin contract (create_engine function signature) +- loader: Plugin discovery and loading +- plugin_manager: Plugin management and engine creation +- compat: Backward compatibility adapter for old create_backend() API + +Usage: + from abogen.tts_plugin import ( + # Types + AudioFormat, + Duration, + VoiceSelection, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + EngineConfig, + # Errors + EngineError, + ModelNotFoundError, + ModelLoadError, + NetworkError, + InvalidInputError, + ConfigurationError, + CancelledError, + InternalError, + # Manifest + PluginManifest, + EngineManifest, + VoiceSourceManifest, + VoiceManifest, + ParameterManifest, + AudioFormatManifest, + EnumOption, + RequirementManifest, + GpuRequirement, + ModelManifest, + # Engine + Engine, + EngineSession, + # Capabilities + VoiceLister, + PreviewGenerator, + StreamingSynthesizer, + CancelableSession, + # Host Context + HostContext, + HttpClient, + # Plugin Manager + get_plugin_manager, + reset_plugin_manager, + # Compatibility + create_backend, + ) +""" + +from abogen.tts_plugin.capabilities import ( + CancelableSession, + PreviewGenerator, + StreamingSynthesizer, + VoiceLister, +) +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import ( + CancelledError, + ConfigurationError, + EngineError, + InternalError, + InvalidInputError, + ModelLoadError, + ModelNotFoundError, + NetworkError, +) +from abogen.tts_plugin.host_context import HttpClient, HostContext +from abogen.tts_plugin.manifest import ( + AudioFormatManifest, + EngineManifest, + EnumOption, + GpuRequirement, + ModelManifest, + ParameterManifest, + PluginManifest, + RequirementManifest, + VoiceManifest, + VoiceSourceManifest, +) +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + EngineConfig, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + +# Plugin Manager and Compatibility +from abogen.tts_plugin.plugin_manager import get_plugin_manager, reset_plugin_manager +from abogen.tts_plugin.compat import create_backend + +__all__ = [ + # Types + "AudioFormat", + "Duration", + "VoiceSelection", + "ParameterValues", + "SynthesisRequest", + "SynthesizedAudio", + "EngineConfig", + # Errors + "EngineError", + "ModelNotFoundError", + "ModelLoadError", + "NetworkError", + "InvalidInputError", + "ConfigurationError", + "CancelledError", + "InternalError", + # Manifest + "PluginManifest", + "EngineManifest", + "VoiceSourceManifest", + "VoiceManifest", + "ParameterManifest", + "AudioFormatManifest", + "EnumOption", + "RequirementManifest", + "GpuRequirement", + "ModelManifest", + # Engine + "Engine", + "EngineSession", + # Capabilities + "VoiceLister", + "PreviewGenerator", + "StreamingSynthesizer", + "CancelableSession", + # Host Context + "HostContext", + "HttpClient", + # Plugin Manager + "get_plugin_manager", + "reset_plugin_manager", + # Compatibility + "create_backend", +] diff --git a/abogen/tts_plugin/compat.py b/abogen/tts_plugin/compat.py new file mode 100644 index 0000000..db4384d --- /dev/null +++ b/abogen/tts_plugin/compat.py @@ -0,0 +1,137 @@ +"""TTS Backend Compatibility Adapter + +Provides a drop-in replacement for the old `create_backend()` function +that uses the new Plugin Architecture under the hood. + +Usage: + # Old way: + from abogen.tts_backend_registry import create_backend + pipeline = create_backend("kokoro", lang_code="a", device="cpu") + + # New way (same interface): + from abogen.tts_plugin.compat import create_backend + pipeline = create_backend("kokoro", lang_code="a", device="cpu") + +The adapter wraps the new Engine/EngineSession into a callable that +matches the old TTSBackend protocol. +""" + +from typing import Any, Callable, Iterable, Iterator, List, Mapping, Optional, Tuple + +import numpy as np + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.plugin_manager import get_plugin_manager + + +class CompatBackend: + """Compatibility wrapper that makes a new Engine look like the old TTSBackend. + + This adapter wraps the new Engine/EngineSession into a callable that + matches the old Kokoro pipeline interface: + pipeline(text, voice=..., speed=..., split_pattern=...) -> Iterator[Segment] + """ + + def __init__(self, engine: Engine, **engine_kwargs: Any) -> None: + self._engine = engine + self._engine_kwargs = engine_kwargs + self._session: Optional[EngineSession] = None + + def _ensure_session(self) -> EngineSession: + """Ensure we have an active session.""" + if self._session is None: + self._session = self._engine.createSession() + return self._session + + def __call__( + self, + text: str, + voice: str = "default", + speed: float = 1.0, + split_pattern: str = r"\n+", + **kwargs: Any, + ) -> Iterator[Any]: + """Call the backend like the old Kokoro pipeline. + + Returns an iterator of segment-like objects with .graphemes and .audio attributes. + """ + session = self._ensure_session() + + # Build synthesis request using the new API types + from abogen.tts_plugin.types import ( + AudioFormat, + ParameterValues, + SynthesisRequest, + VoiceSelection, + ) + + # Convert voice string to VoiceSelection + voice_selection = VoiceSelection(source="builtin", key=voice) + + # Convert speed and split_pattern to parameters + parameters = ParameterValues(values={"speed": speed, "split_pattern": split_pattern}) + + # Create request with default audio format + request = SynthesisRequest( + text=text, + voice=voice_selection, + parameters=parameters, + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + # Synthesize + result = session.synthesize(request) + + # Convert result to old-style segment iterator + from dataclasses import dataclass + + @dataclass + class Segment: + graphemes: str + audio: np.ndarray + + # Convert bytes back to numpy array + audio_array = np.frombuffer(result.data, dtype=np.float32) + + # The new API returns a single audio result, but the old API returns + # an iterator of segments. We need to split the text and audio accordingly. + # For now, return a single segment with the full text and audio. + yield Segment( + graphemes=text, + audio=audio_array, + ) + + def dispose(self) -> None: + """Dispose the session.""" + if self._session is not None: + try: + self._session.dispose() + except Exception: + pass + self._session = None + + def __del__(self) -> None: + """Cleanup on garbage collection.""" + self.dispose() + + +def create_backend(backend_id: str, **kwargs: Any) -> Any: + """Create a TTS backend using the new Plugin Architecture. + + This is a drop-in replacement for the old `create_backend()` function + from `abogen.tts_backend_registry`. + + Args: + backend_id: The backend/plugin ID (e.g., "kokoro") + **kwargs: Arguments passed to the engine constructor + + Returns: + A callable backend that matches the old TTSBackend protocol + + Raises: + KeyError: If plugin_id is not found + Exception: If engine creation fails + """ + manager = get_plugin_manager() + engine = manager.create_engine(backend_id, **kwargs) + return CompatBackend(engine, **kwargs) diff --git a/abogen/tts_plugin/plugin_manager.py b/abogen/tts_plugin/plugin_manager.py new file mode 100644 index 0000000..1589d08 --- /dev/null +++ b/abogen/tts_plugin/plugin_manager.py @@ -0,0 +1,153 @@ +"""Plugin Manager + +Provides a simple interface for consumers to access TTS engines via the +new Plugin Architecture. Discovers, loads, and manages plugins from the +plugins directory. + +Usage: + from abogen.tts_plugin.plugin_manager import get_plugin_manager + + manager = get_plugin_manager() + engine = manager.create_engine("kokoro", lang_code="a", device="cpu") + session = engine.create_session() + try: + result = session.synthesize("Hello world") + finally: + session.dispose() +""" + +from typing import Any, Dict, List, Optional, Type + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.manifest import PluginManifest +from abogen.tts_plugin.types import AudioFormat + + +class PluginManager: + """Manages TTS plugins and provides a simple interface for consumers.""" + + def __init__(self) -> None: + self._plugins: Dict[str, dict] = {} + self._engines: Dict[str, Engine] = {} + self._loaded = False + + def discover(self, plugins_dir: str = "plugins") -> None: + """Discover and load all plugins from the given directory.""" + import os + from pathlib import Path + from abogen.tts_plugin.loader import load_plugin_from_dir + + self._plugins.clear() + self._engines.clear() + + plugins_path = Path(plugins_dir) + if not plugins_path.exists(): + self._loaded = True + return + + for entry in plugins_path.iterdir(): + if entry.is_dir() and (entry / "__init__.py").exists(): + try: + result = load_plugin_from_dir(entry) + if result.success and result.manifest is not None: + self._plugins[result.manifest.id] = { + "manifest": result.manifest, + "create_engine": result.create_engine, + "module": result.module, + } + except Exception as e: + # Log error but continue with other plugins + print(f"Warning: Failed to load plugin from {entry}: {e}") + + self._loaded = True + + def _ensure_loaded(self) -> None: + """Ensure plugins have been discovered.""" + if not self._loaded: + self.discover() + + def list_plugins(self) -> List[PluginManifest]: + """Return manifests for all loaded plugins.""" + self._ensure_loaded() + return [info["manifest"] for info in self._plugins.values()] + + def get_plugin(self, plugin_id: str) -> Optional[dict]: + """Get plugin info by ID.""" + self._ensure_loaded() + return self._plugins.get(plugin_id) + + def has_plugin(self, plugin_id: str) -> bool: + """Check if a plugin is loaded.""" + self._ensure_loaded() + return plugin_id in self._plugins + + def create_engine(self, plugin_id: str, **kwargs: Any) -> Engine: + """Create an engine instance for the given plugin. + + Args: + plugin_id: The plugin identifier (e.g., "kokoro") + **kwargs: Arguments passed to the engine constructor + + Returns: + An Engine instance + + Raises: + KeyError: If plugin_id is not found + Exception: If engine creation fails + """ + self._ensure_loaded() + + if plugin_id not in self._plugins: + raise KeyError(f"Plugin not found: {plugin_id}") + + plugin_info = self._plugins[plugin_id] + create_engine_func = plugin_info["create_engine"] + + # Create engine using the plugin's factory + engine = create_engine_func(**kwargs) + return engine + + def get_or_create_engine(self, plugin_id: str, **kwargs: Any) -> Engine: + """Get an existing engine or create a new one. + + Engines are cached by plugin_id. If you need multiple instances + with different parameters, use create_engine() directly. + """ + self._ensure_loaded() + + cache_key = plugin_id + if cache_key in self._engines: + return self._engines[cache_key] + + engine = self.create_engine(plugin_id, **kwargs) + self._engines[cache_key] = engine + return engine + + def dispose_all(self) -> None: + """Dispose all cached engines.""" + for engine in self._engines.values(): + try: + engine.dispose() + except Exception: + pass # dispose() should never raise + self._engines.clear() + + +# Global singleton +_manager: Optional[PluginManager] = None + + +def get_plugin_manager() -> PluginManager: + """Get the global PluginManager instance.""" + global _manager + if _manager is None: + _manager = PluginManager() + return _manager + + +def reset_plugin_manager() -> None: + """Reset the global PluginManager (for testing).""" + global _manager + if _manager is not None: + _manager.dispose_all() + _manager = None diff --git a/abogen/webui/routes/utils/preview.py b/abogen/webui/routes/utils/preview.py index 6ab6ad9..f96790b 100644 --- a/abogen/webui/routes/utils/preview.py +++ b/abogen/webui/routes/utils/preview.py @@ -1,234 +1,234 @@ -import io -import threading -from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple -import numpy as np -import soundfile as sf -from flask import current_app, send_file -from flask.typing import ResponseReturnValue - - -SPLIT_PATTERN = r"\n+" -SAMPLE_RATE = 24000 - -_preview_pipelines: Dict[Tuple[str, str], Any] = {} -_preview_pipeline_lock = threading.Lock() - - -def _select_device() -> str: - import platform - - try: - import torch # type: ignore[import-not-found] - except Exception: - return "cpu" - - system = platform.system() - if system == "Darwin" and platform.processor() == "arm": - try: - if torch.backends.mps.is_available(): - return "mps" - except Exception: - pass - return "cpu" - - try: - if torch.cuda.is_available(): - return "cuda" - except Exception: - pass - return "cpu" - - -def _resolve_pipeline(language: str, use_gpu: bool) -> Tuple[Any, bool]: - devices: List[str] = ["cpu"] - if use_gpu: - preferred = _select_device() - if preferred != "cpu": - devices.insert(0, preferred) - - last_error: Optional[Exception] = None - for device in devices: - try: - return get_preview_pipeline(language, device), device != "cpu" - except Exception as exc: - last_error = exc - - raise RuntimeError("Preview pipeline is unavailable") from last_error - - -def _to_float32(audio_segment) -> np.ndarray: - if audio_segment is None: - return np.zeros(0, dtype="float32") - - tensor = audio_segment - if hasattr(tensor, "detach"): - tensor = tensor.detach() - if hasattr(tensor, "cpu"): - try: - tensor = tensor.cpu() - except Exception: - pass - if hasattr(tensor, "numpy"): - return np.asarray(tensor.numpy(), dtype="float32").reshape(-1) - return np.asarray(tensor, dtype="float32").reshape(-1) - -def get_preview_pipeline(language: str, device: str) -> Any: - key = (language, device) - with _preview_pipeline_lock: - pipeline = _preview_pipelines.get(key) - if pipeline is not None: - return pipeline - from abogen.tts_backend_registry import create_backend - - pipeline = create_backend("kokoro", lang_code=language, device=device) - _preview_pipelines[key] = pipeline - return pipeline - -def generate_preview_audio( - text: str, - voice_spec: str, - language: str, - speed: float, - use_gpu: bool, - tts_provider: str = "kokoro", - supertonic_total_steps: int = 5, - max_seconds: float = 8.0, - pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None, - manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None, - speakers: Optional[Mapping[str, Any]] = None, -) -> bytes: - if not text.strip(): - raise ValueError("Preview text is required") - - provider = (tts_provider or "kokoro").strip().lower() - - # Apply pronunciation/manual overrides first so tokens like `Unfu*k` still match - # before any downstream normalization potentially strips punctuation. - source_text = text - if pronunciation_overrides or manual_overrides or speakers: - try: - from abogen.webui import conversion_runner as runner - - class _PreviewJob: - def __init__(self): - self.language = language - self.voice = voice_spec - self.speakers = speakers - self.manual_overrides = list(manual_overrides or []) - self.pronunciation_overrides = list(pronunciation_overrides or []) - - job = _PreviewJob() - merged = runner._merge_pronunciation_overrides(job) - rules = runner._compile_pronunciation_rules(merged) - source_text = runner._apply_pronunciation_rules(source_text, rules) - except Exception: - current_app.logger.exception("Preview override application failed; using raw text") - source_text = text - - normalized_text = source_text - if provider != "supertonic": - try: - from abogen.kokoro_text_normalization import normalize_for_pipeline - - normalized_text = normalize_for_pipeline(source_text) - except Exception: - current_app.logger.exception("Preview normalization failed; using raw text") - normalized_text = source_text - - if provider == "supertonic": - from abogen.tts_backend_registry import create_backend - - pipeline = create_backend("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps) - segments = pipeline( - normalized_text, - voice=voice_spec, - speed=speed, - split_pattern=SPLIT_PATTERN, - total_steps=supertonic_total_steps, - ) - else: - pipeline, pipeline_uses_gpu = _resolve_pipeline(language, use_gpu) - if pipeline is None: - raise RuntimeError("Preview pipeline is unavailable") - - voice_choice: Any = voice_spec - if voice_spec and "*" in voice_spec: - from abogen.voice_formulas import get_new_voice - - voice_choice = get_new_voice(pipeline, voice_spec, pipeline_uses_gpu) - - segments = pipeline( - normalized_text, - voice=voice_choice, - speed=speed, - split_pattern=SPLIT_PATTERN, - ) - - audio_chunks: List[np.ndarray] = [] - accumulated = 0 - max_samples = int(max(1.0, max_seconds) * SAMPLE_RATE) - - for segment in segments: - graphemes = getattr(segment, "graphemes", "").strip() - if not graphemes: - continue - audio = _to_float32(getattr(segment, "audio", None)) - if audio.size == 0: - continue - remaining = max_samples - accumulated - if remaining <= 0: - break - if audio.shape[0] > remaining: - audio = audio[:remaining] - audio_chunks.append(audio) - accumulated += audio.shape[0] - if accumulated >= max_samples: - break - - if not audio_chunks: - raise RuntimeError("Preview could not be generated") - - audio_data = np.concatenate(audio_chunks) - buffer = io.BytesIO() - sf.write(buffer, audio_data, SAMPLE_RATE, format="WAV") - return buffer.getvalue() - -def synthesize_preview( - text: str, - voice_spec: str, - language: str, - speed: float, - use_gpu: bool, - tts_provider: str = "kokoro", - supertonic_total_steps: int = 5, - max_seconds: float = 8.0, - pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None, - manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None, - speakers: Optional[Mapping[str, Any]] = None, -) -> ResponseReturnValue: - try: - audio_bytes = generate_preview_audio( - text=text, - voice_spec=voice_spec, - language=language, - speed=speed, - use_gpu=use_gpu, - tts_provider=tts_provider, - supertonic_total_steps=supertonic_total_steps, - max_seconds=max_seconds, - pronunciation_overrides=pronunciation_overrides, - manual_overrides=manual_overrides, - speakers=speakers, - ) - except Exception as e: - raise e - - buffer = io.BytesIO(audio_bytes) - response = send_file( - buffer, - mimetype="audio/wav", - as_attachment=False, - download_name="speaker_preview.wav", - ) - response.headers["Cache-Control"] = "no-store" - return response +import io +import threading +from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple +import numpy as np +import soundfile as sf +from flask import current_app, send_file +from flask.typing import ResponseReturnValue + + +SPLIT_PATTERN = r"\n+" +SAMPLE_RATE = 24000 + +_preview_pipelines: Dict[Tuple[str, str], Any] = {} +_preview_pipeline_lock = threading.Lock() + + +def _select_device() -> str: + import platform + + try: + import torch # type: ignore[import-not-found] + except Exception: + return "cpu" + + system = platform.system() + if system == "Darwin" and platform.processor() == "arm": + try: + if torch.backends.mps.is_available(): + return "mps" + except Exception: + pass + return "cpu" + + try: + if torch.cuda.is_available(): + return "cuda" + except Exception: + pass + return "cpu" + + +def _resolve_pipeline(language: str, use_gpu: bool) -> Tuple[Any, bool]: + devices: List[str] = ["cpu"] + if use_gpu: + preferred = _select_device() + if preferred != "cpu": + devices.insert(0, preferred) + + last_error: Optional[Exception] = None + for device in devices: + try: + return get_preview_pipeline(language, device), device != "cpu" + except Exception as exc: + last_error = exc + + raise RuntimeError("Preview pipeline is unavailable") from last_error + + +def _to_float32(audio_segment) -> np.ndarray: + if audio_segment is None: + return np.zeros(0, dtype="float32") + + tensor = audio_segment + if hasattr(tensor, "detach"): + tensor = tensor.detach() + if hasattr(tensor, "cpu"): + try: + tensor = tensor.cpu() + except Exception: + pass + if hasattr(tensor, "numpy"): + return np.asarray(tensor.numpy(), dtype="float32").reshape(-1) + return np.asarray(tensor, dtype="float32").reshape(-1) + +def get_preview_pipeline(language: str, device: str) -> Any: + key = (language, device) + with _preview_pipeline_lock: + pipeline = _preview_pipelines.get(key) + if pipeline is not None: + return pipeline + from abogen.tts_plugin.compat import create_backend + + pipeline = create_backend("kokoro", lang_code=language, device=device) + _preview_pipelines[key] = pipeline + return pipeline + +def generate_preview_audio( + text: str, + voice_spec: str, + language: str, + speed: float, + use_gpu: bool, + tts_provider: str = "kokoro", + supertonic_total_steps: int = 5, + max_seconds: float = 8.0, + pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None, + manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None, + speakers: Optional[Mapping[str, Any]] = None, +) -> bytes: + if not text.strip(): + raise ValueError("Preview text is required") + + provider = (tts_provider or "kokoro").strip().lower() + + # Apply pronunciation/manual overrides first so tokens like `Unfu*k` still match + # before any downstream normalization potentially strips punctuation. + source_text = text + if pronunciation_overrides or manual_overrides or speakers: + try: + from abogen.webui import conversion_runner as runner + + class _PreviewJob: + def __init__(self): + self.language = language + self.voice = voice_spec + self.speakers = speakers + self.manual_overrides = list(manual_overrides or []) + self.pronunciation_overrides = list(pronunciation_overrides or []) + + job = _PreviewJob() + merged = runner._merge_pronunciation_overrides(job) + rules = runner._compile_pronunciation_rules(merged) + source_text = runner._apply_pronunciation_rules(source_text, rules) + except Exception: + current_app.logger.exception("Preview override application failed; using raw text") + source_text = text + + normalized_text = source_text + if provider != "supertonic": + try: + from abogen.kokoro_text_normalization import normalize_for_pipeline + + normalized_text = normalize_for_pipeline(source_text) + except Exception: + current_app.logger.exception("Preview normalization failed; using raw text") + normalized_text = source_text + + if provider == "supertonic": + from abogen.tts_plugin.compat import create_backend + + pipeline = create_backend("supertonic", sample_rate=SAMPLE_RATE, auto_download=True, total_steps=supertonic_total_steps) + segments = pipeline( + normalized_text, + voice=voice_spec, + speed=speed, + split_pattern=SPLIT_PATTERN, + total_steps=supertonic_total_steps, + ) + else: + pipeline, pipeline_uses_gpu = _resolve_pipeline(language, use_gpu) + if pipeline is None: + raise RuntimeError("Preview pipeline is unavailable") + + voice_choice: Any = voice_spec + if voice_spec and "*" in voice_spec: + from abogen.voice_formulas import get_new_voice + + voice_choice = get_new_voice(pipeline, voice_spec, pipeline_uses_gpu) + + segments = pipeline( + normalized_text, + voice=voice_choice, + speed=speed, + split_pattern=SPLIT_PATTERN, + ) + + audio_chunks: List[np.ndarray] = [] + accumulated = 0 + max_samples = int(max(1.0, max_seconds) * SAMPLE_RATE) + + for segment in segments: + graphemes = getattr(segment, "graphemes", "").strip() + if not graphemes: + continue + audio = _to_float32(getattr(segment, "audio", None)) + if audio.size == 0: + continue + remaining = max_samples - accumulated + if remaining <= 0: + break + if audio.shape[0] > remaining: + audio = audio[:remaining] + audio_chunks.append(audio) + accumulated += audio.shape[0] + if accumulated >= max_samples: + break + + if not audio_chunks: + raise RuntimeError("Preview could not be generated") + + audio_data = np.concatenate(audio_chunks) + buffer = io.BytesIO() + sf.write(buffer, audio_data, SAMPLE_RATE, format="WAV") + return buffer.getvalue() + +def synthesize_preview( + text: str, + voice_spec: str, + language: str, + speed: float, + use_gpu: bool, + tts_provider: str = "kokoro", + supertonic_total_steps: int = 5, + max_seconds: float = 8.0, + pronunciation_overrides: Optional[Iterable[Mapping[str, Any]]] = None, + manual_overrides: Optional[Iterable[Mapping[str, Any]]] = None, + speakers: Optional[Mapping[str, Any]] = None, +) -> ResponseReturnValue: + try: + audio_bytes = generate_preview_audio( + text=text, + voice_spec=voice_spec, + language=language, + speed=speed, + use_gpu=use_gpu, + tts_provider=tts_provider, + supertonic_total_steps=supertonic_total_steps, + max_seconds=max_seconds, + pronunciation_overrides=pronunciation_overrides, + manual_overrides=manual_overrides, + speakers=speakers, + ) + except Exception as e: + raise e + + buffer = io.BytesIO(audio_bytes) + response = send_file( + buffer, + mimetype="audio/wav", + as_attachment=False, + download_name="speaker_preview.wav", + ) + response.headers["Cache-Control"] = "no-store" + return response diff --git a/tests/contracts/test_integration_pr5.py b/tests/contracts/test_integration_pr5.py new file mode 100644 index 0000000..5737443 --- /dev/null +++ b/tests/contracts/test_integration_pr5.py @@ -0,0 +1,400 @@ +"""Integration tests for PR #5: Migrate First Consumer to Plugin Architecture. + +These tests verify: +1. Consumer Flow Test: consumer → plugin → engine → session → synthesis → result +2. Lifecycle Test: dispose, no leaks, error handling +3. Regression Test: old path vs new path equivalence + +Tests use mock plugins to avoid requiring real TTS dependencies. +""" + +import pytest +from typing import Any, Iterator +from unittest.mock import MagicMock, patch + +import numpy as np + +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.errors import EngineError +from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager +from abogen.tts_plugin.compat import CompatBackend, create_backend +from abogen.tts_plugin.types import ( + AudioFormat, + Duration, + ParameterValues, + SynthesisRequest, + SynthesizedAudio, + VoiceSelection, +) + + +class MockEngineSession: + """Mock EngineSession that records calls for verification.""" + + def __init__(self): + self._disposed = False + self.synthesize_calls = [] + + def synthesize(self, request: SynthesisRequest) -> SynthesizedAudio: + if self._disposed: + raise EngineError("Session disposed") + + self.synthesize_calls.append(request) + + # Return fake audio + audio = np.ones(1000, dtype=np.float32) * 0.5 + return SynthesizedAudio( + data=audio.tobytes(), + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=1000 / 24000), + ) + + def dispose(self) -> None: + self._disposed = True + + +class MockEngine: + """Mock Engine that creates MockEngineSessions.""" + + def __init__(self, **kwargs): + self.kwargs = kwargs + self._disposed = False + self.sessions_created = [] + + def createSession(self) -> MockEngineSession: + if self._disposed: + raise EngineError("Engine disposed") + session = MockEngineSession() + self.sessions_created.append(session) + return session + + def dispose(self) -> None: + self._disposed = True + + +def create_mock_plugin(create_engine_func=None): + """Helper to create a mock plugin module.""" + if create_engine_func is None: + create_engine_func = lambda **kwargs: MockEngine(**kwargs) + + from abogen.tts_plugin.manifest import PluginManifest, EngineManifest + + manifest = PluginManifest( + id="mock_tts", + name="Mock TTS", + version="1.0.0", + api_version="1.0", + description="Mock TTS for testing", + author="Test", + capabilities=(), + requires=None, + engine=EngineManifest( + voiceSources=(), + parameters=(), + audioFormats=(), + ), + ) + + return { + "PLUGIN_MANIFEST": manifest, + "MODEL_REQUIREMENTS": [], + "create_engine": create_mock_plugin_engine if create_engine_func is None else create_engine_func, + } + + +def create_mock_plugin_engine(**kwargs): + """Default mock plugin engine factory.""" + return MockEngine(**kwargs) + + +class TestConsumerFlow: + """Consumer Flow Test: consumer → plugin → engine → session → synthesis → result""" + + def test_full_consumer_flow(self): + """Verify complete flow from consumer to audio output.""" + manager = PluginManager() + + # Register mock plugin + mock_plugin = create_mock_plugin() + manager._plugins["mock_tts"] = mock_plugin + manager._loaded = True + + # Step 1: Consumer gets plugin + assert manager.has_plugin("mock_tts") is True + + # Step 2: Plugin creates engine + engine = manager.create_engine("mock_tts") + assert engine is not None + assert isinstance(engine, MockEngine) + + # Step 3: Engine creates session + session = engine.createSession() + assert session is not None + assert isinstance(session, MockEngineSession) + + # Step 4: Session synthesizes + request = SynthesisRequest( + text="Hello world", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(values={"speed": 1.0}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + + # Step 5: Result returned + assert result is not None + assert isinstance(result, SynthesizedAudio) + assert len(result.data) > 0 + assert result.format.mime == "audio/wav" + assert result.duration.seconds > 0 + + def test_consumer_flow_via_compat_adapter(self): + """Verify flow through compatibility adapter matches direct flow.""" + manager = PluginManager() + + # Register mock plugin + mock_plugin = create_mock_plugin() + manager._plugins["mock_tts"] = mock_plugin + manager._loaded = True + + # Use compat adapter + with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager): + backend = create_backend("mock_tts") + + # Call like old TTSBackend + segments = list(backend("Hello world", voice="default", speed=1.0)) + + # Verify result + assert len(segments) >= 1 + segment = segments[0] + assert hasattr(segment, "graphemes") + assert hasattr(segment, "audio") + assert segment.graphemes == "Hello world" + + +class TestLifecycle: + """Lifecycle Test: dispose, no leaks, error handling""" + + def test_session_dispose_is_idempotent(self): + """dispose() can be called multiple times safely.""" + session = MockEngineSession() + + session.dispose() + session.dispose() # Should not raise + assert session._disposed is True + + def test_session_synthesize_after_dispose_raises(self): + """synthesize() after dispose() raises EngineError.""" + session = MockEngineSession() + session.dispose() + + request = SynthesisRequest( + text="test", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + with pytest.raises(EngineError): + session.synthesize(request) + + def test_engine_dispose_is_idempotent(self): + """Engine dispose() can be called multiple times safely.""" + engine = MockEngine() + + engine.dispose() + engine.dispose() # Should not raise + assert engine._disposed is True + + def test_engine_create_session_after_dispose_raises(self): + """createSession() after dispose() raises EngineError.""" + engine = MockEngine() + engine.dispose() + + with pytest.raises(EngineError): + engine.createSession() + + def test_full_lifecycle(self): + """Test complete lifecycle: create → use → dispose.""" + engine = MockEngine() + + # Create and use session + session = engine.createSession() + request = SynthesisRequest( + text="test", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + result = session.synthesize(request) + assert len(result.data) > 0 + + # Dispose session + session.dispose() + assert session._disposed is True + + # Dispose engine + engine.dispose() + assert engine._disposed is True + + def test_no_session_leak_on_engine_dispose(self): + """Engine can be disposed even if sessions were created.""" + engine = MockEngine() + + # Create multiple sessions + session1 = engine.createSession() + session2 = engine.createSession() + + # Use sessions + request = SynthesisRequest( + text="test", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + session1.synthesize(request) + session2.synthesize(request) + + # Dispose engine (sessions still exist but engine is disposed) + engine.dispose() + assert engine._disposed is True + + # Sessions can still be used (they hold reference to pipeline) + result = session1.synthesize(request) + assert len(result.data) > 0 + + def test_error_handling_in_synthesis(self): + """Error during synthesis is handled correctly.""" + class FailingSession: + def synthesize(self, request): + raise EngineError("Synthesis failed") + + def dispose(self): + pass + + session = FailingSession() + request = SynthesisRequest( + text="test", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + + with pytest.raises(EngineError, match="Synthesis failed"): + session.synthesize(request) + + +class TestRegression: + """Regression Test: old path vs new path equivalence""" + + def test_old_path_vs_new_path_same_result(self): + """Both paths should produce equivalent results.""" + # Setup mock plugin + manager = PluginManager() + mock_plugin = create_mock_plugin() + manager._plugins["mock_tts"] = mock_plugin + manager._loaded = True + + # New path: Plugin Manager → Engine → Session → Synthesis + with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager): + new_backend = create_backend("mock_tts") + new_segments = list(new_backend("Hello world", voice="default", speed=1.0)) + + # Old path: Direct MockEngine (simulating old registry) + old_engine = MockEngine() + old_session = old_engine.createSession() + request = SynthesisRequest( + text="Hello world", + voice=VoiceSelection(source="builtin", key="default"), + parameters=ParameterValues(values={"speed": 1.0}), + format=AudioFormat(mime="audio/wav", extension="wav"), + ) + old_result = old_session.synthesize(request) + + # Compare results + # New path returns segments, old path returns SynthesizedAudio + # But both should have valid audio data + assert len(new_segments) >= 1 + assert len(old_result.data) > 0 + + # Both should have same format + assert new_segments[0].audio.dtype == np.float32 + + def test_compat_adapter_matches_old_interface(self): + """Compat adapter should match old TTSBackend interface.""" + manager = PluginManager() + mock_plugin = create_mock_plugin() + manager._plugins["mock_tts"] = mock_plugin + manager._loaded = True + + with patch("abogen.tts_plugin.compat.get_plugin_manager", return_value=manager): + backend = create_backend("mock_tts", lang_code="a", device="cpu") + + # Old interface: pipeline(text, voice=..., speed=..., split_pattern=...) + segments = list(backend( + "Hello world", + voice="af_heart", + speed=1.0, + split_pattern=r"\n+" + )) + + # Should return segments with graphemes and audio + assert len(segments) >= 1 + segment = segments[0] + assert segment.graphemes == "Hello world" + assert isinstance(segment.audio, np.ndarray) + assert segment.audio.dtype == np.float32 + assert len(segment.audio) > 0 + + +class TestPluginManagerIntegration: + """Integration tests for PluginManager.""" + + def test_plugin_manager_singleton_pattern(self): + """Global plugin manager follows singleton pattern.""" + reset_plugin_manager() + + manager1 = get_plugin_manager() + manager2 = get_plugin_manager() + + assert manager1 is manager2 + + reset_plugin_manager() + + manager3 = get_plugin_manager() + assert manager1 is not manager3 + + def test_plugin_manager_discover_plugins(self): + """Plugin manager can discover plugins from directory.""" + manager = PluginManager() + + # Discover from test plugins directory + manager.discover("tests/plugins") + + # Should find valid_plugin + # (This depends on test plugins existing) + plugins = manager.list_plugins() + assert isinstance(plugins, list) + + def test_plugin_manager_dispose_all(self): + """Plugin manager can dispose all cached engines.""" + manager = PluginManager() + + # Register mock plugin + mock_plugin = create_mock_plugin() + manager._plugins["mock_tts"] = mock_plugin + manager._loaded = True + + # Create engines + engine1 = manager.get_or_create_engine("mock_tts") + engine2 = manager.get_or_create_engine("mock_tts") + + # Dispose all + manager.dispose_all() + + # Engines should be disposed + assert engine1._disposed is True + assert engine2._disposed is True + + # Cache should be empty + assert len(manager._engines) == 0 diff --git a/tests/contracts/test_plugin_manager_contract.py b/tests/contracts/test_plugin_manager_contract.py new file mode 100644 index 0000000..1a92b73 --- /dev/null +++ b/tests/contracts/test_plugin_manager_contract.py @@ -0,0 +1,264 @@ +"""Integration tests for Plugin Manager and compatibility adapter.""" + +import pytest +from unittest.mock import MagicMock, patch + +from abogen.tts_plugin.plugin_manager import PluginManager, get_plugin_manager, reset_plugin_manager +from abogen.tts_plugin.compat import CompatBackend, create_backend +from abogen.tts_plugin.engine import Engine, EngineSession +from abogen.tts_plugin.types import SynthesisRequest, SynthesizedAudio, AudioFormat + + +class FakeEngine: + """Fake Engine for testing.""" + + def __init__(self, **kwargs): + self.kwargs = kwargs + self._disposed = False + + def createSession(self): + return FakeEngineSession() + + def dispose(self): + self._disposed = True + + @property + def manifest(self): + return MagicMock() + + +class FakeEngineSession: + """Fake EngineSession for testing.""" + + def __init__(self): + self._disposed = False + + def synthesize(self, request): + # Return fake audio + import numpy as np + from abogen.tts_plugin.types import AudioFormat, Duration, SynthesizedAudio + + audio = np.zeros(1000, dtype=np.float32) + return SynthesizedAudio( + data=audio.tobytes(), + format=AudioFormat(mime="audio/wav", extension="wav"), + duration=Duration(seconds=0.04167), # 1000 samples / 24000 Hz + ) + + def dispose(self): + self._disposed = True + + +class TestPluginManager: + """Test PluginManager functionality.""" + + def test_plugin_manager_creation(self): + """PluginManager can be created.""" + manager = PluginManager() + assert manager is not None + + def test_plugin_manager_list_discovers_plugins(self): + """PluginManager discovers plugins from plugins directory.""" + manager = PluginManager() + manager.discover("plugins") + plugins = manager.list_plugins() + # Should discover kokoro plugin if it exists + assert isinstance(plugins, list) + + def test_plugin_manager_has_plugin_after_discover(self): + """PluginManager reports plugins after discovery.""" + manager = PluginManager() + manager.discover("plugins") + # kokoro plugin should be discovered if plugins/kokoro exists + # This is expected behavior + assert isinstance(manager._plugins, dict) + + def test_plugin_manager_get_plugin_after_discover(self): + """PluginManager returns plugin info after discovery.""" + manager = PluginManager() + manager.discover("plugins") + # kokoro plugin should be discovered if plugins/kokoro exists + assert isinstance(manager._plugins, dict) + + def test_plugin_manager_create_engine_not_found(self): + """PluginManager raises KeyError for unknown plugins.""" + manager = PluginManager() + with pytest.raises(KeyError, match="Plugin not found"): + manager.create_engine("nonexistent") + + def test_plugin_manager_discover_with_empty_dir(self): + """PluginManager handles missing plugins directory.""" + manager = PluginManager() + manager.discover("/nonexistent/path") + plugins = manager.list_plugins() + assert plugins == [] + + def test_global_plugin_manager_singleton(self): + """Global PluginManager is a singleton.""" + reset_plugin_manager() + manager1 = get_plugin_manager() + manager2 = get_plugin_manager() + assert manager1 is manager2 + reset_plugin_manager() + + def test_reset_plugin_manager(self): + """reset_plugin_manager clears the singleton.""" + manager1 = get_plugin_manager() + reset_plugin_manager() + manager2 = get_plugin_manager() + assert manager1 is not manager2 + reset_plugin_manager() + + +class TestCompatBackend: + """Test CompatBackend functionality.""" + + def test_compat_backend_creation(self): + """CompatBackend can be created.""" + engine = FakeEngine() + backend = CompatBackend(engine) + assert backend is not None + + def test_compat_backend_callable(self): + """CompatBackend is callable like old TTSBackend.""" + engine = FakeEngine() + backend = CompatBackend(engine) + + # Should be callable + assert callable(backend) + + def test_compat_backend_synthesize(self): + """CompatBackend can synthesize text.""" + engine = FakeEngine() + backend = CompatBackend(engine) + + # Call the backend + segments = list(backend("Hello world", voice="default", speed=1.0)) + + # Should return at least one segment + assert len(segments) >= 1 + + # Segment should have graphemes and audio + segment = segments[0] + assert hasattr(segment, "graphemes") + assert hasattr(segment, "audio") + assert segment.graphemes == "Hello world" + + def test_compat_backend_dispose(self): + """CompatBackend can be disposed.""" + engine = FakeEngine() + backend = CompatBackend(engine) + + # Create a session by calling + list(backend("test")) + + # Dispose should not raise + backend.dispose() + + # Double dispose should be safe + backend.dispose() + + +class TestCreateBackendCompat: + """Test create_backend compatibility function.""" + + def test_create_backend_returns_callable(self): + """create_backend returns a callable backend.""" + # Mock the plugin manager + with patch("abogen.tts_plugin.compat.get_plugin_manager") as mock_get_manager: + mock_manager = MagicMock() + mock_get_manager.return_value = mock_manager + + mock_engine = FakeEngine() + mock_manager.create_engine.return_value = mock_engine + + backend = create_backend("kokoro", lang_code="a", device="cpu") + + assert callable(backend) + mock_manager.create_engine.assert_called_once_with("kokoro", lang_code="a", device="cpu") + + def test_create_backend_raises_for_unknown_plugin(self): + """create_backend raises KeyError for unknown plugins.""" + with patch("abogen.tts_plugin.compat.get_plugin_manager") as mock_get_manager: + mock_manager = MagicMock() + mock_get_manager.return_value = mock_manager + mock_manager.create_engine.side_effect = KeyError("Plugin not found") + + with pytest.raises(KeyError): + create_backend("nonexistent") + + +class TestPluginManagerWithFakePlugins: + """Test PluginManager with fake plugin loading.""" + + def test_plugin_manager_create_engine_from_plugin(self): + """PluginManager creates engine from loaded plugin.""" + manager = PluginManager() + + # Manually add a fake plugin + def fake_create_engine(**kwargs): + return FakeEngine(**kwargs) + + manager._plugins["fake"] = { + "manifest": MagicMock(), + "create_engine": fake_create_engine, + } + manager._loaded = True + + # Create engine + engine = manager.create_engine("fake", param="value") + + assert isinstance(engine, FakeEngine) + assert engine.kwargs == {"param": "value"} + + def test_plugin_manager_get_or_create_engine(self): + """PluginManager caches engines.""" + manager = PluginManager() + + call_count = 0 + + def fake_create_engine(**kwargs): + nonlocal call_count + call_count += 1 + return FakeEngine(**kwargs) + + manager._plugins["fake"] = { + "manifest": MagicMock(), + "create_engine": fake_create_engine, + } + manager._loaded = True + + # Get engine twice + engine1 = manager.get_or_create_engine("fake") + engine2 = manager.get_or_create_engine("fake") + + # Should be same instance + assert engine1 is engine2 + assert call_count == 1 + + def test_plugin_manager_dispose_all(self): + """PluginManager disposes all cached engines.""" + manager = PluginManager() + + def fake_create_engine(**kwargs): + return FakeEngine(**kwargs) + + manager._plugins["fake"] = { + "manifest": MagicMock(), + "create_engine": fake_create_engine, + } + manager._loaded = True + + # Create and cache engines + engine1 = manager.get_or_create_engine("fake") + engine2 = manager.get_or_create_engine("fake") + + # Dispose all + manager.dispose_all() + + # Engines should be disposed + assert engine1._disposed is True + assert engine2._disposed is True + + # Cache should be empty + assert len(manager._engines) == 0 diff --git a/tests/test_preview_applies_manual_overrides.py b/tests/test_preview_applies_manual_overrides.py index 69630c1..8501fa1 100644 --- a/tests/test_preview_applies_manual_overrides.py +++ b/tests/test_preview_applies_manual_overrides.py @@ -1,60 +1,60 @@ -from abogen.webui.routes.utils import preview - - -def test_preview_applies_manual_override_before_normalization(monkeypatch): - # Don't run real TTS/normalization; just exercise the override stage by - # forcing provider=kokoro and then stubbing normalize_for_pipeline. - - monkeypatch.setattr(preview, "get_preview_pipeline", lambda language, device: None) - - # Stub normalize_for_pipeline to be identity; we only care that overrides run. - class _Norm: - @staticmethod - def normalize_for_pipeline(text): - return text - - monkeypatch.setitem( - __import__("sys").modules, "abogen.kokoro_text_normalization", _Norm - ) - - # And stub the kokoro pipeline path so generate_preview_audio won't proceed. - # We'll instead validate by calling the override logic through generate_preview_audio - # with provider=supertonic and stub create_backend to capture input. - captured = {} - - class DummyPipeline: - def __init__(self, **kwargs): - pass - - def __call__(self, text, **kwargs): - captured["text"] = text - return iter(()) - - from abogen import tts_backend_registry - - original_create_backend = tts_backend_registry.create_backend - - def _mock_create_backend(backend_id, **kwargs): - if backend_id == "supertonic": - return DummyPipeline(**kwargs) - return original_create_backend(backend_id, **kwargs) - - monkeypatch.setattr(tts_backend_registry, "create_backend", _mock_create_backend) - - try: - preview.generate_preview_audio( - text="He said Unfu*k loudly.", - voice_spec="M1", - language="en", - speed=1.0, - use_gpu=False, - tts_provider="supertonic", - manual_overrides=[{"token": "Unfu*k", "pronunciation": "Unfuck"}], - ) - except Exception: - # generate_preview_audio will raise because no audio chunks; that's fine. - pass - - assert "text" in captured - assert "Unfuck" in captured["text"] - assert "Unfu*k" not in captured["text"] +from abogen.webui.routes.utils import preview + + +def test_preview_applies_manual_override_before_normalization(monkeypatch): + # Don't run real TTS/normalization; just exercise the override stage by + # forcing provider=kokoro and then stubbing normalize_for_pipeline. + + monkeypatch.setattr(preview, "get_preview_pipeline", lambda language, device: None) + + # Stub normalize_for_pipeline to be identity; we only care that overrides run. + class _Norm: + @staticmethod + def normalize_for_pipeline(text): + return text + + monkeypatch.setitem( + __import__("sys").modules, "abogen.kokoro_text_normalization", _Norm + ) + + # And stub the kokoro pipeline path so generate_preview_audio won't proceed. + # We'll instead validate by calling the override logic through generate_preview_audio + # with provider=supertonic and stub create_backend to capture input. + captured = {} + + class DummyPipeline: + def __init__(self, **kwargs): + pass + + def __call__(self, text, **kwargs): + captured["text"] = text + return iter(()) + + from abogen.tts_plugin import compat + + original_create_backend = compat.create_backend + + def _mock_create_backend(backend_id, **kwargs): + if backend_id == "supertonic": + return DummyPipeline(**kwargs) + return original_create_backend(backend_id, **kwargs) + + monkeypatch.setattr(compat, "create_backend", _mock_create_backend) + + try: + preview.generate_preview_audio( + text="He said Unfu*k loudly.", + voice_spec="M1", + language="en", + speed=1.0, + use_gpu=False, + tts_provider="supertonic", + manual_overrides=[{"token": "Unfu*k", "pronunciation": "Unfuck"}], + ) + except Exception: + # generate_preview_audio will raise because no audio chunks; that's fine. + pass + + assert "text" in captured + assert "Unfuck" in captured["text"] + assert "Unfu*k" not in captured["text"]