323 lines
18 KiB
Python
323 lines
18 KiB
Python
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()
|