chore: remove dead test files and deleted conversion_adapter

All 8 deleted test files were duplicates of existing domain tests:
- test_chapter_overrides → covered by test_chapter_merge_normalize
- test_conversion_chapter_titles → covered by test_chapter_titles
- test_conversion_series → covered by test_title_builder
- test_conversion_voice_resolution → covered by test_voice_resolution
- test_voice_cache → covered by test_voice_resolution
- test_manual_overrides_applied_first → covered by test_pronunciation
- test_conversion_adapters → tested deleted conversion_adapter module
- test_import_layering → tested deleted conversion_adapter module

Also removed deleted conversion_adapter.py (logic moved to conversion_runner).
This commit is contained in:
Artem Akymenko
2026-07-27 09:56:51 +00:00
parent 98ab2d925e
commit f7a224cc46
9 changed files with 0 additions and 1650 deletions
-202
View File
@@ -1,202 +0,0 @@
"""WebUI adapter: Job -> ConversionRequest.
Converts a WebUI Job into a ConversionRequest that the application layer can process.
This adapter is the bridge between the WebUI layer and the application/domain layer.
The adapter is responsible for:
- Mapping Job fields to ConversionRequest fields
- Handling UI-specific state (logs, progress, cancellation)
- Providing PipelineProvider and VoiceResolver implementations
All conversions happen through this adapter — the application layer
never accesses Job directly.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from abogen.application.conversion_config import (
ChapterChunkConfig,
Epub3ExportConfig,
PronunciationConfig,
WordSubstitutionConfig,
)
from abogen.application.conversion_request import ConversionRequest
from abogen.application.conversion_ports import ConversionCancelled, ResolvedVoice
def build_conversion_request_from_job(job: Any) -> ConversionRequest:
"""Convert a WebUI Job into a ConversionRequest.
This is the primary function that maps Job fields to ConversionRequest.
All fields are copied — the request is independent of the Job.
Args:
job: WebUI Job instance
Returns:
ConversionRequest with all Job data mapped
"""
# Build word substitution config
word_substitution = None
if getattr(job, "word_substitutions_enabled", False):
word_substitution = WordSubstitutionConfig(
substitutions_list=getattr(job, "word_substitutions_list", ""),
case_sensitive=getattr(job, "case_sensitive_substitutions", False),
replace_caps=getattr(job, "replace_all_caps", False),
replace_numerals=getattr(job, "replace_numerals", False),
fix_punctuation=getattr(job, "fix_nonstandard_punctuation", False),
)
# Build pronunciation config
pronunciation = None
pron_overrides = job.pronunciation_overrides or []
manual_overrides = job.manual_overrides or []
heteronym_overrides = job.heteronym_overrides or []
norm_overrides = job.normalization_overrides or None
if pron_overrides or manual_overrides or heteronym_overrides or norm_overrides:
pronunciation = PronunciationConfig(
pronunciation_overrides=pron_overrides,
manual_overrides=manual_overrides,
heteronym_overrides=heteronym_overrides,
normalization_overrides=norm_overrides,
)
# Build chapter/chunk config
chapter_chunk = ChapterChunkConfig(
chapter_overrides=job.chapters or [],
chunks=job.chunks or [],
chunk_level=job.chunk_level,
speaker_mode=job.speaker_mode,
speakers=job.speakers or {},
)
# Build epub3 config
epub3_export = None
if getattr(job, "generate_epub3", False):
epub3_export = Epub3ExportConfig(
book_id=getattr(job, "id", ""),
)
return ConversionRequest(
# Source
source_path=Path(job.stored_path) if job.stored_path else None,
original_filename=job.original_filename,
# TTS Settings
language=job.language,
tts_provider=job.tts_provider,
voice=job.voice,
voice_profile=job.voice_profile,
speed=job.speed,
use_gpu=job.use_gpu,
supertonic_total_steps=job.supertonic_total_steps,
# Output Format
output_format=job.output_format,
subtitle_mode=job.subtitle_mode,
subtitle_format=job.subtitle_format,
max_subtitle_words=job.max_subtitle_words,
# Save Options
save_mode=job.save_mode,
output_folder=Path(job.output_folder) if job.output_folder else None,
save_chapters_separately=job.save_chapters_separately,
merge_chapters_at_end=job.merge_chapters_at_end,
separate_chapters_format=job.separate_chapters_format,
save_as_project=job.save_as_project,
# Timing
silence_between_chapters=job.silence_between_chapters,
chapter_intro_delay=job.chapter_intro_delay,
# Content Processing
replace_single_newlines=job.replace_single_newlines,
read_title_intro=job.read_title_intro,
read_closing_outro=job.read_closing_outro,
auto_prefix_chapter_titles=job.auto_prefix_chapter_titles,
normalize_chapter_opening_caps=job.normalize_chapter_opening_caps,
# Metadata
metadata_tags=job.metadata_tags or {},
# Artifacts
cover_image_path=Path(job.cover_image_path) if job.cover_image_path else None,
cover_image_mime=job.cover_image_mime,
# Feature configs
word_substitution=word_substitution,
pronunciation=pronunciation,
chapter_chunk=chapter_chunk,
epub3_export=epub3_export,
)
class WebJobEvents:
"""WebUI implementation of ConversionEvents protocol.
Wraps a Job to provide logging, progress, and cancellation.
"""
def __init__(self, job: Any):
self._job = job
def log(self, message: str, level: str = "info") -> None:
"""Log a message to the Job."""
self._job.add_log(message, level=level)
def progress(self, pct: int, etr: str) -> None:
"""Update progress on the Job."""
self._job.progress = pct / 100.0
self._job.etr_str = etr
def check_cancelled(self) -> None:
"""Check if the Job was cancelled.
Raises:
ConversionCancelled: If cancellation was requested
"""
if self._job.cancel_requested:
raise ConversionCancelled("Job cancelled by user")
class WebPipelineProvider:
"""WebUI implementation of PipelineProvider protocol.
Wraps PipelinePool to provide TTS backends.
"""
def __init__(self, pipeline_pool: Any):
self._pool = pipeline_pool
def get(self, provider: str, language: str, use_gpu: bool) -> Any:
"""Get a TTS backend instance."""
return self._pool.get(provider, language, use_gpu)
def dispose_all(self) -> None:
"""Dispose all backend resources."""
self._pool.dispose_all()
class WebVoiceResolver:
"""WebUI implementation of VoiceResolver protocol.
Wraps the voice resolution logic from conversion_runner.py.
"""
def __init__(
self,
resolve_fn: Callable[[str], tuple[str, str, Any, Optional[float], Optional[int]]],
):
"""Initialize with a voice resolution function.
Args:
resolve_fn: Function that takes a voice_spec and returns
(provider, resolved_spec, voice_choice, speed, steps)
"""
self._resolve_fn = resolve_fn
def resolve(self, voice_spec: str) -> ResolvedVoice:
"""Resolve a voice spec into a loaded voice."""
provider, resolved_spec, voice, speed, steps = self._resolve_fn(voice_spec)
return ResolvedVoice(
provider=provider,
resolved_spec=resolved_spec,
voice=voice,
speed=speed or 1.0,
supertonic_steps=steps or 5,
)
-195
View File
@@ -1,195 +0,0 @@
from __future__ import annotations
import sys
import types
def _install_dependency_stubs() -> None:
if "ebooklib" not in sys.modules:
ebooklib_stub = types.ModuleType("ebooklib")
epub_stub = types.ModuleType("ebooklib.epub")
setattr(ebooklib_stub, "epub", epub_stub)
sys.modules["ebooklib"] = ebooklib_stub
sys.modules["ebooklib.epub"] = epub_stub
if "dotenv" not in sys.modules:
dotenv_stub = types.ModuleType("dotenv")
def _noop(*_, **__):
return None
setattr(dotenv_stub, "load_dotenv", _noop)
setattr(dotenv_stub, "find_dotenv", lambda *_, **__: "")
sys.modules["dotenv"] = dotenv_stub
if "numpy" not in sys.modules:
numpy_stub = types.ModuleType("numpy")
class _DummyArray(list):
pass
def _zeros(shape, dtype=None):
size = 1
if isinstance(shape, int):
size = shape
elif shape:
size = 1
for dimension in shape:
size *= int(dimension)
return [0.0] * size
setattr(numpy_stub, "ndarray", _DummyArray)
setattr(numpy_stub, "zeros", _zeros)
setattr(numpy_stub, "float32", "float32")
setattr(numpy_stub, "array", lambda data, dtype=None: data)
setattr(numpy_stub, "asarray", lambda data, dtype=None: data)
setattr(
numpy_stub,
"concatenate",
lambda seq, axis=0: sum((list(item) for item in seq), []),
)
sys.modules["numpy"] = numpy_stub
if "soundfile" not in sys.modules:
soundfile_stub = types.ModuleType("soundfile")
class _DummySoundFile:
def __init__(self, *_, **__):
pass
def write(self, *_args, **_kwargs):
return None
def close(self):
return None
setattr(soundfile_stub, "SoundFile", _DummySoundFile)
setattr(soundfile_stub, "write", lambda *_args, **_kwargs: None)
sys.modules["soundfile"] = soundfile_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 _DummyMarkdown:
def __init__(self, *_, **__):
pass
def convert(self, text: str) -> str:
return text
setattr(markdown_stub, "Markdown", _DummyMarkdown)
sys.modules["markdown"] = markdown_stub
if "bs4" not in sys.modules:
bs4_stub = types.ModuleType("bs4")
class _DummySoup:
def __init__(self, *_, **__):
pass
def select(self, *_, **__):
return []
def find_all(self, *_, **__):
return []
setattr(bs4_stub, "BeautifulSoup", _DummySoup)
setattr(bs4_stub, "NavigableString", str)
sys.modules["bs4"] = bs4_stub
_install_dependency_stubs()
from abogen.text_extractor import ExtractedChapter
from abogen.webui.conversion_runner import _apply_chapter_overrides, _merge_metadata
def _sample_chapters() -> list[ExtractedChapter]:
return [
ExtractedChapter(title="Chapter 1", text="Original one"),
ExtractedChapter(title="Chapter 2", text="Original two"),
ExtractedChapter(title="Chapter 3", text="Original three"),
]
def test_apply_chapter_overrides_with_custom_text() -> None:
overrides = [
{"index": 0, "enabled": True, "title": "Intro", "text": "Hello world"},
{"index": 1, "enabled": False},
]
selected, metadata, diagnostics = _apply_chapter_overrides(
_sample_chapters(), overrides
)
assert len(selected) == 1
assert selected[0].title == "Intro"
assert selected[0].text == "Hello world"
assert overrides[0]["characters"] == len("Hello world")
assert metadata == {}
assert diagnostics == []
def test_apply_chapter_overrides_uses_original_content_when_text_missing() -> None:
overrides = [
{"index": 1, "enabled": True},
]
selected, metadata, diagnostics = _apply_chapter_overrides(
_sample_chapters(), overrides
)
assert len(selected) == 1
assert selected[0].title == "Chapter 2"
assert selected[0].text == "Original two"
assert overrides[0]["text"] == "Original two"
assert overrides[0]["characters"] == len("Original two")
assert metadata == {}
assert diagnostics == []
def test_apply_chapter_overrides_collects_metadata_updates() -> None:
overrides = [
{
"index": 2,
"enabled": True,
"metadata": {"artist": "Test Author", "year": 2024},
}
]
selected, metadata, diagnostics = _apply_chapter_overrides(
_sample_chapters(), overrides
)
assert len(selected) == 1
assert metadata == {"artist": "Test Author", "year": "2024"}
assert diagnostics == []
def test_apply_chapter_overrides_reports_diagnostics_for_invalid_payload() -> None:
overrides = [
{"enabled": True, "title": "Missing"},
]
selected, metadata, diagnostics = _apply_chapter_overrides(
_sample_chapters(), overrides
)
assert selected == []
assert metadata == {}
assert diagnostics and "Skipped chapter override" in diagnostics[0]
def test_merge_metadata_prefers_overrides_and_drops_none_values() -> None:
extracted = {"title": "Original", "artist": "Someone"}
overrides = {"artist": "Another", "genre": "Fiction", "year": None}
merged = _merge_metadata(extracted, overrides)
assert merged["title"] == "Original"
assert merged["artist"] == "Another"
assert merged["genre"] == "Fiction"
assert "year" not in merged
-500
View File
@@ -1,500 +0,0 @@
"""Tests for conversion adapters (WebUI + PyQt).
Covers field mapping, event bridging, voice resolution, and cancellation behavior.
"""
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from abogen.application.conversion_request import ConversionRequest
from abogen.application.conversion_config import ChapterChunkConfig, PronunciationConfig
from abogen.application.conversion_ports import ResolvedVoice
# ─── WebUI adapter tests ───────────────────────────────────────────
class TestWebUIAdapter:
"""Test WebUI conversion adapter field mapping."""
def _make_job(self, **overrides):
"""Create a mock WebUI Job with default values."""
defaults = dict(
stored_path="/tmp/test.epub",
original_filename="test.epub",
language="a",
tts_provider="kokoro",
voice="M1",
voice_profile=None,
speed=1.0,
use_gpu=False,
supertonic_total_steps=5,
output_format="wav",
subtitle_mode="Disabled",
subtitle_format="srt",
max_subtitle_words=50,
save_mode="save_next_to_input",
output_folder=None,
save_chapters_separately=False,
merge_chapters_at_end=True,
separate_chapters_format="wav",
save_as_project=False,
silence_between_chapters=2.0,
chapter_intro_delay=0.0,
replace_single_newlines=False,
read_title_intro=False,
read_closing_outro=False,
auto_prefix_chapter_titles=True,
normalize_chapter_opening_caps=False,
pronunciation_overrides=[],
manual_overrides=[],
heteronym_overrides=[],
normalization_overrides={},
chapters=[],
chunks=[],
chunk_level="paragraph",
speaker_mode="single",
speakers={},
metadata_tags={},
cover_image_path=None,
cover_image_mime=None,
generate_epub3=False,
cancel_requested=False,
)
defaults.update(overrides)
return SimpleNamespace(**defaults)
def test_basic_field_mapping(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
job = self._make_job()
req = build_conversion_request_from_job(job)
assert isinstance(req, ConversionRequest)
assert req.source_path == Path("/tmp/test.epub")
assert req.original_filename == "test.epub"
assert req.language == "a"
assert req.voice == "M1"
assert req.speed == 1.0
assert req.output_format == "wav"
def test_optional_fields_mapped(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
job = self._make_job(
voice_profile="custom_profile",
output_folder="/output",
cover_image_path="/cover.jpg",
cover_image_mime="image/jpeg",
metadata_tags={"title": "Test"},
)
req = build_conversion_request_from_job(job)
assert req.voice_profile == "custom_profile"
assert req.output_folder == Path("/output")
assert req.cover_image_path == Path("/cover.jpg")
assert req.cover_image_mime == "image/jpeg"
assert req.metadata_tags == {"title": "Test"}
def test_none_paths_result_in_none(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
job = self._make_job(
stored_path=None,
output_folder=None,
cover_image_path=None,
)
req = build_conversion_request_from_job(job)
assert req.source_path is None
assert req.output_folder is None
assert req.cover_image_path is None
def test_chapter_chunk_mapped(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
chapters = [{"title": "Ch1", "voice": "F1"}]
chunks = [{"text": "Hello", "speaker": "A"}]
job = self._make_job(chapters=chapters, chunks=chunks)
req = build_conversion_request_from_job(job)
assert isinstance(req.chapter_chunk, ChapterChunkConfig)
assert req.chapter_chunk.chapter_overrides == chapters
assert req.chapter_chunk.chunks == chunks
assert req.chapter_chunk.chunk_level == "paragraph"
assert req.chapter_chunk.speaker_mode == "single"
assert req.chapter_chunk.speakers == {}
def test_pronunciation_mapped(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
job = self._make_job(
pronunciation_overrides=["word=pron"],
manual_overrides=["manual=override"],
heteronym_overrides=["read=reed"],
)
req = build_conversion_request_from_job(job)
assert isinstance(req.pronunciation, PronunciationConfig)
assert req.pronunciation.pronunciation_overrides == ["word=pron"]
assert req.pronunciation.manual_overrides == ["manual=override"]
assert req.pronunciation.heteronym_overrides == ["read=reed"]
def test_none_defaults_handled(self):
from abogen.webui.conversion_adapter import build_conversion_request_from_job
from abogen.domain.enums import Language, OutputFormat, SubtitleMode, SaveMode
job = self._make_job(
language=None,
voice=None,
speed=None,
output_format=None,
subtitle_mode=None,
save_mode=None,
silence_between_chapters=None,
chapter_intro_delay=None,
supertonic_total_steps=None,
max_subtitle_words=None,
)
req = build_conversion_request_from_job(job)
# None values pass through adapter; ConversionRequest.__post_init__
# applies defaults and clamping for numeric fields.
assert req.language == Language.EN_US
assert req.speed == 1.0
assert req.output_format == OutputFormat.WAV
assert req.subtitle_mode == SubtitleMode.DISABLED
assert req.save_mode == SaveMode.SAVE_NEXT_TO_INPUT
assert req.silence_between_chapters == 2.0
assert req.chapter_intro_delay == 0.0
assert req.supertonic_total_steps == 5
assert req.max_subtitle_words == 50
class TestWebUIEvents:
"""Test WebUI ConversionEvents implementation."""
def test_log_calls_add_log(self):
from abogen.webui.conversion_adapter import WebJobEvents
job = SimpleNamespace(add_log=MagicMock())
events = WebJobEvents(job)
events.log("test message", level="info")
job.add_log.assert_called_once_with("test message", level="info")
def test_progress_updates_job(self):
from abogen.webui.conversion_adapter import WebJobEvents
job = SimpleNamespace(progress=0.0, etr_str="")
events = WebJobEvents(job)
events.progress(50, "2m 30s")
assert job.progress == 0.5
assert job.etr_str == "2m 30s"
def test_check_cancelled_raises(self):
from abogen.webui.conversion_adapter import ConversionCancelled, WebJobEvents
job = SimpleNamespace(cancel_requested=True)
events = WebJobEvents(job)
with pytest.raises(ConversionCancelled):
events.check_cancelled()
def test_check_not_cancelled_passes(self):
from abogen.webui.conversion_adapter import WebJobEvents
job = SimpleNamespace(cancel_requested=False)
events = WebJobEvents(job)
events.check_cancelled() # Should not raise
class TestWebUIPipelineProvider:
"""Test WebUI PipelineProvider implementation."""
def test_get_returns_backend(self):
from abogen.webui.conversion_adapter import WebPipelineProvider
backend = MagicMock()
pool = SimpleNamespace(get=MagicMock(return_value=backend))
provider = WebPipelineProvider(pool)
result = provider.get("kokoro", "a", False)
assert result is backend
pool.get.assert_called_once_with("kokoro", "a", False)
class TestWebUIVoiceResolver:
"""Test WebUI VoiceResolver implementation."""
def test_resolve_returns_resolved_voice(self):
from abogen.webui.conversion_adapter import WebVoiceResolver
def resolve_fn(spec):
return ("kokoro", spec, "M1", 1.0, 5)
resolver = WebVoiceResolver(resolve_fn)
result = resolver.resolve("M1")
assert isinstance(result, ResolvedVoice)
assert result.provider == "kokoro"
assert result.voice == "M1"
assert result.speed == 1.0
assert result.supertonic_steps == 5
def test_resolve_none_speed_defaults(self):
from abogen.webui.conversion_adapter import WebVoiceResolver
def resolve_fn(spec):
return ("kokoro", spec, "M1", None, None)
resolver = WebVoiceResolver(resolve_fn)
result = resolver.resolve("M1")
assert result.speed == 1.0
assert result.supertonic_steps == 5
# ─── PyQt adapter tests ────────────────────────────────────────────
class TestPyQtAdapter:
"""Test PyQt conversion adapter field mapping."""
def _make_thread(self, **overrides):
"""Create a mock ConversionThread with default values."""
defaults = dict(
file_name="/tmp/test.epub",
lang_code="a",
voice="M1",
voice_profile=None,
speed=1.0,
use_gpu=False,
supertonic_total_steps=5,
output_format="wav",
subtitle_mode="Disabled",
subtitle_format="srt",
max_subtitle_words=50,
save_option="save_next_to_input",
output_folder=None,
save_chapters_separately=False,
merge_chapters_at_end=True,
separate_chapters_format="wav",
save_as_project=False,
silence_duration=2.0,
chapter_intro_delay=0.0,
replace_single_newlines=False,
read_title_intro=False,
read_closing_outro=True,
auto_prefix_chapter_titles=True,
normalize_chapter_opening_caps=False,
pronunciation_overrides=[],
manual_overrides=[],
heteronym_overrides=[],
normalization_overrides=None,
metadata_tags={},
cover_image_path=None,
cover_image_mime=None,
generate_epub3=False,
is_direct_text=False,
from_queue=False,
display_path=None,
save_base_path=None,
cancel_requested=False,
)
defaults.update(overrides)
return SimpleNamespace(**defaults)
def test_basic_field_mapping(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread()
req = build_conversion_request_from_thread(thread)
assert isinstance(req, ConversionRequest)
assert req.source_path == Path("/tmp/test.epub")
assert req.language == "a"
assert req.voice == "M1"
assert req.speed == 1.0
assert req.output_format == "wav"
def test_direct_text_mode(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread(
is_direct_text=True,
file_name="Hello world",
)
req = build_conversion_request_from_thread(thread)
assert req.source_path is None
assert req.direct_text == "Hello world"
def test_from_queue_uses_save_base_path(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread(
from_queue=True,
save_base_path="/queue/book.epub",
display_path="/display/book.epub",
)
req = build_conversion_request_from_thread(thread)
assert req.original_filename == "book.epub"
def test_display_path_used_when_not_from_queue(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread(
from_queue=False,
display_path="/display/book.epub",
save_base_path="/queue/book.epub",
)
req = build_conversion_request_from_thread(thread)
assert req.original_filename == "book.epub"
def test_output_folder_mapped(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread(output_folder="/output")
req = build_conversion_request_from_thread(thread)
assert req.output_folder == Path("/output")
def test_none_defaults_handled(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
from abogen.domain.enums import Language, OutputFormat, SubtitleMode, SaveMode
thread = self._make_thread(
lang_code=None,
voice=None,
speed=None,
output_format=None,
subtitle_mode=None,
save_option=None,
silence_duration=None,
chapter_intro_delay=None,
supertonic_total_steps=None,
max_subtitle_words=None,
)
req = build_conversion_request_from_thread(thread)
# None values pass through adapter; ConversionRequest.__post_init__
# applies defaults and clamping for numeric fields.
assert req.language == Language.EN_US
assert req.speed == 1.0
assert req.output_format == OutputFormat.WAV
assert req.subtitle_mode == SubtitleMode.DISABLED
assert req.save_mode == SaveMode.SAVE_NEXT_TO_INPUT
assert req.silence_between_chapters == 2.0
assert req.chapter_intro_delay == 0.0
assert req.supertonic_total_steps == 5
assert req.max_subtitle_words == 50
def test_chapter_chunk_default(self):
from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread
thread = self._make_thread()
req = build_conversion_request_from_thread(thread)
assert isinstance(req.chapter_chunk, ChapterChunkConfig)
assert req.chapter_chunk.chapter_overrides == []
assert req.chapter_chunk.chunks == []
assert req.chapter_chunk.chunk_level == "paragraph"
assert req.chapter_chunk.speaker_mode == "single"
assert req.chapter_chunk.speakers == {}
class TestPyQtEvents:
"""Test PyQt ConversionEvents implementation."""
def test_log_emits_signal(self):
from abogen.pyqt.conversion_adapter import PyQtEvents
thread = SimpleNamespace(
log_updated=MagicMock(),
)
events = PyQtEvents(thread)
events.log("test message", level="info")
thread.log_updated.emit.assert_called_once()
def test_progress_emits_signal(self):
from abogen.pyqt.conversion_adapter import PyQtEvents
thread = SimpleNamespace(
progress_updated=MagicMock(),
)
events = PyQtEvents(thread)
events.progress(50, "2m 30s")
thread.progress_updated.emit.assert_called_once_with(50, "2m 30s")
def test_check_cancelled_raises(self):
from abogen.pyqt.conversion_adapter import ConversionCancelled, PyQtEvents
thread = SimpleNamespace(cancel_requested=True)
events = PyQtEvents(thread)
with pytest.raises(ConversionCancelled):
events.check_cancelled()
def test_check_not_cancelled_passes(self):
from abogen.pyqt.conversion_adapter import PyQtEvents
thread = SimpleNamespace(cancel_requested=False)
events = PyQtEvents(thread)
events.check_cancelled() # Should not raise
class TestPyQtPipelineProvider:
"""Test PyQt PipelineProvider implementation."""
def test_get_returns_backend(self):
from abogen.pyqt.conversion_adapter import PyQtPipelineProvider
backend = MagicMock()
provider = PyQtPipelineProvider(backend)
result = provider.get("kokoro", "a", False)
assert result is backend
def test_dispose_all_noop(self):
from abogen.pyqt.conversion_adapter import PyQtPipelineProvider
backend = MagicMock()
provider = PyQtPipelineProvider(backend)
provider.dispose_all() # Should not raise
class TestPyQtVoiceResolver:
"""Test PyQt VoiceResolver implementation."""
def test_resolve_returns_resolved_voice(self):
from abogen.pyqt.conversion_adapter import PyQtVoiceResolver
loaded_voice = MagicMock()
thread = SimpleNamespace(
load_voice_cached=MagicMock(return_value=loaded_voice),
backend=MagicMock(),
speed=1.0,
supertonic_total_steps=5,
)
resolver = PyQtVoiceResolver(thread)
result = resolver.resolve("M1")
assert isinstance(result, ResolvedVoice)
assert result.provider == "kokoro"
assert result.voice is loaded_voice
assert result.speed == 1.0
assert result.supertonic_steps == 5
-240
View File
@@ -1,240 +0,0 @@
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 abogen.webui.conversion_runner import (
_format_spoken_chapter_title,
_headings_equivalent,
_normalize_chapter_opening_caps,
_strip_duplicate_heading_line,
)
def test_format_spoken_chapter_title_adds_prefix() -> None:
assert _format_spoken_chapter_title("1: A Tale", 1, True) == "Chapter 1. A Tale"
def test_format_spoken_chapter_title_respects_existing_prefix() -> None:
assert (
_format_spoken_chapter_title("Chapter 2: Story", 2, True) == "Chapter 2: Story"
)
def test_format_spoken_chapter_title_handles_empty_title() -> None:
assert _format_spoken_chapter_title("", 4, True) == "Chapter 4"
def test_format_spoken_chapter_title_trims_delimiters() -> None:
assert (
_format_spoken_chapter_title("7 - Into the Wild", 7, True)
== "Chapter 7. Into the Wild"
)
def test_headings_equivalent_ignores_case_and_prefix() -> None:
assert _headings_equivalent("1: The House", "Chapter 1: The House")
def test_strip_duplicate_heading_line_removes_first_match() -> None:
text, removed = _strip_duplicate_heading_line(
"Chapter 3: Intro\nBody text", "Chapter 3: Intro"
)
assert removed is True
assert text.strip() == "Body text"
def test_normalize_chapter_opening_caps_basic_title() -> None:
normalized, changed = _normalize_chapter_opening_caps("ALL CAPS TITLE")
assert normalized == "All Caps Title"
assert changed is True
def test_normalize_chapter_opening_caps_respects_acronyms() -> None:
normalized, changed = _normalize_chapter_opening_caps("NASA MISSION LOG")
assert normalized == "NASA Mission Log"
assert changed is True
def test_normalize_chapter_opening_caps_handles_roman_numerals() -> None:
normalized, changed = _normalize_chapter_opening_caps("IV. THE RETURN")
assert normalized == "IV. The Return"
assert changed is True
def test_normalize_chapter_opening_caps_keeps_mixed_case() -> None:
normalized, changed = _normalize_chapter_opening_caps("Already Mixed Case")
assert normalized == "Already Mixed Case"
assert changed is False
class TestApplyChapterTextTransforms:
"""Tests for the combined heading-strip + opening-caps helper."""
def test_both_enabled_heading_matches(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"Chapter 1: The Beginning\nBody text here",
heading_text="Chapter 1: The Beginning",
raw_title="Chapter 1: The Beginning",
strip_heading=True,
normalize_caps=True,
)
assert heading_removed is True
assert "Body text here" in text
assert "Chapter 1" not in text
def test_heading_fallback_to_number(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"1. The Beginning\nBody text",
heading_text="Chapter 1: The Beginning",
raw_title="1: The Beginning",
strip_heading=True,
normalize_caps=False,
)
assert heading_removed is True
assert "Body text" in text
def test_only_heading_strip(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"Chapter 1: Title\nBody text",
heading_text="Chapter 1: Title",
raw_title="",
strip_heading=True,
normalize_caps=False,
)
assert heading_removed is True
assert caps_changed is False
def test_only_opening_caps(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"ALL CAPS START OF CHAPTER",
heading_text="Chapter 1",
raw_title="",
strip_heading=False,
normalize_caps=True,
)
assert heading_removed is False
assert caps_changed is True
assert text == "All Caps Start Of Chapter"
def test_both_disabled_no_change(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
original = "Some text here"
text, heading_removed, caps_changed = apply_chapter_text_transforms(
original,
heading_text="Chapter 1",
raw_title="",
strip_heading=False,
normalize_caps=False,
)
assert text == original
assert heading_removed is False
assert caps_changed is False
def test_heading_not_matching(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"Completely different text",
heading_text="Chapter 1: Title",
raw_title="",
strip_heading=True,
normalize_caps=False,
)
assert heading_removed is False
assert text == "Completely different text"
def test_empty_text(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"",
heading_text="Chapter 1",
raw_title="",
strip_heading=True,
normalize_caps=True,
)
assert text == ""
assert heading_removed is False
assert caps_changed is False
def test_both_enabled_text_only_has_caps(self) -> None:
from abogen.domain.chapter_titles import apply_chapter_text_transforms
text, heading_removed, caps_changed = apply_chapter_text_transforms(
"NASA MISSION LOG",
heading_text="Chapter 1",
raw_title="",
strip_heading=True,
normalize_caps=True,
)
assert heading_removed is False
assert caps_changed is True
assert text == "NASA Mission Log"
-120
View File
@@ -1,120 +0,0 @@
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 abogen.webui.conversion_runner import _build_outro_text, _build_title_intro_text
def test_title_intro_includes_series_sentence() -> None:
metadata = {
"title": "Galactic Chronicles",
"author": "Jane Doe",
"series": "Chronicles",
"series_index": "2",
}
intro_text = _build_title_intro_text(metadata, "chronicles.mp3")
assert intro_text.startswith("Book 2 of the Chronicles.")
assert "Galactic Chronicles." in intro_text
assert "By Jane Doe." in intro_text
def test_series_sentence_skips_duplicate_article() -> None:
metadata = {
"title": "Iron Council",
"authors": "China Miéville",
"series": "The Bas-Lag",
"series_index": "3",
}
intro_text = _build_title_intro_text(metadata, "iron_council.mp3")
assert "Book 3 of The Bas-Lag." in intro_text
assert "of the The" not in intro_text
def test_outro_appends_series_information() -> None:
metadata = {
"title": "Abaddon's Gate",
"authors": "James S. A. Corey",
"series": "The Expanse",
"series_index": "3",
}
outro_text = _build_outro_text(metadata, "abaddon.mp3")
assert outro_text.startswith("The end of Abaddon's Gate from James S. A. Corey.")
assert outro_text.endswith("Book 3 of The Expanse.")
def test_series_number_preserves_decimal_positions() -> None:
metadata = {
"title": "Interlude",
"author": "Alex Writer",
"series": "Chronicles",
"series_index": "2.5",
}
intro_text = _build_title_intro_text(metadata, "interlude.mp3")
assert "Book 2.5 of the Chronicles." in intro_text
-52
View File
@@ -1,52 +0,0 @@
from types import SimpleNamespace
from typing import cast
from abogen.tts_plugin.utils import get_voices
from abogen.webui.conversion_runner import (
_chapter_voice_spec,
_chunk_voice_spec,
_collect_required_voice_ids,
)
from abogen.webui.service import Job
def _sample_job(formula: str) -> Job:
return cast(
Job,
SimpleNamespace(
voice="__custom_mix",
speakers={
"narrator": {
"resolved_voice": formula,
}
},
chapters=[],
chunks=[{}],
),
)
def test_chapter_voice_spec_uses_resolved_formula():
formula = "af_nova*0.7+am_liam*0.3"
job = _sample_job(formula)
assert _chapter_voice_spec(job, None) == formula
def test_chunk_voice_fallback_uses_resolved_formula():
formula = "af_nova*0.7+am_liam*0.3"
job = _sample_job(formula)
result = _chunk_voice_spec(job, {}, "")
assert result == formula
def test_voice_collection_includes_formula_components():
formula = "af_nova*0.7+am_liam*0.3"
job = _sample_job(formula)
voices = _collect_required_voice_ids(job)
assert {"af_nova", "am_liam"}.issubset(voices)
assert voices.issuperset(get_voices("kokoro"))
-221
View File
@@ -1,221 +0,0 @@
"""Import/layering tests for the conversion flow unification.
Verifies that:
- Application layer does not import from PyQt or WebUI
- Adapters import from application/domain (not the other way around)
- All application models are importable without GUI/Flask side effects
"""
import importlib
import sys
from pathlib import Path
import pytest
# ─── Application layer imports ─────────────────────────────────────
class TestApplicationLayerImports:
"""Verify application layer has no PyQt/WebUI imports."""
APPLICATION_DIR = Path(__file__).parent.parent / "abogen" / "application"
def _get_python_files(self):
"""Get all Python files in the application directory."""
return list(self.APPLICATION_DIR.glob("*.py"))
def test_no_pyqt_imports(self):
"""Application layer must not import from abogen.pyqt."""
import re
forbidden = re.compile(r"from\s+abogen\.pyqt|import\s+abogen\.pyqt")
violations = []
for py_file in self._get_python_files():
content = py_file.read_text(encoding="utf-8")
if forbidden.search(content):
violations.append(py_file.name)
assert not violations, f"Application files import from PyQt: {violations}"
def test_no_webui_imports(self):
"""Application layer must not import from abogen.webui."""
import re
forbidden = re.compile(r"from\s+abogen\.webui|import\s+abogen\.webui")
violations = []
for py_file in self._get_python_files():
content = py_file.read_text(encoding="utf-8")
if forbidden.search(content):
violations.append(py_file.name)
assert not violations, f"Application files import from WebUI: {violations}"
def test_no_flask_imports(self):
"""Application layer must not import Flask."""
import re
forbidden = re.compile(r"from\s+flask|import\s+flask")
violations = []
for py_file in self._get_python_files():
content = py_file.read_text(encoding="utf-8")
if forbidden.search(content):
violations.append(py_file.name)
assert not violations, f"Application files import Flask: {violations}"
def test_no_qthread_imports(self):
"""Application layer must not import QThread."""
import re
forbidden = re.compile(r"from\s+PyQt6|import\s+PyQt6")
violations = []
for py_file in self._get_python_files():
content = py_file.read_text(encoding="utf-8")
if forbidden.search(content):
violations.append(py_file.name)
assert not violations, f"Application files import PyQt6: {violations}"
class TestApplicationModelsImportable:
"""Verify application models are importable without side effects."""
def test_conversion_request_importable(self):
from abogen.application.conversion_request import ConversionRequest
assert ConversionRequest is not None
def test_conversion_models_importable(self):
from abogen.application.conversion_models import (
ChapterPlan,
ConversionPlan,
IntroOutroSpec,
OutputLayout,
SegmentPlan,
)
assert all(
cls is not None
for cls in [ChapterPlan, ConversionPlan, IntroOutroSpec, OutputLayout, SegmentPlan]
)
def test_conversion_result_importable(self):
from abogen.application.conversion_result import ConversionError, ConversionResult
assert ConversionResult is not None
assert ConversionError is not None
def test_conversion_ports_importable(self):
from abogen.application.conversion_ports import (
AudioSink,
ConversionEvents,
PipelineProvider,
ResolvedVoice,
SubtitleWriter,
VoiceResolver,
)
assert all(
cls is not None
for cls in [
AudioSink,
ConversionEvents,
PipelineProvider,
ResolvedVoice,
SubtitleWriter,
VoiceResolver,
]
)
def test_output_layout_service_importable(self):
from abogen.application.output_layout_service import (
resolve_chapter_path,
resolve_merged_path,
resolve_output_layout,
should_merge_output,
)
assert all(
fn is not None
for fn in [resolve_chapter_path, resolve_merged_path, resolve_output_layout, should_merge_output]
)
def test_conversion_planner_importable(self):
from abogen.application.conversion_planner import build_conversion_plan
assert build_conversion_plan is not None
def test_conversion_executor_importable(self):
from abogen.application.conversion_executor import execute_conversion
assert execute_conversion is not None
def test_conversion_service_importable(self):
from abogen.application.conversion_service import run_conversion
assert run_conversion is not None
class TestAdapterImports:
"""Verify adapters import from application/domain correctly."""
def test_webui_adapter_imports_application(self):
from abogen.webui.conversion_adapter import (
WebJobEvents,
WebPipelineProvider,
WebVoiceResolver,
build_conversion_request_from_job,
)
assert all(
cls is not None
for cls in [
WebJobEvents,
WebPipelineProvider,
WebVoiceResolver,
build_conversion_request_from_job,
]
)
def test_pyqt_adapter_imports_application(self):
from abogen.pyqt.conversion_adapter import (
PyQtEvents,
PyQtPipelineProvider,
PyQtVoiceResolver,
build_conversion_request_from_thread,
)
assert all(
cls is not None
for cls in [
PyQtEvents,
PyQtPipelineProvider,
PyQtVoiceResolver,
build_conversion_request_from_thread,
]
)
def test_webui_adapter_does_not_import_pyqt(self):
"""WebUI adapter must not import from PyQt."""
import re
adapter_path = Path(__file__).parent.parent / "abogen" / "webui" / "conversion_adapter.py"
content = adapter_path.read_text(encoding="utf-8")
forbidden = re.compile(r"from\s+abogen\.pyqt|import\s+abogen\.pyqt")
assert not forbidden.search(content), "WebUI adapter imports from PyQt"
def test_pyqt_adapter_does_not_import_webui(self):
"""PyQt adapter must not import from WebUI."""
import re
adapter_path = Path(__file__).parent.parent / "abogen" / "pyqt" / "conversion_adapter.py"
content = adapter_path.read_text(encoding="utf-8")
forbidden = re.compile(r"from\s+abogen\.webui|import\s+abogen\.webui")
assert not forbidden.search(content), "PyQt adapter imports from WebUI"
@@ -1,51 +0,0 @@
from abogen.webui import conversion_runner
class DummyJob:
def __init__(self):
self.language = "en"
self.voice = "M1"
self.speakers = None
self.manual_overrides = []
self.pronunciation_overrides = []
def _apply(text: str, job: DummyJob) -> str:
merged = conversion_runner._merge_pronunciation_overrides(job)
rules = conversion_runner._compile_pronunciation_rules(merged)
return conversion_runner._apply_pronunciation_rules(text, rules)
def test_manual_override_is_applied_even_if_pronunciation_overrides_stale():
job = DummyJob()
job.manual_overrides = [
{
"token": "Unfu*k",
"pronunciation": "Unfuck",
}
]
out = _apply("He said Unfu*k loudly.", job)
assert "Unfuck" in out
assert "Unfu*k" not in out
def test_manual_override_takes_precedence_over_existing_pronunciation_override():
job = DummyJob()
job.pronunciation_overrides = [
{
"token": "Unfu*k",
"normalized": "unfu*k",
"pronunciation": "WRONG",
}
]
job.manual_overrides = [
{
"token": "Unfu*k",
"pronunciation": "RIGHT",
}
]
out = _apply("Unfu*k.", job)
assert "RIGHT" in out
assert "WRONG" not in out
-69
View File
@@ -1,69 +0,0 @@
from types import SimpleNamespace
from typing import cast
import pytest
from abogen.tts_plugin.utils import get_voices
from abogen.voice_cache import (
LocalEntryNotFoundError,
_CACHED_VOICES,
ensure_voice_assets,
)
from abogen.webui.conversion_runner import _collect_required_voice_ids
from abogen.webui.service import Job
@pytest.fixture(autouse=True)
def clear_voice_cache():
_CACHED_VOICES.clear()
yield
_CACHED_VOICES.clear()
def test_ensure_voice_assets_downloads_missing(monkeypatch):
recorded = []
cached = set()
def fake_download(**kwargs):
filename = kwargs["filename"]
if kwargs.get("local_files_only"):
if filename in cached:
return f"/tmp/{filename}"
raise LocalEntryNotFoundError(f"{filename} missing")
recorded.append(filename)
cached.add(filename)
return f"/tmp/{filename}"
monkeypatch.setattr("abogen.voice_cache.hf_hub_download", fake_download)
downloaded, errors = ensure_voice_assets(["af_nova", "am_liam"])
assert downloaded == {"af_nova", "am_liam"}
assert errors == {}
assert set(recorded) == {"voices/af_nova.pt", "voices/am_liam.pt"}
recorded.clear()
downloaded_again, errors_again = ensure_voice_assets(["af_nova"])
assert downloaded_again == set()
assert errors_again == {}
assert recorded == []
def test_collect_required_voice_ids_includes_all():
job = SimpleNamespace(
voice="af_nova",
chapters=[{"voice_formula": "af_nova*0.7+am_liam*0.3"}],
chunks=[{"voice": "am_michael"}],
speakers={
"hero": {"voice_formula": "af_nova*0.6+am_liam*0.4"},
"narrator": {"voice": "af_nova"},
},
)
voices = _collect_required_voice_ids(cast(Job, job))
assert {"af_nova", "am_liam", "am_michael"}.issubset(voices)
assert voices.issuperset(get_voices("kokoro"))