extract resolve_fallback_voice_spec to domain/voice_resolution.py; fix missing get_default_voice import and __custom_mix reset bug

abogen/domain/voice_resolution.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 from typing import Any, Dict, Optional, Set
 
-from abogen.tts_plugin.utils import get_voices
+from abogen.tts_plugin.utils import get_voices, get_default_voice
 from abogen.voice_formulas import extract_voice_ids
 from abogen.voice_cache import ensure_voice_assets
 
@@ -164,3 +164,27 @@ def chunk_voice_spec(job: Any, chunk: Dict[str, Any], fallback: str) -> str:
     if fallback:
         return fallback
     return job_voice_fallback(job)
+
+
+def resolve_fallback_voice_spec(
+    base_spec: str,
+    job_voice: str,
+    voice_cache_keys: list[str],
+    provider: str = "kokoro",
+) -> str:
+    """Resolve the voice spec for intro/outro with a priority fallback chain.
+
+    Priority: base_spec → job_voice → first voice_cache key → default voice.
+    ``"__custom_mix"`` is treated as empty (it is not a usable voice spec).
+    """
+    spec = base_spec or job_voice
+    if spec == "__custom_mix":
+        spec = job_voice or ""
+    if not spec:
+        for key in voice_cache_keys:
+            if key and key != "__custom_mix":
+                spec = key.split(":", 1)[-1]
+                break
+    if not spec:
+        spec = get_default_voice(provider)
+    return spec

abogen/webui/conversion_runner.py

@@ -84,6 +84,7 @@ from abogen.domain.voice_resolution import (
     initialize_voice_cache as _initialize_voice_cache,
     chapter_voice_spec as _chapter_voice_spec,
     chunk_voice_spec as _chunk_voice_spec,
+    resolve_fallback_voice_spec as _resolve_fallback_voice_spec,
 )
 from abogen.domain.chapter_overrides import apply_chapter_overrides as _apply_chapter_overrides
 from abogen.domain.metadata_merge import merge_metadata as _merge_metadata
@@ -535,15 +536,9 @@ def run_conversion_job(job: Job) -> None:
                 preview = book_intro_text if len(book_intro_text) <= 120 else f"{book_intro_text[:117]}…"
                 job.add_log(f"Title intro enabled: {preview}", level="debug")
 
