"""Tests for the managed Python environment (backends/envs.py).""" import json import sys import unittest from pathlib import Path from unittest.mock import patch from backends import envs class EnvPathTests(unittest.TestCase): """Platform-aware path helpers (no venv actually created).""" def test_env_dir_under_envs_tts(self): self.assertEqual(envs.ENV_DIR.name, "tts") self.assertEqual(envs.ENV_DIR.parent.name, "envs") def test_backend_env_dirs_are_separate_from_the_app_env(self): # Each pip-installed backend gets its own venv next to the app's: # qwen-tts and faster-qwen3-tts both ship a qwen_tts module with # conflicting transformers pins, so they must never share one. self.assertEqual(envs.QWEN_ENV_DIR.name, "qwen") self.assertEqual(envs.FASTER_ENV_DIR.name, "faster") self.assertEqual(envs.QWEN_ENV_DIR.parent, envs.ENV_DIR.parent) distinct = {envs.ENV_DIR, envs.QWEN_ENV_DIR, envs.FASTER_ENV_DIR} self.assertEqual(len(distinct), 3) def test_env_python_posix(self): with patch.object(envs, "_is_windows", return_value=False): self.assertEqual(envs.env_python(), envs.ENV_DIR / "bin" / "python") self.assertEqual( envs.env_python(envs.QWEN_ENV_DIR), envs.QWEN_ENV_DIR / "bin" / "python") def test_env_python_windows(self): with patch.object(envs, "_is_windows", return_value=True): self.assertEqual(envs.env_python(), envs.ENV_DIR / "Scripts" / "python.exe") self.assertEqual( envs.env_python(envs.FASTER_ENV_DIR), envs.FASTER_ENV_DIR / "Scripts" / "python.exe") def test_env_script_posix(self): with patch.object(envs, "_is_windows", return_value=False): self.assertEqual(envs.env_script("qwen-tts-demo"), envs.ENV_DIR / "bin" / "qwen-tts-demo") self.assertEqual( envs.env_script("qwen-tts-demo", envs.QWEN_ENV_DIR), envs.QWEN_ENV_DIR / "bin" / "qwen-tts-demo") def test_env_script_windows(self): with patch.object(envs, "_is_windows", return_value=True): self.assertEqual(envs.env_script("qwen-tts-demo"), envs.ENV_DIR / "Scripts" / "qwen-tts-demo.exe") self.assertEqual( envs.env_script("qwen-tts-demo", envs.QWEN_ENV_DIR), envs.QWEN_ENV_DIR / "Scripts" / "qwen-tts-demo.exe") def test_env_exists_false_when_python_missing(self): with patch.object(envs, "env_python", return_value=Path("/no/such/path/python")): self.assertFalse(envs.env_exists()) def test_is_managed_env_compares_resolved_executable(self): fake_env_python = Path("/tmp/opencode/managed-env/bin/python") with patch.object(envs, "env_python", return_value=fake_env_python), \ patch.object(sys, "executable", str(fake_env_python)): self.assertTrue(envs.is_managed_env()) with patch.object(envs, "env_python", return_value=fake_env_python), \ patch.object(sys, "executable", "/usr/bin/python3"): self.assertFalse(envs.is_managed_env()) class CreateEnvTests(unittest.TestCase): def test_create_env_invokes_venv_module(self): with patch.object(envs.common, "run_console_subprocess", return_value=0) as run: rc = envs.create_env() self.assertEqual(rc, 0) argv = run.call_args[0][0] self.assertEqual(argv[0], sys.executable) self.assertEqual(argv[1], "-m") self.assertEqual(argv[2], "venv") self.assertEqual(argv[3], str(envs.ENV_DIR)) def test_create_env_targets_the_requested_env_dir(self): with patch.object(envs.common, "run_console_subprocess", return_value=0) as run: envs.create_env(envs.FASTER_ENV_DIR) argv = run.call_args[0][0] self.assertEqual(argv[3], str(envs.FASTER_ENV_DIR)) def test_create_env_reports_remediation_on_failure(self): with patch.object(envs.common, "run_console_subprocess", return_value=1): rc = envs.create_env() self.assertEqual(rc, 1) class PipInstallTests(unittest.TestCase): def test_creates_env_first_when_missing(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs, "create_env", return_value=0) as mk, \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): rc = envs.pip_install(["qwen-tts"]) self.assertEqual(rc, 0) mk.assert_called_once_with(None) # The actual pip call targets the venv's python. self.assertEqual(calls[0][0], str(envs.env_python())) self.assertIn("pip", calls[0]) self.assertIn("qwen-tts", calls[0]) def test_backend_env_targets_the_backend_python_and_env(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs, "create_env", return_value=0) as mk, \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): rc = envs.pip_install(["qwen-tts"], env_dir=envs.QWEN_ENV_DIR) self.assertEqual(rc, 0) # Both create-if-missing and pip itself are scoped to the qwen env; # the app env is never touched. mk.assert_called_once_with(envs.QWEN_ENV_DIR) self.assertEqual(calls[0][0], str(envs.env_python(envs.QWEN_ENV_DIR))) def test_skips_create_when_env_exists(self): with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs, "create_env") as mk, \ patch.object(envs.common, "run_console_subprocess", return_value=0): envs.pip_install(["qwen-tts"]) mk.assert_not_called() def test_returns_nonzero_when_create_fails(self): with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs, "create_env", return_value=1), \ patch.object(envs.common, "run_console_subprocess") as run: rc = envs.pip_install(["qwen-tts"]) self.assertEqual(rc, 1) run.assert_not_called() def test_upgrade_adds_the_u_flag(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): self.assertEqual(envs.pip_install(["qwen-tts"], upgrade=True), 0) self.assertIn("-U", calls[0]) # -U sits before the packages; nothing else about the argv changes. self.assertLess(calls[0].index("-U"), calls[0].index("qwen-tts")) def test_no_upgrade_flag_by_default(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): self.assertEqual(envs.pip_install(["qwen-tts"]), 0) self.assertNotIn("-U", calls[0]) class PipUninstallTests(unittest.TestCase): def test_missing_env_is_success_without_running_pip(self): with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs.common, "run_console_subprocess") as run: rc = envs.pip_uninstall(["qwen-tts"]) self.assertEqual(rc, 0) run.assert_not_called() def test_missing_backend_env_is_also_a_noop(self): with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs.common, "run_console_subprocess") as run: rc = envs.pip_uninstall(["qwen-tts"], env_dir=envs.QWEN_ENV_DIR) self.assertEqual(rc, 0) run.assert_not_called() def test_runs_pip_uninstall_against_the_venv_python(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): rc = envs.pip_uninstall(["qwen-tts"]) self.assertEqual(rc, 0) # The uninstall targets the venv's python. self.assertEqual(calls[0][0], str(envs.env_python())) self.assertIn("uninstall", calls[0]) self.assertIn("-y", calls[0]) self.assertIn("qwen-tts", calls[0]) def test_targets_the_requested_env_python(self): calls = [] def fake_run(argv, **kwargs): calls.append(list(argv)) return 0 with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", side_effect=fake_run): envs.pip_uninstall(["faster-qwen3-tts"], env_dir=envs.FASTER_ENV_DIR) self.assertEqual(calls[0][0], str(envs.env_python(envs.FASTER_ENV_DIR))) def test_streams_to_emit_when_given(self): with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", return_value=0) as run: rc = envs.pip_uninstall(["qwen-tts"], emit="EMIT") self.assertEqual(rc, 0) # The task view's emit is forwarded so pip never touches the # terminal behind curses. self.assertEqual(run.call_args.kwargs.get("emit"), "EMIT") def test_console_path_passes_no_emit(self): with patch.object(envs, "env_exists", return_value=True), \ patch.object(envs.common, "run_console_subprocess", return_value=0) as run: rc = envs.pip_uninstall(["qwen-tts"]) self.assertEqual(rc, 0) self.assertIsNone(run.call_args.kwargs.get("emit")) class ModuleAvailableTests(unittest.TestCase): def test_false_when_env_missing(self): with patch.object(envs, "env_exists", return_value=False): self.assertFalse(envs.module_available("qwen_tts")) def test_true_when_subprocess_exits_zero(self): import subprocess fake = subprocess.CompletedProcess(args=["x"], returncode=0) with patch.object(envs, "env_exists", return_value=True), \ patch("subprocess.run", return_value=fake) as run: self.assertTrue(envs.module_available("qwen_tts")) argv = run.call_args[0][0] self.assertEqual(argv[0], str(envs.env_python())) self.assertIn("import qwen_tts", argv[2]) def test_probes_the_requested_env_interpreter(self): import subprocess fake = subprocess.CompletedProcess(args=["x"], returncode=0) with patch.object(envs, "env_exists", return_value=True), \ patch("subprocess.run", return_value=fake) as run: self.assertTrue( envs.module_available("qwen_tts", envs.QWEN_ENV_DIR)) argv = run.call_args[0][0] self.assertEqual(argv[0], str(envs.env_python(envs.QWEN_ENV_DIR))) self.assertIn("import qwen_tts", argv[2]) def test_false_when_subprocess_exits_nonzero(self): import subprocess fake = subprocess.CompletedProcess(args=["x"], returncode=1) with patch.object(envs, "env_exists", return_value=True), \ patch("subprocess.run", return_value=fake): self.assertFalse(envs.module_available("qwen_tts")) def test_false_on_timeout(self): import subprocess with patch.object(envs, "env_exists", return_value=True), \ patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="x", timeout=1)): 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): """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) 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): with patch.object(envs, "env_exists", return_value=False), \ patch.object(envs, "create_env", return_value=1): with self.assertRaises(RuntimeError): envs.ensure_app_env() 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_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 install, \ patch.object(envs, "ensure_importable") as verify: envs.ensure_app_env() 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): def test_noop_when_already_managed(self): with patch.object(envs, "is_managed_env", return_value=True), \ patch.object(envs, "ensure_app_env") as mk, \ patch("os.execv") as ex: envs.bootstrap("/path/to/audiobook.py") mk.assert_not_called() ex.assert_not_called() def test_ensures_env_then_execvs(self): with patch.object(envs, "is_managed_env", return_value=False), \ patch.object(envs, "ensure_app_env") as mk_env, \ patch("os.execv") as ex, \ patch.object(sys, "argv", ["audiobook.py", "--backend", "qwen"]): envs.bootstrap("/path/to/audiobook.py") mk_env.assert_called_once_with() py = str(envs.env_python()) args = ex.call_args[0] self.assertEqual(args[0], py) self.assertEqual(args[1][0], py) self.assertTrue(args[1][1].endswith("audiobook.py")) self.assertEqual(args[1][2:], ["--backend", "qwen"]) def test_exits_when_ensure_raises(self): with patch.object(envs, "is_managed_env", return_value=False), \ patch.object(envs, "ensure_app_env", side_effect=RuntimeError("boom")), \ patch("os.execv") as ex, \ self.assertRaises(SystemExit): envs.bootstrap("/path/to/audiobook.py") ex.assert_not_called() if __name__ == "__main__": unittest.main()