claude-codeandClaude Opus 5.5
f9ad6763c8
A changed file, or the old path of a renamed one, that matches [hard_orange] paths gives ORANGE with one aggregated reason; such a file is not listed under the YELLOW tiers. The section is required; the repository layer only adds paths. hsadmin-ng and werkprobe move RBAC, authentication and secrets to [hard_orange]. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
293 lines
16 KiB
Python
293 lines
16 KiB
Python
import contextlib
|
|
import io
|
|
import json
|
|
import os
|
|
import tomllib
|
|
import unittest
|
|
import unittest.mock
|
|
|
|
from tests.fakes import BASE, HEAD, ROOT, FakeGitea, FakeModel, config, deps, diff_for, gitea_file, pr_ref
|
|
from werkrevisor import __version__
|
|
from werkrevisor.cli import 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.redact import Redactor
|
|
from werkrevisor.verdict import Verdict
|
|
|
|
WERKPROBE = os.path.join(ROOT, "config", "werkprobe.toml")
|
|
|
|
|
|
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_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_hard_yellow_paths_are_added(self):
|
|
cfg = build_config(host_data(), repo('[hard_yellow]\npaths = ["werkrevisor/config.py", ".gitea/**"]\n'))
|
|
self.assertIn("**/db/**", cfg.hard_yellow.paths)
|
|
self.assertEqual(cfg.hard_yellow.paths[-1], "werkrevisor/config.py")
|
|
self.assertEqual(cfg.hard_yellow.paths.count(".gitea/**"), 1)
|
|
|
|
def test_hard_orange_paths_are_added(self):
|
|
host = host_data()
|
|
host["hard_orange"]["paths"] = ["**/auth/**"]
|
|
cfg = build_config(host, repo('[hard_orange]\npaths = ["werkrevisor/gitea.py", "**/auth/**"]\n'))
|
|
self.assertEqual(cfg.hard_orange.paths, ("**/auth/**", "werkrevisor/gitea.py"))
|
|
|
|
def test_repository_cannot_remove_hard_orange_paths(self):
|
|
host = host_data()
|
|
host["hard_orange"]["paths"] = ["**/auth/**"]
|
|
self.assertEqual(build_config(host, repo("[hard_orange]\npaths = []\n")).hard_orange.paths, ("**/auth/**",))
|
|
|
|
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, 80000),
|
|
("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_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 base bbbbbbb must not set [gate] yellow_gate",
|
|
err.getvalue())
|
|
self.assertEqual(gitea.raw_requests, [(".werkrevisor.toml", BASE)])
|
|
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()
|