diff options
Diffstat (limited to 'tests/test_backends_envs.py')
| -rw-r--r-- | tests/test_backends_envs.py | 204 |
1 files changed, 204 insertions, 0 deletions
diff --git a/tests/test_backends_envs.py b/tests/test_backends_envs.py new file mode 100644 index 0000000..cf4ecc6 --- /dev/null +++ b/tests/test_backends_envs.py @@ -0,0 +1,204 @@ +"""Tests for the managed Python environment (backends/envs.py).""" + +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): + 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 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 EnsureAppEnvTests(unittest.TestCase): + def test_creates_env_then_installs_when_marker_invalid(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, "_write_marker") as mk: + envs.ensure_app_env() + 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_raises_when_install_fails(self): + with patch.object(envs, "env_exists", return_value=True), \ + patch.object(envs, "_marker_valid", return_value=False), \ + patch.object(envs, "install_requirements", return_value=1): + with self.assertRaises(RuntimeError): + envs.ensure_app_env() + + def test_skips_install_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: + envs.ensure_app_env() + mk.assert_not_called() + + +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() |