153 lines
7.0 KiB
Python
153 lines
7.0 KiB
Python
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
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))
|