test: adapter field mapping + import layering tests

- 28 adapter tests (WebUI + PyQt): field mapping, events, provider, resolver - 15 import/layering tests: no PyQt/WebUI in app layer, all models importable
2026-09-20 11:40:57 +02:00 · 2026-07-21 11:47:46 +00:00
parent d0e42ee691
commit 7f317ca784
2 changed files with 717 additions and 0 deletions
@@ -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
@@ -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"