path: root/app/tests/test_backends_envs.py

"""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_env_python_posix(self):
        with patch.object(envs, "_is_windows", return_value=False):
            self.assertEqual(envs.env_python(),
                             envs.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")

    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")

    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")

    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_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()
        # 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_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()


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_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_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_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()