werkrevisor/tests/test_config_source_and_isolation.py

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