-                intro_voice_spec = base_voice_spec or job.voice
-                if intro_voice_spec == "__custom_mix":
-                    intro_voice_spec = base_voice_spec or ""
-                if not intro_voice_spec:
-                    fallback_key = next(iter(voice_cache.keys()), "")
-                    if fallback_key and fallback_key != "__custom_mix":
-                        intro_voice_spec = fallback_key.split(":", 1)[-1]
-                if not intro_voice_spec:
-                    intro_voice_spec = get_default_voice("kokoro")
+                intro_voice_spec = _resolve_fallback_voice_spec(
+                    base_voice_spec, job.voice, list(voice_cache.keys())
+                )
 
                 if intro_voice_spec:
                     intro_provider, _, intro_voice_choice, intro_speed, intro_steps = resolve_voice_choice(
@@ -960,16 +955,9 @@ def run_conversion_job(job: Job) -> None:
 
         if getattr(job, "read_closing_outro", True):
             outro_text = _build_outro_text(job.metadata_tags, job.original_filename)
-            outro_voice_spec = base_voice_spec or job.voice
-            if outro_voice_spec == "__custom_mix":
-                outro_voice_spec = base_voice_spec or ""
-            if not outro_voice_spec:
-                fallback_key = next(iter(voice_cache.keys()), "")
-                if fallback_key and fallback_key != "__custom_mix":
-                    # `voice_cache` keys are internal and include provider prefixes.
-                    outro_voice_spec = fallback_key.split(":", 1)[-1]
-            if not outro_voice_spec:
-                outro_voice_spec = get_default_voice("kokoro")
+            outro_voice_spec = _resolve_fallback_voice_spec(
+                base_voice_spec, job.voice, list(voice_cache.keys())
+            )
 
             if outro_text and outro_voice_spec:
                 outro_start_time = current_time

tests/test_voice_fallback.py

@@ -0,0 +1,139 @@
+import sys
+import types
+
+if "soundfile" not in sys.modules:
+    soundfile_stub = types.ModuleType("soundfile")
+
+    class _SoundFileStub:  # pragma: no cover - placeholder to satisfy imports
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            raise RuntimeError("soundfile is not installed in the test environment")
+
+    soundfile_stub.SoundFile = _SoundFileStub  # type: ignore[attr-defined]
+    sys.modules["soundfile"] = soundfile_stub
+
+if "static_ffmpeg" not in sys.modules:
+    sys.modules["static_ffmpeg"] = types.ModuleType("static_ffmpeg")
+
+if "ebooklib" not in sys.modules:
+    ebooklib_stub = types.ModuleType("ebooklib")
+    ebooklib_epub_stub = types.ModuleType("ebooklib.epub")
+    ebooklib_stub.epub = ebooklib_epub_stub  # type: ignore[attr-defined]
+    sys.modules["ebooklib"] = ebooklib_stub
+    sys.modules["ebooklib.epub"] = ebooklib_epub_stub
+
+if "fitz" not in sys.modules:
+    sys.modules["fitz"] = types.ModuleType("fitz")
+
+if "markdown" not in sys.modules:
+    markdown_stub = types.ModuleType("markdown")
+
+    class _MarkdownStub:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            self.toc_tokens = []
+
+        def convert(self, text: str) -> str:
+            return text
+
+    markdown_stub.Markdown = _MarkdownStub  # type: ignore[attr-defined]
+    sys.modules["markdown"] = markdown_stub
+
+if "bs4" not in sys.modules:
+    bs4_stub = types.ModuleType("bs4")
+
+    class _BeautifulSoupStub:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            self._text = ""
+
+        def find(self, *args: object, **kwargs: object) -> None:
+            return None
+
+        def get_text(self) -> str:
+            return self._text
+
+        def decompose(self) -> None:  # pragma: no cover - compatibility shim
+            return None
+
+    class _NavigableStringStub(str):
+        pass
+
+    bs4_stub.BeautifulSoup = _BeautifulSoupStub  # type: ignore[attr-defined]
+    bs4_stub.NavigableString = _NavigableStringStub  # type: ignore[attr-defined]
+    sys.modules["bs4"] = bs4_stub
+
+
+from unittest.mock import patch, MagicMock
+
+
+class TestResolveFallbackVoiceSpec:
+    """Tests for the voice fallback resolution helper."""
+
+    def test_uses_base_voice_spec(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="af_heart",
+            job_voice="af_bella",
+            voice_cache_keys=[],
+        )
+        assert result == "af_heart"
+
+    def test_falls_back_to_job_voice(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="",
+            job_voice="af_bella",
+            voice_cache_keys=[],
+        )
+        assert result == "af_bella"
+
+    def test_skips_custom_mix_uses_job_voice(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="__custom_mix",
+            job_voice="af_bella",
+            voice_cache_keys=[],
+        )
+        assert result == "af_bella"
+
+    def test_falls_back_to_voice_cache(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="",
+            job_voice="",
+            voice_cache_keys=["kokoro:af_heart"],
+        )
+        assert result == "af_heart"
+
+    def test_skips_custom_mix_in_cache(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="",
+            job_voice="",
+            voice_cache_keys=["__custom_mix", "kokoro:af_heart"],
+        )
+        assert result == "af_heart"
+
+    def test_falls_back_to_default_voice(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        with patch("abogen.domain.voice_resolution.get_default_voice", return_value="af_heart"):
+            result = resolve_fallback_voice_spec(
+                base_spec="",
+                job_voice="",
+                voice_cache_keys=[],
+            )
+        assert result == "af_heart"
+
+    def test_empty_base_and_job_with_cache(self) -> None:
+        from abogen.domain.voice_resolution import resolve_fallback_voice_spec
+
+        result = resolve_fallback_voice_spec(
+            base_spec="",
+            job_voice="",
+            voice_cache_keys=["kokoro:af_bella", "kokoro:af_heart"],
+        )
+        assert result == "af_bella"