import contextlib import io import json import os import subprocess import sys import unittest from tests.fakes import ROOT, FakeGitea, FakeModel, config, deps, diff_for, gitea_file, pr_ref from werkrevisor import isolation from werkrevisor.cli import execute, main from werkrevisor.config import ConfigError, load_operator_config from werkrevisor.redact import Redactor WERKPROBE = os.path.join(ROOT, "config", "werkprobe.toml") SECRET_IN_TOML = "sk-or-v1-planted-secret-0123456789" def inline(model_id="openrouter/test/inline-model"): with open(WERKPROBE, encoding="utf-8") as f: text = f.read() return text.replace('id = "openrouter/anthropic/claude-sonnet-5.5"', f'id = "{model_id}"') class ConfigSourceTest(unittest.TestCase): def test_env_wins_over_file(self): cfg, source = load_operator_config({"WERKREVISOR_CONFIG_TOML": inline(), "WERKREVISOR_CONFIG": WERKPROBE}, WERKPROBE) self.assertEqual((cfg.model.id, source), ("openrouter/test/inline-model", "env")) def test_file_fallback(self): for environ, argument in (({"WERKREVISOR_CONFIG": WERKPROBE}, None), ({"WERKREVISOR_CONFIG_TOML": " \n"}, WERKPROBE)): with self.subTest(environ=environ): cfg, source = load_operator_config(environ, argument) self.assertEqual((cfg.model.id, source), ("openrouter/anthropic/claude-sonnet-5.5", WERKPROBE)) def test_no_source(self): with self.assertRaisesRegex(ConfigError, "WERKREVISOR_CONFIG_TOML or WERKREVISOR_CONFIG"): load_operator_config({}) def test_invalid_env_toml_names_position_only(self): with self.assertRaises(ConfigError) as caught: load_operator_config({"WERKREVISOR_CONFIG_TOML": f'[model]\nid = "{SECRET_IN_TOML}\n'}) self.assertIn("WERKREVISOR_CONFIG_TOML at line", str(caught.exception)) self.assertNotIn("planted", str(caught.exception)) def test_env_toml_is_validated_like_a_file(self): with self.assertRaisesRegex(ConfigError, "unknown keys: modle"): load_operator_config({"WERKREVISOR_CONFIG_TOML": inline().replace("[review]", "[review]\nmodle = 1")}) class MainConfigTest(unittest.TestCase): def run_main(self, environ): err = io.StringIO() with contextlib.redirect_stderr(err): code = main([], environ) return code, err.getvalue() def test_invalid_env_toml_is_error_without_content(self): code, err = self.run_main({"WERKREVISOR_CONFIG_TOML": f'x = "{SECRET_IN_TOML}'}) self.assertEqual(code, 30) self.assertIn("configuration error", err) self.assertNotIn("planted", err) def test_logs_the_source_when_both_are_set(self): code, err = self.run_main({"WERKREVISOR_CONFIG_TOML": inline(), "WERKREVISOR_CONFIG": WERKPROBE}) self.assertEqual(code, 30) # stops later: no WERKATOR_* environment self.assertIn("operator config from WERKREVISOR_CONFIG_TOML (the config file is ignored)", err) self.assertNotIn("inline-model", err) def test_usage_record_has_config_source(self): gitea = FakeGitea(files=[gitea_file("doc/a.md")], diff=diff_for("doc/a.md", added=["x"])) d = deps(gitea, FakeModel(prompt_files=["doc/a.md"])) execute(config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor(), config_source="env") self.assertEqual(json.loads(d.out.getvalue())["config_source"], "env") class IsolationTest(unittest.TestCase): def test_user_site_is_refused(self): self.assertIn("python -s", isolation.interpreter_problem({"HOME": "/home/x"}, ["/app"], True)) def test_sys_path_under_home_is_refused(self): problem = isolation.interpreter_problem({"HOME": "/home/x"}, ["/app", "/home/x/.local/lib"], False) self.assertIn("/home/x/.local/lib", problem) def test_clean_interpreter_passes(self): self.assertIsNone(isolation.interpreter_problem({"HOME": "/home/x"}, ["/app", "/home/xy"], None)) def test_working_directory_is_dropped(self): cwd = os.getcwd() path = ["", cwd, "/app"] isolation.drop_working_directory(path, "/werkrevisor-elsewhere") self.assertEqual(path, ["/app"]) def test_isolate_home(self): environ = {"HOME": "/root"} home = isolation.isolate_home(environ) self.assertNotEqual(home, "/root") self.assertTrue(environ["XDG_CACHE_HOME"].startswith(home)) self.assertEqual((environ["LITELLM_MODE"], environ["LITELLM_LOCAL_MODEL_COST_MAP"]), ("PRODUCTION", "True")) class EntrypointTest(unittest.TestCase): def env(self, home, extra_env): env = {k: v for k, v in os.environ.items() if k != "PYTHONNOUSERSITE"} env.update(HOME=home, PYTHONPATH=os.pathsep.join([ROOT] + [p for p in sys.path if p.startswith("/app")])) env.update(extra_env) return env def start(self, *python_args, home="/nonexistent-werkrevisor-home", executable=sys.executable, **extra_env): return subprocess.run([executable, *python_args, "-m", "werkrevisor", "--version"], env=self.env(home, extra_env), capture_output=True, text=True, cwd=ROOT, timeout=60) def interpreter_with_user_site(self): # A venv interpreter disables the user site by itself; the base interpreter shows the check. for candidate in (sys.executable, "/usr/local/bin/python3", "/usr/bin/python3"): if not os.path.exists(candidate): continue probe = subprocess.run([candidate, "-c", "import site; print(site.ENABLE_USER_SITE)"], env=self.env("/nonexistent-werkrevisor-home", {}), capture_output=True, text=True) if probe.stdout.strip() == "True": return candidate self.skipTest("no interpreter with user site-packages enabled") def test_refuses_with_user_site(self): result = self.start(executable=self.interpreter_with_user_site()) self.assertEqual(result.returncode, 30) self.assertIn("refusing to run: user site-packages are enabled", result.stderr) def test_runs_with_python_s(self): result = self.start("-s", PYTHONNOUSERSITE="1") self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("werkrevisor", result.stdout) def test_refuses_code_under_home(self): result = self.start("-s", home=ROOT, PYTHONNOUSERSITE="1") self.assertEqual(result.returncode, 30) self.assertIn("lies under HOME", result.stderr) class MissingSectionTest(unittest.TestCase): def test_a_config_without_every_section_is_refused(self): from werkrevisor.config import ConfigError, load_operator_config partial = '[model]\nid = "openrouter/x/y"\n[budget]\nmax_usd_per_review = 0.1\n' with self.assertRaises(ConfigError) as caught: load_operator_config({"WERKREVISOR_CONFIG_TOML": partial}) self.assertIn("[hard_yellow]", str(caught.exception)) def test_a_config_without_hard_orange_is_refused(self): import tomllib from werkrevisor.config import ConfigError, build_config with open(os.path.join(ROOT, "config", "werkprobe.toml"), "rb") as f: host = tomllib.load(f) del host["hard_orange"] with self.assertRaisesRegex(ConfigError, r"sections missing: \[hard_orange\]"): build_config(host, {}) if __name__ == "__main__": unittest.main()