diff options
Diffstat (limited to 'app/tests')
| -rw-r--r-- | app/tests/test_backends_envs.py | 370 | ||||
| -rw-r--r-- | app/tests/test_extractors.py | 24 |
2 files changed, 388 insertions, 6 deletions
diff --git a/app/tests/test_backends_envs.py b/app/tests/test_backends_envs.py index 82d903a..184a3b3 100644 --- a/app/tests/test_backends_envs.py +++ b/app/tests/test_backends_envs.py @@ -1,5 +1,6 @@ """Tests for the managed Python environment (backends/envs.py).""" +import json import sys import unittest from pathlib import Path @@ -181,14 +182,113 @@ class ModuleAvailableTests(unittest.TestCase): self.assertFalse(envs.module_available("qwen_tts")) +class RequirementSpecsTests(unittest.TestCase): + """requirements.txt parsing (specs, optional tags, markers).""" + + SAMPLE = ( + "# Core dependencies\n" + "gradio_client>=0.7.0\n" + "\n" + "pypdf\n" + "beautifulsoup4>=4.11.0 # optional: HTML cleaning (stdlib fallback)\n" + "faster-whisper>=1.0.0 # optional: transcription\n" + "windows-curses>=2.3; sys_platform == \"win32\" # TUI on Windows\n" + "-e ./local\n" + "--extra-index-url https://example.com/simple\n" + ) + + def setUp(self): + import tempfile + self._tmp = tempfile.TemporaryDirectory() + self.path = Path(self._tmp.name) / "requirements.txt" + self.addCleanup(self._tmp.cleanup) + + def _patch_path(self, content): + self.path.write_text(content, encoding="utf-8") + return patch.object(envs, "REQUIREMENTS_PATH", self.path) + + def test_parses_specs_and_optional_tags(self): + with self._patch_path(self.SAMPLE): + specs = envs.requirement_specs() + # posix host: the win32-marker line is dropped, option lines ignored, + # comments stripped, version specifiers kept verbatim. + self.assertEqual(specs, [ + ("gradio_client>=0.7.0", False), + ("pypdf", False), + ("beautifulsoup4>=4.11.0", True), + ("faster-whisper>=1.0.0", True), + ]) + + def test_win32_marker_applies_on_windows(self): + with self._patch_path(self.SAMPLE), \ + patch.object(envs, "_is_windows", return_value=True): + names = [spec for spec, _ in envs.requirement_specs()] + self.assertIn("windows-curses>=2.3", names) + + def test_missing_file_yields_nothing(self): + with patch.object(envs, "REQUIREMENTS_PATH", + Path("/no/such/requirements.txt")): + self.assertEqual(envs.requirement_specs(), []) + + +class MarkerTests(unittest.TestCase): + """The hash + version marker that gates re-installation.""" + + def setUp(self): + import tempfile + self._tmp = tempfile.TemporaryDirectory() + req = Path(self._tmp.name) / "requirements.txt" + req.write_bytes(b"pypdf\n") + marker = Path(self._tmp.name) / ".audiobook_env_ready" + self.addCleanup(self._tmp.cleanup) + self.patches = [patch.object(envs, "REQUIREMENTS_PATH", req), + patch.object(envs, "MARKER_PATH", marker)] + for p in self.patches: + p.start() + self.addCleanup(p.stop) + self.marker = marker + + def test_valid_marker_matches_hash_and_version(self): + self.marker.write_text(f"{envs._requirements_sha()}:2\n", + encoding="utf-8") + with patch.object(envs, "MARKER_VERSION", "2"): + self.assertTrue(envs._marker_valid()) + + def test_version_mismatch_invalidates_marker(self): + self.marker.write_text(f"{envs._requirements_sha()}:1\n", + encoding="utf-8") + self.assertFalse(envs._marker_valid()) + + def test_legacy_bare_hash_marker_is_invalid(self): + # Markers written by tool versions before MARKER_VERSION existed. + self.marker.write_text(envs._requirements_sha() + "\n", + encoding="utf-8") + self.assertFalse(envs._marker_valid()) + + def test_write_marker_records_current_hash_and_version(self): + envs._write_marker() + self.assertEqual(self.marker.read_text(encoding="utf-8"), + f"{envs._requirements_sha()}:{envs.MARKER_VERSION}\n") + + def test_missing_marker_is_invalid(self): + self.assertFalse(envs._marker_valid()) + + class EnsureAppEnvTests(unittest.TestCase): - def test_creates_env_then_installs_when_marker_invalid(self): + """Bootstrap: install, optional-fallback, verification, marker.""" + + def test_installs_verifies_then_writes_marker(self): with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs, "create_env", return_value=0), \ patch.object(envs, "_marker_valid", return_value=False), \ - patch.object(envs, "install_requirements", return_value=0), \ + patch.object(envs, "install_requirements", + return_value=0) as install, \ + patch.object(envs, "ensure_importable", + return_value=[]) as verify, \ patch.object(envs, "_write_marker") as mk: envs.ensure_app_env() + install.assert_called_once_with() + verify.assert_called_once_with() mk.assert_called_once_with() def test_raises_when_create_fails(self): @@ -197,19 +297,277 @@ class EnsureAppEnvTests(unittest.TestCase): with self.assertRaises(RuntimeError): envs.ensure_app_env() - def test_raises_when_install_fails(self): + def test_retries_without_optionals_when_install_fails(self): + specs = [("core-a", False), ("opt-b", True)] + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "requirement_specs", + return_value=specs), \ + patch.object(envs, "install_requirements", + side_effect=[1, 0]) as install, \ + patch.object(envs, "ensure_importable", return_value=[]), \ + patch.object(envs, "_write_marker"): + envs.ensure_app_env() + self.assertEqual(install.call_args_list[0], ()) + install.assert_any_call(skip_optional=True) + + def test_raises_when_both_install_attempts_fail(self): + specs = [("core-a", False), ("opt-b", True)] + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "requirement_specs", + return_value=specs), \ + patch.object(envs, "install_requirements", return_value=1), \ + patch.object(envs, "ensure_importable") as verify: + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + verify.assert_not_called() + + def test_raises_on_failure_without_optionals(self): + specs = [("core-a", False)] with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "requirement_specs", + return_value=specs), \ patch.object(envs, "install_requirements", return_value=1): with self.assertRaises(RuntimeError): envs.ensure_app_env() - def test_skips_install_when_marker_valid(self): + def test_skips_install_and_verify_when_marker_valid(self): with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs, "_marker_valid", return_value=True), \ - patch.object(envs, "install_requirements") as mk: + patch.object(envs, "install_requirements") as install, \ + patch.object(envs, "ensure_importable") as verify: envs.ensure_app_env() - mk.assert_not_called() + install.assert_not_called() + verify.assert_not_called() + + +class BrokenImportsTests(unittest.TestCase): + def test_empty_names_short_circuits_without_probe(self): + with patch.object(envs, "_imports_ok") as ok: + self.assertEqual(envs.broken_imports([], python=Path("/x/py")), []) + ok.assert_not_called() + + def test_healthy_combined_probe_skips_isolation(self): + with patch.object(envs, "_imports_ok", return_value=True) as ok: + self.assertEqual( + envs.broken_imports(["a", "b"], python=Path("/x/py")), []) + self.assertEqual(ok.call_count, 1) + + def test_isolates_each_broken_name(self): + def fake_ok(names, *, python=None): + # Only the good name imports cleanly; every other probe fails, + # including the combined short-circuit probe. + return names == ["good"] + + with patch.object(envs, "_imports_ok", side_effect=fake_ok): + broken = envs.broken_imports(["good", "bad"], python=Path("/p")) + self.assertEqual(broken, ["bad"]) + + +class ScannedBrokenImportsTests(unittest.TestCase): + @staticmethod + def _proc(stdout): + import subprocess + return subprocess.CompletedProcess(args=[], returncode=0, + stdout=stdout, stderr="") + + def test_parses_reported_failures(self): + with patch("subprocess.run", return_value=self._proc('["lxml"]\n')): + self.assertEqual(envs.scanned_broken_imports( + python=Path("/x/py")), ["lxml"]) + + def test_healthy_env_reports_no_failures(self): + with patch("subprocess.run", return_value=self._proc("[]\n")): + self.assertEqual( + envs.scanned_broken_imports(python=Path("/x/py")), []) + + def test_unparsable_output_returns_none(self): + with patch("subprocess.run", return_value=self._proc("")): + self.assertIsNone( + envs.scanned_broken_imports(python=Path("/x/py"))) + + def test_missing_interpreter_returns_none(self): + with patch("subprocess.run", side_effect=OSError("nope")): + self.assertIsNone( + envs.scanned_broken_imports(python=Path("/x/py"))) + + +class VenvTagsTests(unittest.TestCase): + @staticmethod + def _tags_for(suffix, vi=(3, 14)): + import subprocess + payload = json.dumps({"suffix": suffix, "vi": list(vi)}) + proc = subprocess.CompletedProcess(args=[], returncode=0, + stdout=payload + "\n", stderr="") + return patch("subprocess.run", return_value=proc) + + def test_musl_suffix(self): + with self._tags_for(".cpython-314-x86_64-linux-musl.so"): + tags = envs._venv_tags(Path("/x/py")) + self.assertEqual(tags, {"musl": True, "pyver": "3.14", "impl": "cp", + "abi": "cp314", "arch": "x86_64"}) + + def test_glibc_suffix(self): + with self._tags_for(".cpython-312-x86_64-linux-gnu.so", (3, 12)): + tags = envs._venv_tags(Path("/x/py")) + self.assertFalse(tags["musl"]) + self.assertEqual(tags["abi"], "cp312") + + def test_unparsable_suffix_returns_none(self): + with self._tags_for("weird"): + self.assertIsNone(envs._venv_tags(Path("/x/py"))) + + def test_dead_interpreter_returns_none(self): + with patch("subprocess.run", side_effect=OSError("nope")): + self.assertIsNone(envs._venv_tags(Path("/x/py"))) + + +class InstalledSpecsTests(unittest.TestCase): + @staticmethod + def _proc(payload): + import subprocess + return subprocess.CompletedProcess(args=[], returncode=0, + stdout=payload, stderr="") + + def test_pins_exact_name_and_version(self): + payload = json.dumps([{"name": "BeautifulSoup4", "version": "4.15.0"}, + {"name": "lxml", "version": "6.1.2"}]) + with patch("subprocess.run", return_value=self._proc(payload)): + specs = envs._installed_specs(Path("/x/py")) + self.assertEqual(specs.get("beautifulsoup4"), + "BeautifulSoup4==4.15.0") + self.assertEqual(specs.get("lxml"), "lxml==6.1.2") + + def test_bad_output_yields_no_specs(self): + with patch("subprocess.run", return_value=self._proc("garbage")): + self.assertEqual(envs._installed_specs(Path("/x/py")), {}) + + +class SpecForImportTests(unittest.TestCase): + INSTALLED = {"beautifulsoup4": "beautifulsoup4==4.15.0", + "lxml": "lxml==6.1.2"} + + def test_direct_distribution_hit(self): + self.assertEqual(envs._spec_for_import("lxml", self.INSTALLED), + "lxml==6.1.2") + + def test_mapped_import_name(self): + self.assertEqual(envs._spec_for_import("bs4", self.INSTALLED), + "beautifulsoup4==4.15.0") + + def test_unknown_import_has_no_spec(self): + self.assertIsNone(envs._spec_for_import("mystery", self.INSTALLED)) + + +class RepairImportsTests(unittest.TestCase): + MUSL_TAGS = {"musl": True, "pyver": "3.14", "impl": "cp", + "abi": "cp314", "arch": "x86_64"} + + def repair(self, broken, *, rc=0, scan=None): + """Run repair_imports with the subprocess layer fully mocked.""" + calls = [] + + def fake_run(argv, emit=None): + calls.append(list(argv)) + return rc + + with patch.object(envs, "_venv_tags", return_value=dict(self.MUSL_TAGS)), \ + patch.object(envs, "_site_packages", + return_value=Path("/env/site-packages")), \ + patch.object(envs, "_installed_specs", + return_value={"lxml": "lxml==6.1.2", + "ctranslate2": + "ctranslate2==4.8.1"}), \ + patch.object(envs.common, "run_console_subprocess", + side_effect=fake_run), \ + patch.object(envs, "scanned_broken_imports", + return_value=scan): + remaining = envs.repair_imports(broken, python=Path("/env/py")) + return remaining, calls + + def test_installs_via_target_with_musllinux_overrides(self): + remaining, calls = self.repair(["lxml"], rc=0, scan=[]) + self.assertEqual(remaining, []) + installs = [argv for argv in calls if "install" in argv] + self.assertEqual(len(installs), 1) + argv = installs[0] + self.assertIn("--target", argv) + self.assertIn(str(Path("/env/site-packages")), argv) + self.assertIn("--only-binary=:all:", argv) + self.assertIn("--upgrade", argv) + self.assertIn("--abi", argv) + self.assertEqual(argv[argv.index("--abi") + 1], "cp314") + platforms = [argv[i + 1] for i, part in enumerate(argv) + if part == "--platform"] + self.assertEqual(platforms, ["musllinux_1_2_x86_64", + "musllinux_1_1_x86_64"]) + self.assertEqual(argv[-1], "lxml==6.1.2") + + def test_uninstalls_before_reinstalling(self): + _, calls = self.repair(["lxml"], rc=0, scan=[]) + uninstalls = [argv for argv in calls if "uninstall" in argv] + self.assertEqual(len(uninstalls), 1) + self.assertIn("-y", uninstalls[0]) + self.assertIn("lxml", uninstalls[0]) + + def test_failed_pip_call_keeps_package_broken(self): + remaining, _ = self.repair(["lxml"], rc=1, scan=["lxml"]) + self.assertEqual(remaining, ["lxml"]) + + def test_unknown_distribution_is_not_repaired(self): + remaining, calls = self.repair(["mystery"], rc=0, scan=["mystery"]) + self.assertEqual(remaining, ["mystery"]) + self.assertEqual(calls, []) + + def test_deep_scan_vetoes_shallow_success(self): + # The reinstall exits 0 but the deep scan still flags lxml. + remaining, _ = self.repair(["lxml"], rc=0, scan=["lxml"]) + self.assertEqual(remaining, ["lxml"]) + + def test_non_musl_env_skips_repair_entirely(self): + calls = [] + with patch.object(envs, "_venv_tags", + return_value={"musl": False, "pyver": "3.14", + "impl": "cp", "abi": "cp314", + "arch": "x86_64"}), \ + patch.object(envs.common, "run_console_subprocess", + side_effect=lambda a, emit=None: + calls.append(a)): + remaining = envs.repair_imports(["lxml"], python=Path("/env/py")) + self.assertEqual(remaining, ["lxml"]) + self.assertEqual(calls, []) + + +class EnsureImportableTests(unittest.TestCase): + def test_healthy_scan_repairs_nothing(self): + with patch.object(envs, "scanned_broken_imports", return_value=[]), \ + patch.object(envs, "repair_imports") as repair: + self.assertEqual(envs.ensure_importable(), []) + repair.assert_not_called() + + def test_failed_scan_warns_without_touching_pip(self): + with patch.object(envs, "scanned_broken_imports", + return_value=None), \ + patch.object(envs, "repair_imports") as repair: + self.assertEqual(envs.ensure_importable(), []) + repair.assert_not_called() + + def test_broken_packages_are_repaired(self): + with patch.object(envs, "scanned_broken_imports", + return_value=["lxml"]), \ + patch.object(envs, "repair_imports", + return_value=[]) as repair: + self.assertEqual(envs.ensure_importable(), []) + repair.assert_called_once_with(["lxml"], emit=None) + + def test_unrepairable_packages_are_returned(self): + with patch.object(envs, "scanned_broken_imports", + return_value=["ctranslate2"]), \ + patch.object(envs, "repair_imports", + return_value=["ctranslate2"]): + self.assertEqual(envs.ensure_importable(), ["ctranslate2"]) class BootstrapTests(unittest.TestCase): diff --git a/app/tests/test_extractors.py b/app/tests/test_extractors.py index ae1794c..64666ba 100644 --- a/app/tests/test_extractors.py +++ b/app/tests/test_extractors.py @@ -7,6 +7,25 @@ from pathlib import Path from converter.extractors import extract_text +def _ebooklib_usable() -> bool: + """True when ebooklib's EPUB reader imports (it needs a working lxml).""" + try: + from ebooklib import epub # noqa: F401 + except Exception: + return False + return True + + +# The managed env can end up with compiled wheels that cannot load on this +# platform (e.g. glibc lxml under a musl interpreter) — the tool repairs or +# degrades at runtime, and these tests must degrade with it instead of +# failing. Relaunch audiobook.py once (or delete app/envs/tts) to rebuild. +requires_epub = unittest.skipUnless( + _ebooklib_usable(), + "ebooklib is unusable in this environment " + "(its compiled dependency failed to import)") + + class TxtExtractionTests(unittest.TestCase): def _extract(self, data: bytes) -> str: with tempfile.TemporaryDirectory() as tmp: @@ -67,6 +86,7 @@ class EpubExtractionTests(unittest.TestCase): except ImportError: self.skipTest("ebooklib not installed") + @requires_epub def test_ebooklib_extraction(self): # Regression test: the ebooklib path used to silently return "" due to # isinstance(item, ebooklib.ITEM_DOCUMENT) (an int, not a class). @@ -81,6 +101,7 @@ class EpubExtractionTests(unittest.TestCase): self.assertIn("First chapter text.", html) self.assertIn("Second chapter text.", html) + @requires_epub def test_epub_extraction_follows_spine_order(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "book.epub" @@ -100,6 +121,7 @@ class ExtractSectionsTests(unittest.TestCase): except ImportError: self.skipTest("ebooklib not installed") + @requires_epub def test_epub_sections_split_on_chapters(self): from converter.extractors import extract_sections @@ -126,6 +148,7 @@ class ExtractSectionsTests(unittest.TestCase): self.assertEqual(sections[0].title, "book") self.assertEqual(sections[0].text, "Hello world.") + @requires_epub def test_single_chapter_epub_keeps_chapter_title(self): from converter.extractors import extract_sections @@ -152,6 +175,7 @@ class ExtractBookTests(unittest.TestCase): self.assertEqual(book.author, "") self.assertEqual(len(book.sections), 1) + @requires_epub def test_epub_metadata_harvested(self): try: import ebooklib # noqa: F401 |