import contextlib import io import json import os import tomllib import unittest import unittest.mock from tests.fakes import (BASE, HEAD, MAIN, ROOT, TEST_CONFIG, FakeGitea, FakeModel, config, deps, diff_for, gitea_file, pr_ref) from werkrevisor import __version__ from werkrevisor.cli import default_branch_head, execute, main, resolve_config from werkrevisor.config import ( ConfigError, HostConfig, build_config, check_repository_config, load_host_config, merge_repository_layer, parse_repository_config, ) from werkrevisor.gitea import GiteaError from werkrevisor.rules import ChangedFile, evaluate_rules from werkrevisor.redact import Redactor from werkrevisor.verdict import Verdict WERKPROBE = TEST_CONFIG def host_data(): with open(WERKPROBE, "rb") as f: return tomllib.load(f) def repo(text): return parse_repository_config(text.encode()) class RawFiles: """Gitea raw-file API with different content per commit; a missing file is HTTP 404.""" def __init__(self, files): self.files = files self.requests = [] def __call__(self, path, ref): self.requests.append((path, ref)) if (path, ref) not in self.files: raise GiteaError(f"GET /raw/{path} failed with HTTP 404", 404) return self.files[(path, ref)] class LayeringTest(unittest.TestCase): def test_repository_wins_over_host_and_host_over_defaults(self): host = host_data() del host["review"]["max_findings"] self.assertEqual(build_config(host).review.max_findings, 3) host["review"]["max_findings"] = 5 self.assertEqual(build_config(host).review.max_findings, 5) cfg = build_config(host, repo("[review]\nmax_findings = 7\n[size]\nmax_files = 12\n")) self.assertEqual((cfg.review.max_findings, cfg.size.max_files, cfg.size.max_changed_lines), (7, 12, 800)) def test_repository_sections_complete_a_host_config(self): with open(os.path.join(ROOT, "config", "werkrevisor.toml"), "rb") as f: host = tomllib.load(f) with self.assertRaisesRegex(ConfigError, "sections missing: .*neither the host config nor .werkrevisor.toml"): build_config(host) with open(os.path.join(ROOT, ".werkrevisor.toml"), "rb") as f: cfg = build_config(host, tomllib.load(f)) self.assertEqual((cfg.pr_doc.globs, cfg.gate.yellow_gate), (("docs/prs/*.md",), "approval")) def test_the_repository_chooses_the_codex_mode(self): self.assertEqual(build_config(host_data()).codex.mode, "off") self.assertEqual(build_config(host_data(), repo('[codex]\nmode = "gate"\n')).codex.mode, "gate") with self.assertRaisesRegex(ConfigError, r"\[codex\] mode must be one of off, advise, gate"): build_config(host_data(), repo('[codex]\nmode = "on"\n')) def test_repository_sets_the_writing_rules(self): host = host_data() self.assertEqual(build_config(host).writing.max_line_length, 128) cfg = build_config(host, repo('[writing]\nmax_line_length = 100\nglobs = ["docs/**"]\nenabled = false\n')) self.assertEqual((cfg.writing.max_line_length, cfg.writing.globs, cfg.writing.enabled), (100, ("docs/**",), False)) del host["writing"] cfg = build_config(host, repo("[writing]\nmax_sentences_per_paragraph = 4\n")) self.assertEqual((cfg.writing.max_sentences_per_paragraph, cfg.writing.max_line_length), (4, 128)) with self.assertRaisesRegex(ConfigError, r"sections missing: \[writing\]"): build_config(host) def test_the_repository_replaces_the_hard_and_judged_paths(self): host = host_data() host["hard_yellow"]["paths"] = ["tools/**"] cfg = build_config(host, repo('[hard_yellow]\npaths = ["**/*.pem"]\njudged_paths = ["Dockerfile*"]\n')) self.assertEqual((cfg.hard_yellow.paths, cfg.hard_yellow.judged_paths), (("**/*.pem",), ("Dockerfile*",))) # Its own config stays hard whatever the repository lists, since it may loosen the rules of later PRs. files = [ChangedFile(".werkrevisor.toml", "modified", 1), ChangedFile("tools/deploy.sh", "modified", 1)] rules = evaluate_rules(files, {}, cfg) risky = next(r.text for r in rules.reasons if r.code == "risky_paths") self.assertEqual(risky, "risky paths changed: .werkrevisor.toml (built in)") def test_extra_instructions_are_appended(self): host = host_data() host["review"]["extra_instructions"] = "Host rule." cfg = build_config(host, repo('[review]\nextra_instructions = "Repository rule."\n')) self.assertEqual(cfg.review.extra_instructions, "Host rule.\n\nRepository rule.") def test_host_only_keys_are_refused(self): for section, line in (("model", 'id = "openrouter/x/y"'), ("model", "input_price_usd_per_mtok = 0.1"), ("model", 'reasoning_effort = "high"'), ("budget", "min_key_remaining_usd = 0.0"), ("gate", 'yellow_gate = "block"'), ("gate", 'human_reviewers = ["dev"]'), ("gate", 'assessment_context = "x"'), ("engine", 'mode = "classifier"'), ("classifier", 'endpoint = "https://evil.example.org"'), ("classifier", "input_price_usd_per_mtok = 0.0")): with self.subTest(key=f"[{section}] {line}"): key = line.split(" ")[0] with self.assertRaisesRegex(ConfigError, rf"must not set \[{section}\] {key}: only the host config"): build_config(host_data(), repo(f"[{section}]\n{line}\n")) def test_unknown_keys_and_sections_are_refused(self): for text, message in (("[review]\nmodle = 1\n", "unknown keys: modle"), ("[secrets]\ntoken = 1\n", r"unknown section \[secrets\]")): with self.subTest(text=text): with self.assertRaisesRegex(ConfigError, message): build_config(host_data(), repo(text)) def test_limits_may_only_be_lowered(self): for section, key, lower, higher in (("model", "max_context_tokens", 100000, 200000), ("model", "max_output_tokens", 3000, 8000), ("budget", "max_usd_per_review", 0.3, 2.0), ("review", "max_full_file_chars", 20000, 90000), ("review", "max_full_files_chars", 60000, 500000)): with self.subTest(key=key): cfg = build_config(host_data(), repo(f"[{section}]\n{key} = {lower}\n")) self.assertEqual(getattr(getattr(cfg, section), key), lower) with self.assertRaisesRegex(ConfigError, rf"\[{section}\] {key} = {higher} exceeds the host value"): build_config(host_data(), repo(f"[{section}]\n{key} = {higher}\n")) def test_excerpt_context_is_host_only_and_validated(self): with self.assertRaisesRegex(ConfigError, r"must not set \[review\] excerpt_context_lines: only the host config"): build_config(host_data(), repo("[review]\nexcerpt_context_lines = 400\n")) host = host_data() for value in (-1, 501, "40", True): with self.subTest(value=value): host["review"]["excerpt_context_lines"] = value with self.assertRaisesRegex(ConfigError, r"excerpt_context_lines must be an integer between 0 and 500"): build_config(host) def test_secret_scan_cannot_be_switched_off(self): with self.assertRaisesRegex(ConfigError, r"must not switch off \[hard_yellow\] secret_scan"): build_config(host_data(), repo("[hard_yellow]\nsecret_scan = false\n")) def test_classifier_questions_need_a_host_classifier(self): host = host_data() questions = host["classifier"]["section_questions"][:1] cfg = build_config(host, {"classifier": {"section_questions": questions, "min_confidence": 0.5}}) self.assertEqual(([q.name for q in cfg.classifier.section_questions], cfg.classifier.min_confidence), (["hard_to_understand"], 0.5)) del host["classifier"] with self.assertRaisesRegex(ConfigError, r"host config has no \[classifier\]"): merge_repository_layer(host, {"classifier": {"min_confidence": 0.5}}) def test_host_config_rejects_typos_before_merging(self): with self.assertRaisesRegex(ConfigError, "host config: .*unknown keys: modle"): load_host_config({"WERKREVISOR_CONFIG_TOML": "[review]\nmodle = 1\n"}) class RepositoryReadTest(unittest.TestCase): HOST = HostConfig(host_data(), "env") def test_reads_the_base_not_the_head(self): raw = RawFiles({(".werkrevisor.toml", BASE): b"[size]\nmax_files = 11\n", (".werkrevisor.toml", HEAD): b"[size]\nmax_files = 99\n"}) cfg, sources = resolve_config(self.HOST, raw, BASE) self.assertEqual(cfg.size.max_files, 11) self.assertEqual(raw.requests, [(".werkrevisor.toml", BASE)]) self.assertEqual(sources, f"defaults + host + repository (.werkrevisor.toml at base {BASE[:7]})") def test_missing_file_means_no_repository_layer(self): cfg, sources = resolve_config(self.HOST, RawFiles({}), BASE) self.assertEqual((cfg.size.max_files, sources), (30, "defaults + host")) def test_other_read_errors_and_an_unknown_base_are_errors(self): def failing(path, ref): raise GiteaError("GET /raw failed with HTTP 500", 500) with self.assertRaisesRegex(ConfigError, "cannot read .werkrevisor.toml at base"): resolve_config(self.HOST, failing, BASE) with self.assertRaisesRegex(ConfigError, "base commit is unknown"): resolve_config(self.HOST, RawFiles({}), "") def test_invalid_toml_names_the_position_only(self): raw = RawFiles({(".werkrevisor.toml", BASE): b'[review]\nextra_instructions = "sk-or-v1-planted\n'}) with self.assertRaises(ConfigError) as caught: resolve_config(self.HOST, raw, BASE) self.assertIn(f"invalid TOML in .werkrevisor.toml at base {BASE[:7]} at line 2", str(caught.exception)) self.assertNotIn("planted", str(caught.exception)) class MainRepositoryConfigTest(unittest.TestCase): def test_forbidden_repository_key_is_error_exit_30(self): gitea = FakeGitea() gitea.raw_files[".werkrevisor.toml"] = b'[gate]\nyellow_gate = "pass_with_review_request"\n' environ = {"WERKREVISOR_CONFIG": WERKPROBE, "GITEA_BOT_TOKEN": "gitea-token-123", "OPENROUTER_API_KEY": "sk-or-test", "TYPESAFE_API_KEY": "typesafe-test", "WERKATOR_GITEA_BASE_URL": "https://git.example.org", "WERKATOR_GITEA_OWNER": "team", "WERKATOR_GITEA_REPO": "app", "WERKATOR_PR_NUMBER": "7", "WERKATOR_PR_HEAD_SHA": HEAD, "WERKATOR_PR_BASE_SHA": BASE} err = io.StringIO() with unittest.mock.patch("werkrevisor.cli.GiteaClient", return_value=gitea), contextlib.redirect_stderr(err): self.assertEqual(main([], environ), 30) self.assertIn("configuration error: .werkrevisor.toml at main ddddddd must not set [gate] yellow_gate", err.getvalue()) self.assertEqual(gitea.raw_requests, [(".werkrevisor.toml", MAIN)]) def test_the_repository_config_comes_from_the_default_branch(self): gitea = FakeGitea() gitea.raw_files_at[(".werkrevisor.toml", MAIN)] = b"[size]\nmax_files = 11\n" gitea.raw_files_at[(".werkrevisor.toml", BASE)] = b"[size]\nmax_files = 99\n" branch, sha = default_branch_head(gitea) cfg, sources = resolve_config(HostConfig(host_data(), "env"), gitea.get_raw_file, sha, branch) self.assertEqual((branch, sha, cfg.size.max_files), ("main", MAIN, 11)) self.assertEqual(sources, f"defaults + host + repository (.werkrevisor.toml at main {MAIN[:7]})") gitea.branch = {"name": "main"} with self.assertRaisesRegex(ConfigError, "the default branch is unknown, .* Gitea names no commit of main"): default_branch_head(gitea) gitea.get_repo = lambda: {} with self.assertRaisesRegex(ConfigError, "the default branch is unknown, .* Gitea names no default branch"): default_branch_head(gitea) gitea.get_repo = lambda: (_ for _ in ()).throw(OSError("down")) with self.assertRaisesRegex(ConfigError, "the default branch is unknown, .*: down"): default_branch_head(gitea) self.assertEqual(gitea.writes, []) class ChangedRepositoryConfigTest(unittest.TestCase): def run_with(self, content): gitea = FakeGitea(files=[gitea_file(".werkrevisor.toml")], diff=diff_for(".werkrevisor.toml", added=["x"])) gitea.raw_files[".werkrevisor.toml"] = content model = FakeModel(prompt_files=[".werkrevisor.toml"]) d = deps(gitea, model) d.check_repository_config = lambda raw: check_repository_config(host_data(), raw) outcome = execute(config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor(), config_source="defaults + host") return outcome.result, model, gitea def test_invalid_change_is_red_without_model_call(self): result, model, _ = self.run_with(b"[model]\nid = \"openrouter/cheap/model\"\n") self.assertIs(result.verdict, Verdict.RED) self.assertIn("the changed .werkrevisor.toml is invalid: .werkrevisor.toml must not set [model] id", " ".join(r.text for r in result.reasons)) self.assertEqual(model.calls, []) def test_wrong_value_type_is_red(self): result, _, _ = self.run_with(b'[review]\nmax_findings = "many"\n') self.assertIs(result.verdict, Verdict.RED) self.assertIn("repository_config_invalid", [r.code for r in result.reasons]) def test_setting_unknown_to_the_installed_version_is_yellow(self): for content, shown in ((b"[future]\nflag = true\n", "[future]"), (b"[size]\nmax_files = 10\nmax_hunks = 3\n", "[size] max_hunks")): with self.subTest(shown=shown): result, model, _ = self.run_with(content) self.assertIs(result.verdict, Verdict.YELLOW) reason = next(r for r in result.reasons if r.code == "repository_config_unknown") self.assertIs(reason.verdict, Verdict.YELLOW) self.assertEqual(reason.text, "the changed .werkrevisor.toml uses settings unknown to the installed " f"werkrevisor {__version__}: {shown}; deploy a version that knows them " "right after the merge, or every review will end with ERROR") self.assertEqual(len(model.calls), 1) def test_old_red_min_findings_is_yellow(self): result, model, _ = self.run_with(b"[review]\nmax_findings = 3\nred_min_findings = 2\n") self.assertIs(result.verdict, Verdict.YELLOW) self.assertNotIn("repository_config_invalid", [r.code for r in result.reasons]) reason = next(r for r in result.reasons if r.code == "repository_config_unknown") self.assertIn(f"werkrevisor {__version__}: [review] red_min_findings;", reason.text) self.assertEqual(len(model.calls), 1) def test_change_kind_unknown_to_the_installed_version_is_yellow(self): result, model, _ = self.run_with(b'[categories]\nauto_approvable = ["docs_only", "future_kind"]\n') self.assertIs(result.verdict, Verdict.YELLOW) self.assertNotIn("repository_config_invalid", [r.code for r in result.reasons]) reason = next(r for r in result.reasons if r.code == "repository_config_unknown") self.assertEqual(reason.text, "the changed .werkrevisor.toml uses settings unknown to the installed " f"werkrevisor {__version__}: [categories] auto_approvable 'future_kind'; " "deploy a version that knows them right after the merge, " "or every review will end with ERROR") self.assertEqual(len(model.calls), 1) def test_change_kind_of_a_wrong_type_stays_red(self): result, _, _ = self.run_with(b'[categories]\nauto_approvable = ["docs_only", 7]\n') self.assertIs(result.verdict, Verdict.RED) self.assertIn("repository_config_invalid", [r.code for r in result.reasons]) def test_syntax_error_and_unknown_with_invalid_settings_stay_red(self): for content in (b"[size\nmax_files = 10\n", b'[future]\nflag = true\n[review]\nmax_findings = "many"\n', b'[future]\nflag = true\n[gate]\nyellow_gate = "block"\n'): with self.subTest(content=content): result, model, _ = self.run_with(content) self.assertIs(result.verdict, Verdict.RED) self.assertIn("repository_config_invalid", [r.code for r in result.reasons]) self.assertNotIn("repository_config_unknown", [r.code for r in result.reasons]) self.assertEqual(model.calls, []) def test_valid_change_needs_a_human(self): result, model, _ = self.run_with(b"[size]\nmax_files = 10\n") self.assertIs(result.verdict, Verdict.YELLOW) self.assertIn("risky paths changed: .werkrevisor.toml (built in)", [r.text for r in result.reasons]) self.assertEqual(len(model.calls), 1) def test_comment_names_the_config_sources(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"])) sources = f"defaults + host + repository (.werkrevisor.toml at base {BASE[:7]})" execute(config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor(), config_source=sources, host_config="env") self.assertIn(f"config: {sources}", gitea.created_comments[0]) record = json.loads(d.out.getvalue()) self.assertEqual((record["config_source"], record["host_config"]), (sources, "env")) if __name__ == "__main__": unittest.main()