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): """Scenario#112.04""" 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_only_package_and_import_changes(self): """Scenario#112.01""" _, text = hunk_diff("new/A.java", ["-package old;", "+package new;", " ", "-import old.B;", "+import new.B;", " ", " class A {"], previous="old/A.java") self.assertEqual(self.rename(2, 2, text), {"new/A.java": "imports, renamed"}) def test_rename_with_a_logic_change_is_not_mechanical(self): """Scenario#112.02""" _, text = hunk_diff("new/A.java", ["-package old;", "+package new;", " class A {", "- int f() { return 1; }", "+ int f() { return 2; }", " }"], previous="old/A.java") self.assertEqual(self.rename(2, 2, text), {}) def test_rename_keeps_the_rules_of_the_old_name(self): """Scenario#112.02""" for old, new in (("src/worker.py", "src/worker"), ("old/A.py", "new/A.java")): with self.subTest(old=old): gitea, text = hunk_diff(new, [" if ready:", "- process()", "+process()"], status="renamed", previous=old) self.assertEqual(mechanical_files(files_from_gitea([gitea]), parse_unified_diff(text), config().mechanical), {}) 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_moved_files_do_not_count_for_the_size_limits(self): """Scenario#112.01""" parts = [hunk_diff(f"src/main/java/net/javagil/M{i}.java", ["-package net.example;", "+package net.javagil;", " ", " class M {"], status="renamed", previous=f"src/main/java/net/example/M{i}.java") 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_rename_into_another_category_is_not_mechanical(self): """Scenario#112.03""" moved = hunk_diff(MAIN, ["-package net.test;", "+package net.example;", " ", " class A {"], status="renamed", previous=TEST) outcome = evaluate(moved) self.assertEqual(outcome.mechanical, {}) self.assertEqual(outcome.change_kind, "refactoring") 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/auth/Login.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_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): # MiniMax M3 counts close to o200k_base, Anthropic counts more. for name, factor in (("werkprobe.toml", 1.55), ("hsadmin-ng.toml", 1.55), ("werkrevisor.toml", 1.05), ("werkator.toml", 1.05)): 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, factor) self.assertEqual(cfg.engine.mode, "combined") if __name__ == "__main__": unittest.main()