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>
309 lines
16 KiB
Python
309 lines
16 KiB
Python
import os
|
|
import tomllib
|
|
import unittest
|
|
|
|
from tests.fakes import ROOT, FakeGitea, FakeModel, config, deps, diff_for, example_config, gitea_file, pr_ref
|
|
from werkrevisor.cli import execute
|
|
from werkrevisor.config import ConfigError, build_config, merge_repository_layer
|
|
from werkrevisor.diff import parse_unified_diff
|
|
from werkrevisor.mechanical import mechanical_files
|
|
from werkrevisor.redact import Redactor
|
|
from werkrevisor.rules import evaluate_rules, files_from_gitea
|
|
from werkrevisor.verdict import Verdict
|
|
|
|
MAIN = "src/main/java/net/example/A.java"
|
|
IMPORTS_ONLY = "src/main/java/net/example/B.java"
|
|
TEST = "src/test/java/net/example/ATest.java"
|
|
|
|
|
|
def hunk_diff(path, lines, status="modified", previous=""):
|
|
"""A diff with one hunk; each line starts with ' ', '+' or '-' like in a unified diff."""
|
|
old = previous or path
|
|
removed = sum(1 for line in lines if line.startswith("-"))
|
|
added = sum(1 for line in lines if line.startswith("+"))
|
|
context = len(lines) - removed - added
|
|
text = (f"diff --git a/{old} b/{path}\n--- a/{old}\n+++ b/{path}\n"
|
|
f"@@ -1,{context + removed} +1,{context + added} @@\n" + "\n".join(lines) + "\n")
|
|
gitea = {"filename": path, "status": status, "additions": added, "deletions": removed}
|
|
if previous:
|
|
gitea["previous_filename"] = previous
|
|
return gitea, text
|
|
|
|
|
|
def label(path, lines, cfg=None, **kw):
|
|
gitea, text = hunk_diff(path, lines, **kw)
|
|
cfg = cfg or config()
|
|
return mechanical_files(files_from_gitea([gitea]), parse_unified_diff(text), cfg.mechanical).get(path, "")
|
|
|
|
|
|
JAVA_IMPORTS = [" package net.example;", " ", "-import java.util.List;", "+import java.util.Map;",
|
|
"+import java.util.function.*;", " ", " class B {"]
|
|
|
|
|
|
class ImportRuleTest(unittest.TestCase):
|
|
def test_java_and_kotlin_import_and_package_lines(self):
|
|
self.assertEqual(label("B.java", JAVA_IMPORTS), "imports")
|
|
self.assertEqual(label("B.java", ["-package net.old;", "+package net.example;", " ", " import a.B;"]),
|
|
"imports")
|
|
for path in ("B.kt", "build.gradle.kts"):
|
|
with self.subTest(path=path):
|
|
self.assertEqual(label(path, ["-import a.b.C", "+import a.b.D", " ", " fun f() = 1"]), "imports")
|
|
|
|
def test_js_family_named_default_and_namespace_imports(self):
|
|
lines = ["+import { a, b } from './x';", "+import type { T } from 'y'", "+import * as z from \"z\";",
|
|
"+import React, { useState } from 'react';", "-import Old from 'old';", " ", " export const x = 1;"]
|
|
for ext in ("ts", "tsx", "js", "jsx", "mjs"):
|
|
with self.subTest(ext=ext):
|
|
self.assertEqual(label(f"src/a.{ext}", lines), "imports")
|
|
|
|
def test_side_effect_import_never_matches(self):
|
|
for line in ("+import './polyfill';", "+import \"reflect-metadata\""):
|
|
with self.subTest(line=line):
|
|
self.assertEqual(label("src/main.ts", [" import { a } from './a';", line]), "")
|
|
|
|
def test_multi_line_import_counts_when_the_whole_statement_is_in_the_diff(self):
|
|
lines = [" import {", " a,", "+ b,", " } from './x';", " ", " export const y = a;"]
|
|
self.assertEqual(label("src/a.ts", lines), "imports")
|
|
|
|
def test_multi_line_import_cut_off_by_the_hunk_is_not_mechanical(self):
|
|
self.assertEqual(label("src/a.ts", [" a,", "+ b,", " } from './x';", " "]), "")
|
|
|
|
def test_go_import_block_and_single_import(self):
|
|
block = [" import (", ' \t"fmt"', '+\t"os"', '+\tlog "github.com/x/log"', " )"]
|
|
self.assertEqual(label("main.go", block), "imports")
|
|
self.assertEqual(label("main.go", [" package main", " ", '+import "strings"', " ", " func main() {"]),
|
|
"imports")
|
|
|
|
def test_go_blank_identifier_and_dot_imports_never_match(self):
|
|
for line in ('+\t_ "embed"', '+\t. "math"'):
|
|
with self.subTest(line=line):
|
|
self.assertEqual(label("main.go", [" import (", ' \t"fmt"', line, " )"]), "")
|
|
self.assertEqual(label("main.go", [" package main", '+import _ "net/http/pprof"']), "")
|
|
|
|
def test_go_import_line_outside_a_visible_import_block_is_not_mechanical(self):
|
|
self.assertEqual(label("main.go", [' \t"fmt"', '+\t"os"', " )"]), "")
|
|
|
|
def test_rust_use_matches_and_mod_never(self):
|
|
self.assertEqual(label("src/lib.rs", ["+use std::io::{Read, Write};", "+pub use crate::a::*;", " "]),
|
|
"imports")
|
|
self.assertEqual(label("src/lib.rs", ["+mod helpers;", "+use std::io;", " "]), "")
|
|
|
|
def test_python_and_unknown_extensions_never_match_the_import_rule(self):
|
|
self.assertEqual(label("app/a.py", ["+import os", "+from x import y", " "]), "")
|
|
self.assertEqual(label("app/A.scala", ["+import a.b.C", " "]), "")
|
|
|
|
def test_one_non_import_line_makes_the_file_not_mechanical(self):
|
|
self.assertEqual(label("B.java", JAVA_IMPORTS + ["+ int x = 1;"]), "")
|
|
|
|
def test_blank_lines_in_import_files_are_mechanical(self):
|
|
self.assertEqual(label("B.java", [" import a.B;", "+import a.C;", "+", " ", " class B {"]), "imports")
|
|
|
|
|
|
class WhitespaceRuleTest(unittest.TestCase):
|
|
def test_indentation_trailing_whitespace_and_blank_lines(self):
|
|
lines = [" void f() {", "- foo();", "+ foo();", "- bar(); ", "+ bar();", "+", " }"]
|
|
self.assertEqual(label("src/a.c", lines), "whitespace")
|
|
self.assertEqual(label("src/A.java", lines), "whitespace")
|
|
|
|
def test_imports_and_whitespace_together(self):
|
|
lines = ["+import a.C;", " ", " class B {", "- int x;", "+ int x;", " }"]
|
|
self.assertEqual(label("B.java", lines), "imports, whitespace")
|
|
|
|
def test_space_added_inside_a_line_is_not_whitespace_only(self):
|
|
self.assertEqual(label("src/a.c", ['-s = "ab";', '+s = "a b";']), "")
|
|
|
|
def test_moved_line_is_not_whitespace_only(self):
|
|
self.assertEqual(label("src/a.c", ["-a();", " b();", "+a();"]), "")
|
|
|
|
def test_whitespace_significant_files_are_never_whitespace_mechanical(self):
|
|
for path in ("app/a.py", ".gitea/ci.yml", "k8s/x.yaml", "Makefile", "rules.mk", "README.md", "v.haml",
|
|
"v.pug", "a.coffee", "a.nim", "a.fs"):
|
|
with self.subTest(path=path):
|
|
self.assertEqual(label(path, [" a:", "- x", "+ x"]), "")
|
|
self.assertEqual(label(path, [" a", "+", " b"]), "")
|
|
|
|
|
|
class RenameRuleTest(unittest.TestCase):
|
|
def rename(self, additions=0, deletions=0, diff=""):
|
|
gitea = {"filename": "new/A.java", "previous_filename": "old/A.java", "status": "renamed",
|
|
"additions": additions, "deletions": deletions}
|
|
text = diff or ("diff --git a/old/A.java b/new/A.java\nsimilarity index 100%\n"
|
|
"rename from old/A.java\nrename to new/A.java\n")
|
|
return mechanical_files(files_from_gitea([gitea]), parse_unified_diff(text), config().mechanical)
|
|
|
|
def test_rename_with_identical_content(self):
|
|
self.assertEqual(self.rename(), {"new/A.java": "renamed"})
|
|
|
|
def test_rename_with_changes_is_not_mechanical(self):
|
|
_, text = hunk_diff("new/A.java", ["-package old;", "+package new;"], previous="old/A.java")
|
|
self.assertEqual(self.rename(1, 1, text), {})
|
|
|
|
def test_added_and_deleted_files_are_not_mechanical(self):
|
|
self.assertEqual(label("B.java", ["+import a.B;"], status="added"), "")
|
|
self.assertEqual(label("B.java", ["-import a.B;"], status="deleted"), "")
|
|
|
|
def test_diff_shorter_than_the_counts_of_gitea_is_not_mechanical(self):
|
|
gitea, text = hunk_diff("B.java", ["+import a.C;"])
|
|
gitea["additions"] = 40
|
|
self.assertEqual(mechanical_files(files_from_gitea([gitea]), parse_unified_diff(text),
|
|
config().mechanical), {})
|
|
|
|
def test_switched_off(self):
|
|
self.assertEqual(label("B.java", JAVA_IMPORTS, config(mechanical={"enabled": False})), "")
|
|
|
|
|
|
def evaluate(*parts, **overrides):
|
|
files = [gitea for gitea, _ in parts]
|
|
text = "".join(diff for _, diff in parts)
|
|
return evaluate_rules(files_from_gitea(files), parse_unified_diff(text), config(**overrides))
|
|
|
|
|
|
class RulesTest(unittest.TestCase):
|
|
def test_import_only_change_in_a_test_file_is_not_test_weakening(self):
|
|
outcome = evaluate(hunk_diff(MAIN, [" int f() {", "- return 1;", "+ return 2;", " }"]),
|
|
hunk_diff(TEST, ["-import org.junit.Assert;", "+import org.assertj.core.api.Assertions;",
|
|
" ", " class ATest {"]))
|
|
self.assertNotIn("tests_weakened", [r.code for r in outcome.reasons])
|
|
self.assertEqual(outcome.change_kind, "refactoring")
|
|
self.assertEqual(outcome.mechanical, {TEST: "imports"})
|
|
|
|
def test_mechanical_files_do_not_count_for_the_size_limits(self):
|
|
parts = [hunk_diff(f"src/main/java/M{i}.java", JAVA_IMPORTS) for i in range(31)]
|
|
parts.append(hunk_diff(MAIN, [" int f() {", "- return 1;", "+ return 2;", " }"]))
|
|
outcome = evaluate(*parts)
|
|
self.assertEqual((outcome.changed_lines, len(outcome.mechanical)), (2, 31))
|
|
self.assertEqual([r.code for r in outcome.reasons], [])
|
|
|
|
def test_all_mechanical_files_give_the_approvable_kind_mechanical_only(self):
|
|
outcome = evaluate(hunk_diff(IMPORTS_ONLY, JAVA_IMPORTS), hunk_diff(TEST, ["-import a.B;", " class ATest {"]))
|
|
self.assertEqual(outcome.change_kind, "mechanical_only")
|
|
self.assertTrue(outcome.auto_approvable)
|
|
self.assertEqual(outcome.reasons, [])
|
|
|
|
|
|
def run(gitea, model=None, cfg=None):
|
|
model = model or FakeModel(prompt_files=[f["filename"] for f in gitea.files])
|
|
d = deps(gitea, model)
|
|
outcome = execute(cfg or config(gate={"yellow_gate": "block"}), pr_ref(), d, Redactor())
|
|
return outcome, model
|
|
|
|
|
|
def gitea_with(*parts):
|
|
gitea = FakeGitea(files=[g for g, _ in parts], diff="".join(d for _, d in parts))
|
|
gitea.raw_files.update({g["filename"]: f"content of {g['filename']}\n".encode() for g, _ in parts})
|
|
return gitea
|
|
|
|
|
|
class FlowTest(unittest.TestCase):
|
|
def test_all_mechanical_pr_is_green_without_model_call(self):
|
|
outcome, model = run(gitea_with(hunk_diff(IMPORTS_ONLY, JAVA_IMPORTS),
|
|
hunk_diff("src/main/c/a.c", ["- f();", "+ f();"])))
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertEqual(model.calls, [])
|
|
self.assertEqual(outcome.result.change_kind, "mechanical_only")
|
|
self.assertEqual(outcome.result.llm_status, "skipped (mechanical changes only)")
|
|
self.assertEqual(outcome.result.cost_status, "none")
|
|
|
|
def test_mechanical_files_stay_out_of_the_prompt_and_the_coverage_check(self):
|
|
gitea = gitea_with(hunk_diff(MAIN, [" int f() {", "- return 1;", "+ return 2;", " }"]),
|
|
hunk_diff(IMPORTS_ONLY, JAVA_IMPORTS))
|
|
cfg = config(gate={"yellow_gate": "block"})
|
|
self.assertTrue(cfg.review.full_files)
|
|
outcome, model = run(gitea, FakeModel(prompt_files=[MAIN]), cfg)
|
|
self.assertEqual(model.excluded, [(IMPORTS_ONLY,)])
|
|
self.assertIn(f"content of {MAIN}", model.sections[0])
|
|
self.assertNotIn(IMPORTS_ONLY, model.sections[0])
|
|
self.assertNotIn("skipped_files", [r.code for r in outcome.result.reasons])
|
|
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
|
self.assertIn("do not report them as missing", model.calls[0])
|
|
self.assertIn(f"1 file changed only mechanically and was not reviewed by the model: "
|
|
f"{IMPORTS_ONLY} (imports).", gitea.created_comments[0])
|
|
|
|
def test_comment_lists_many_mechanical_files_in_one_line(self):
|
|
parts = [hunk_diff(f"src/main/java/M{i}.java", JAVA_IMPORTS) for i in range(4)]
|
|
parts.append(hunk_diff("src/main/c/a.c", ["- f();", "+ f();"]))
|
|
gitea = gitea_with(*parts)
|
|
run(gitea)
|
|
self.assertIn("5 files changed only mechanically and were not reviewed by the model: "
|
|
"src/main/c/a.c (whitespace), src/main/java/M0.java (imports), src/main/java/M1.java (imports) "
|
|
"and 2 more.", gitea.created_comments[0])
|
|
|
|
def test_mechanical_change_of_a_risky_path_stays_yellow(self):
|
|
path = "src/main/java/net/example/deploy/Deploy.java"
|
|
outcome, model = run(gitea_with(hunk_diff(path, JAVA_IMPORTS)))
|
|
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
|
self.assertIn("risky_paths", [r.code for r in outcome.result.reasons])
|
|
self.assertEqual(model.calls, [])
|
|
|
|
def test_mechanical_change_of_a_critical_path_stays_orange(self):
|
|
path = "src/main/java/net/example/auth/Login.java"
|
|
outcome, model = run(gitea_with(hunk_diff(path, JAVA_IMPORTS)))
|
|
self.assertEqual(outcome.result.change_kind, "mechanical_only")
|
|
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
|
self.assertIn("critical_paths", [r.code for r in outcome.result.reasons])
|
|
self.assertEqual(model.calls, [])
|
|
|
|
def test_mechanical_only_must_be_auto_approvable(self):
|
|
cfg = config(gate={"yellow_gate": "block"}, categories={"auto_approvable": ["docs_only"]})
|
|
outcome, _ = run(gitea_with(hunk_diff(IMPORTS_ONLY, JAVA_IMPORTS)), cfg=cfg)
|
|
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
|
self.assertIn("category", [r.code for r in outcome.result.reasons])
|
|
|
|
|
|
def host_data():
|
|
with open(os.path.join(ROOT, "config", "werkprobe.toml"), "rb") as f:
|
|
return tomllib.load(f)
|
|
|
|
|
|
class ConfigTest(unittest.TestCase):
|
|
def test_invalid_regex_is_a_config_error(self):
|
|
with self.assertRaisesRegex(ConfigError, r"\[mechanical\] imports.java has an invalid regex '\(unclosed'"):
|
|
config(mechanical={"imports": {"java": ["(unclosed"]}})
|
|
with self.assertRaisesRegex(ConfigError, r"\[mechanical\] imports.java has an invalid regex"):
|
|
build_config(host_data(), {"mechanical": {"imports": {"java": ["[a-"]}}})
|
|
|
|
def test_invalid_values_are_config_errors(self):
|
|
for values, message in (({"enabled": "yes"}, "enabled must be true or false"),
|
|
({"imports": {".java": ["x"]}}, "a file extension without the dot"),
|
|
({"imports": {"java": "x"}}, "must be a list of non-empty regexes"),
|
|
({"whitespace_significant": [""]}, "must be a list of non-empty strings"),
|
|
({"unknown": 1}, "has unknown keys: unknown")):
|
|
with self.subTest(values=values):
|
|
with self.assertRaisesRegex(ConfigError, message):
|
|
config(mechanical=values)
|
|
|
|
def test_section_is_required(self):
|
|
host = host_data()
|
|
del host["mechanical"]
|
|
with self.assertRaisesRegex(ConfigError, r"sections missing: \[mechanical\]"):
|
|
build_config(host)
|
|
|
|
def test_layers_merge_the_imports_per_extension(self):
|
|
host = host_data()
|
|
host["mechanical"]["imports"] = {"scala": [r"^import [\w.]+$"]}
|
|
cfg = build_config(host, {"mechanical": {"imports": {"java": []}}})
|
|
self.assertEqual(cfg.mechanical.imports["scala"], (r"^import [\w.]+$",))
|
|
self.assertEqual(cfg.mechanical.imports["java"], ())
|
|
self.assertTrue(cfg.mechanical.imports["ts"])
|
|
self.assertEqual(label("A.scala", ["+import a.b", " "], cfg), "imports")
|
|
self.assertEqual(label("B.java", JAVA_IMPORTS, cfg), "")
|
|
|
|
def test_repository_may_switch_off_and_replace_the_whitespace_list(self):
|
|
merged = merge_repository_layer(host_data(), {"mechanical": {"enabled": False,
|
|
"whitespace_significant": ["*.c"]}})
|
|
cfg = build_config(merged)
|
|
self.assertEqual((cfg.mechanical.enabled, cfg.mechanical.whitespace_significant), (False, ("*.c",)))
|
|
|
|
def test_example_configs(self):
|
|
for name in ("werkprobe.toml", "hsadmin-ng.toml", "werkrevisor.toml"):
|
|
with self.subTest(name=name):
|
|
cfg = example_config(name)
|
|
self.assertTrue(cfg.mechanical.enabled)
|
|
self.assertIn("mechanical_only", cfg.categories.auto_approvable)
|
|
self.assertEqual(cfg.model.token_count_factor, 1.55)
|
|
self.assertEqual(cfg.engine.mode, "combined")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|