diff --git a/tests/test_conversion_adapters.py b/tests/test_conversion_adapters.py new file mode 100644 index 0000000..a65151d --- /dev/null +++ b/tests/test_conversion_adapters.py @@ -0,0 +1,496 @@ +"""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_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_overrides_mapped(self): + from abogen.webui.conversion_adapter import build_conversion_request_from_job + + chapters = [{"title": "Ch1", "voice": "F1"}] + job = self._make_job(chapters=chapters) + req = build_conversion_request_from_job(job) + + assert req.chapter_overrides == chapters + + def test_chunks_mapped(self): + from abogen.webui.conversion_adapter import build_conversion_request_from_job + + chunks = [{"text": "Hello", "speaker": "A"}] + job = self._make_job(chunks=chunks) + req = build_conversion_request_from_job(job) + + assert req.chunks == chunks + + def test_pronunciation_overrides_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 req.pronunciation_overrides == ["word=pron"] + assert req.manual_overrides == ["manual=override"] + assert req.heteronym_overrides == ["read=reed"] + + def test_none_defaults_handled(self): + from abogen.webui.conversion_adapter import build_conversion_request_from_job + + 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) + + assert req.language == "a" + assert req.voice == "M1" + assert req.speed == 1.0 + assert req.output_format == "wav" + assert req.subtitle_mode == "Disabled" + assert req.save_mode == "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 + + 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) + + assert req.language == "a" + assert req.voice == "M1" + assert req.speed == 1.0 + assert req.output_format == "wav" + assert req.subtitle_mode == "Disabled" + assert req.save_mode == "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_chunks_not_mapped(self): + from abogen.pyqt.conversion_adapter import build_conversion_request_from_thread + + thread = self._make_thread() + req = build_conversion_request_from_thread(thread) + + assert req.chapter_overrides == [] + assert req.chunks == [] + assert req.chunk_level == "paragraph" + assert req.speaker_mode == "single" + assert req.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 diff --git a/tests/test_import_layering.py b/tests/test_import_layering.py new file mode 100644 index 0000000..c6d3176 --- /dev/null +++ b/tests/test_import_layering.py @@ -0,0 +1,221 @@ +"""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"