import json import os import tomllib import unittest from tests.fakes import FakeGitea, FakeJev, FakeModel, ROOT, config, diff_for, gitea_file, review_data from tests.test_flow import run from werkrevisor import pathjudge from werkrevisor.cli import Superseded from werkrevisor.config import ConfigError, build_config from werkrevisor.verdict import Verdict BUILD = "server/build.gradle.kts" def judged_config(hard=(), **path_judge): return config(engine={"mode": "combined"}, gate={"yellow_gate": "block"}, hard_yellow={"paths": list(hard), "judged_paths": ["server/*.gradle.kts"]}, path_judge={"judge": "jev", **path_judge}) def build_gitea(added=('version = "0.18.5"',), path=BUILD, previous=None): f = gitea_file(path, status="renamed" if previous else "modified") if previous: f["previous_filename"] = previous return FakeGitea(files=[f], diff=diff_for(path, added=list(added))) def jev_for(weakens, risk, unit=BUILD): return FakeJev(problems={(unit, "weakens_checks"): weakens, (unit, "build_risk"): risk / pathjudge.RISK_TOP}) def reason_codes(outcome): return {r.code: r.verdict for r in outcome.result.reasons} class PathJudgeFlowTest(unittest.TestCase): def test_jev_answers_decide_the_verdict(self): for weakens, risk, code, verdict in ((0.02, 0.00, "judged_paths_judged", Verdict.GREEN), (0.03, 1.30, "judged_paths_judged", Verdict.GREEN), (0.18, 1.93, "judged_paths_risky", Verdict.YELLOW), (0.59, 1.00, "judged_paths_risky", Verdict.YELLOW), (0.98, 2.70, "judged_paths_weakened", Verdict.ORANGE), (0.13, 3.00, "judged_paths_weakened", Verdict.ORANGE)): with self.subTest(weakens=weakens, risk=risk): jev = jev_for(weakens, risk) outcome, _ = run(build_gitea(), FakeModel(prompt_files=[BUILD]), cfg=judged_config(), jev=jev) self.assertIs(outcome.result.verdict, verdict) self.assertIs(reason_codes(outcome)[code], verdict) self.assertNotIn("risky_paths", reason_codes(outcome)) state, questions = next((s, q) for s, q in jev.requests if "weakens_checks" in q) self.assertEqual(state, f'Build file: {BUILD}\n\nDiff:\n+version = "0.18.5"') # Without scope "all", a judged path keeps the two questions it was calibrated with. self.assertEqual(set(questions), {"weakens_checks", "build_risk"}) def test_orange_reason_names_the_values(self): outcome, _ = run(build_gitea(), FakeModel(prompt_files=[BUILD]), cfg=judged_config(), jev=jev_for(0.98, 2.7)) reason = next(r for r in outcome.result.reasons if r.code == "judged_paths_weakened") self.assertEqual(reason.text, "build changes that weaken checks or add risk according to Jev: " f"{BUILD} (weakens checks 0.98, risk 2.70)") def test_unjudged_paths_give_yellow(self): cases = (("Jev fails", build_gitea(), FakeJev(error=OSError("down")), {}, "OSError: down"), ("--no-llm", build_gitea(), FakeJev(), {"use_llm": False}, "(--no-llm)"), ("secret", build_gitea(added=["-----BEGIN " + "PRIVATE KEY-----"]), FakeJev(), {}, "(possible secrets in the diff)")) for name, gitea, jev, kw, text in cases: with self.subTest(name): outcome, _ = run(gitea, FakeModel(prompt_files=[BUILD]), cfg=judged_config(), jev=jev, **kw) reason = next(r for r in outcome.result.reasons if r.code == "judged_paths_not_judged") self.assertIs(reason.verdict, Verdict.YELLOW) self.assertTrue(reason.text.startswith("build files not judged by Jev")) self.assertIn(text, reason.text) self.assertNotEqual(outcome.result.verdict, Verdict.GREEN) if name != "Jev fails": self.assertEqual([q for _, q in jev.requests if "weakens_checks" in q], []) def test_deleted_or_empty_judged_files_are_not_judged(self): deleted = FakeGitea(files=[gitea_file(BUILD, status="removed")], diff=diff_for(BUILD, removed=["x"], deleted=True)) empty = FakeGitea(files=[gitea_file(BUILD)], diff="") for name, gitea, text in (("deleted", deleted, f"{BUILD} (deleted)"), ("empty", empty, f"{BUILD} (no text diff)")): with self.subTest(name): jev = jev_for(0.0, 0.0) outcome, _ = run(gitea, FakeModel(prompt_files=[BUILD]), cfg=judged_config(), jev=jev) reason = next(r for r in outcome.result.reasons if r.code == "judged_paths_not_judged") self.assertEqual(reason.text, f"build files not judged by Jev: {text}") self.assertNotEqual(outcome.result.verdict, Verdict.GREEN) self.assertEqual([q for _, q in jev.requests if "weakens_checks" in q], []) def test_a_renamed_judged_file_is_a_risky_path(self): jev = jev_for(0.0, 0.0) gitea = build_gitea(path="docs/build.gradle.kts.md", previous=BUILD) outcome, _ = run(gitea, FakeModel(prompt_files=["docs/build.gradle.kts.md"]), cfg=judged_config(), jev=jev) self.assertIn(f"{BUILD} (server/*.gradle.kts)", next(r.text for r in outcome.result.reasons if r.code == "risky_paths")) self.assertEqual([q for _, q in jev.requests if "weakens_checks" in q], []) def test_hard_paths_win(self): cases = (("hard path of the config", judged_config(hard=["server/build.gradle*"]), build_gitea()), ("built-in hard path", judged_config(), build_gitea(path=".werkrevisor.toml")), ("renamed from a hard path", judged_config(hard=["legacy/**"]), build_gitea(previous="legacy/build.gradle.kts"))) for name, cfg, gitea in cases: with self.subTest(name): jev = jev_for(0.0, 0.0) outcome, _ = run(gitea, FakeModel(prompt_files=[f["filename"] for f in gitea.files]), cfg=cfg, jev=jev) self.assertIs(reason_codes(outcome)["risky_paths"], Verdict.YELLOW) self.assertEqual([q for _, q in jev.requests if "weakens_checks" in q], []) def test_never_judge_gives_yellow(self): jev = jev_for(0.0, 0.0) cfg = judged_config(judge="never") outcome, _ = run(build_gitea(), FakeModel(prompt_files=[BUILD]), cfg=cfg, jev=jev) self.assertIs(outcome.result.verdict, Verdict.YELLOW) reason = next(r for r in outcome.result.reasons if r.code == "risky_paths") self.assertIn(f"{BUILD} (server/*.gradle.kts)", reason.text) self.assertEqual([q for _, q in jev.requests if "weakens_checks" in q], []) def test_usage_record_holds_the_jev_values(self): gitea = build_gitea() _, d = run(gitea, FakeModel(prompt_files=[BUILD]), cfg=judged_config(), jev=jev_for(0.02, 1.3)) record = json.loads(d.out.getvalue().strip()) [judgment] = record["path_judgments"] self.assertEqual((judgment["file"], judgment["verdict"]), (BUILD, "GREEN")) self.assertAlmostEqual(judgment["build_risk"], 1.3) self.assertEqual((record["path_judge_calls"], record["path_judge_input_tokens"]), (1, 1000)) self.assertIn("build file judge `jev-test` calls 1", gitea.created_comments[0]) APP = "src/main/java/App.java" APP_TEST = "src/test/java/AppTest.java" def all_config(mode="combined", model_role="advises"): return config(engine={"mode": mode, "model_role": model_role}, gate={"yellow_gate": "block"}, hard_yellow={"paths": [], "judged_paths": ["server/*.gradle.kts"]}, path_judge={"judge": "jev", "scope": "all"}) def all_gitea(*paths): return FakeGitea(files=[gitea_file(p) for p in paths], diff="".join(diff_for(p, added=[f"line of {p}"]) for p in paths)) def all_run(jev, paths=(BUILD, APP, APP_TEST), model=None, mode="combined", model_role="advises"): model = model or FakeModel(prompt_files=list(paths)) return run(all_gitea(*paths), model, cfg=all_config(mode, model_role), jev=jev) class JudgeAllFilesTest(unittest.TestCase): def test_a_harmless_judged_path_needs_no_human(self): jev = FakeJev(problems={(BUILD, "weakens_checks"): 0.02, (BUILD, "build_risk"): 0.1}) outcome, _ = all_run(jev) build_questions = next(q for s, q in jev.requests if s.startswith(f"Build file: {BUILD}")) self.assertEqual(set(build_questions), {"weakens_checks", "build_risk", "security"}) self.assertIs(outcome.result.verdict, Verdict.GREEN) codes = reason_codes(outcome) self.assertIs(codes["judged_paths_judged"], Verdict.GREEN) self.assertIs(codes["files_judged"], Verdict.GREEN) self.assertNotIn("risky_paths", codes) def test_a_change_that_touches_security_needs_a_human(self): jev = FakeJev(problems={(APP, "security"): 0.97, (APP_TEST, "security"): 0.99}) outcome, _ = all_run(jev) self.assertIs(outcome.result.verdict, Verdict.YELLOW) reason = next(r for r in outcome.result.reasons if r.code == "files_touch_security") self.assertEqual(reason.text, f"changes that touch security according to Jev: {APP} (weakens checks 0.00, " "security 0.97)") # A test that names a role tests the access rule, so Jev is not asked about security there. test_questions = next(q for s, q in jev.requests if s.startswith(f"Changed file: {APP_TEST}")) self.assertEqual(set(test_questions), {"weakens_checks"}) def test_a_build_file_that_touches_security_is_named_as_such(self): outcome, _ = all_run(FakeJev(problems={(BUILD, "security"): 0.96})) codes = reason_codes(outcome) self.assertIs(codes["files_touch_security"], Verdict.YELLOW) self.assertNotIn("judged_paths_risky", codes) self.assertIn(f"{BUILD} (weakens checks 0.00, risk 0.00, security 0.96)", next(r.text for r in outcome.result.reasons if r.code == "files_touch_security")) def test_a_change_that_weakens_the_checks_gets_orange(self): outcome, _ = all_run(FakeJev(problems={(APP_TEST, "weakens_checks"): 0.9})) self.assertIs(outcome.result.verdict, Verdict.ORANGE) self.assertIs(reason_codes(outcome)["files_weaken_checks"], Verdict.ORANGE) def test_doubts_of_the_model_are_hints_when_jev_judged_all_files(self): data = review_data(risk="high", issues=[APP], header="[medium] Possible Bug") unit = f"{APP}:3-5" jev = FakeJev(problems={(unit, "real_defect"): 0.2, (unit, "severity"): 0.9}) outcome, _ = all_run(jev, model=FakeModel(data=data, prompt_files=[BUILD, APP, APP_TEST])) codes = reason_codes(outcome) self.assertIs(codes["finding_unconfirmed"], Verdict.GREEN) self.assertIs(codes["model_high_risk"], Verdict.GREEN) self.assertIs(outcome.result.verdict, Verdict.GREEN) self.assertIn("(advisory: Jev judged every changed file)", next(r.text for r in outcome.result.reasons if r.code == "model_high_risk")) def test_a_pr_too_large_for_the_model_is_a_hint_when_the_model_only_advises(self): for role, verdict in (("advises", Verdict.GREEN), ("decides", Verdict.YELLOW)): with self.subTest(role=role): cfg = config(engine={"mode": "combined", "model_role": role}, gate={"yellow_gate": "block"}, budget={"max_usd_per_review": 0.081}, hard_yellow={"paths": [], "judged_paths": ["server/*.gradle.kts"]}, path_judge={"judge": "jev", "scope": "all"}) model = FakeModel(prompt_files=[APP]) outcome, _ = run(all_gitea(APP), model, cfg=cfg, jev=FakeJev()) self.assertEqual(model.calls, []) self.assertIs(reason_codes(outcome)["too_large_tokens"], verdict) self.assertIs(outcome.result.verdict, verdict) def test_a_failed_advising_model_leaves_the_verdict_to_jev(self): for role, triage_ok, verdict in (("advises", True, Verdict.GREEN), ("advises", False, Verdict.YELLOW), ("decides", True, Verdict.ERROR)): with self.subTest(role=role, triage_ok=triage_ok): jev = FakeJev() if triage_ok else FakeJev(error=OSError("down")) model = FakeModel(error=TimeoutError("review exceeded 720 seconds"), prompt_files=[BUILD, APP, APP_TEST]) outcome, _ = all_run(jev, model=model, model_role=role) self.assertIs(outcome.result.verdict, verdict) if role == "advises": self.assertIs(reason_codes(outcome)["model_failed"], verdict) self.assertEqual(outcome.result.budget_remaining_after, 10.0) def test_a_superseded_head_still_ends_the_review_when_the_model_only_advises(self): model = FakeModel(error=Superseded("a newer head"), prompt_files=[BUILD, APP, APP_TEST]) outcome, _ = all_run(FakeJev(), model=model) self.assertTrue(outcome.superseded) self.assertNotIn("model_failed", reason_codes(outcome)) def test_a_deleted_file_is_judged_on_its_removed_lines(self): files = [gitea_file(APP, status="removed", additions=0, deletions=1), gitea_file(APP_TEST, status="removed", additions=0, deletions=1)] diff = diff_for(APP, removed=["check(role)"], deleted=True) + diff_for(APP_TEST, removed=["test()"], deleted=True) jev = FakeJev(problems={(APP, "security"): 0.97}) outcome, _ = run(FakeGitea(files=files, diff=diff), FakeModel(prompt_files=[]), cfg=all_config(), jev=jev) codes = reason_codes(outcome) self.assertIs(codes["files_touch_security"], Verdict.YELLOW) # A deleted test file is a lost test, whatever Jev answers. self.assertIs(codes["tests_weakened"], Verdict.YELLOW) self.assertIn(f"Changed file: {APP}\n\nDiff:\n-check(role)", [s for s, _ in jev.requests]) def test_by_default_the_model_decides_as_before(self): data = review_data(risk="high", issues=[APP], header="[medium] Possible Bug") unit = f"{APP}:3-5" jev = FakeJev(problems={(unit, "real_defect"): 0.2, (unit, "severity"): 0.9}) outcome, _ = all_run(jev, model=FakeModel(data=data, prompt_files=[BUILD, APP, APP_TEST]), model_role="decides") codes = reason_codes(outcome) self.assertIs(codes["finding_unconfirmed"], Verdict.YELLOW) self.assertIs(codes["model_high_risk"], Verdict.YELLOW) self.assertIs(outcome.result.verdict, Verdict.YELLOW) self.assertEqual(config().engine.model_role, "decides") def test_the_model_still_holds_a_pr_on_security_or_when_jev_missed_a_file(self): security = review_data(security="Yes, the token is logged.") outcome, _ = all_run(FakeJev(), model=FakeModel(data=security, prompt_files=[BUILD, APP, APP_TEST])) self.assertIs(reason_codes(outcome)["model_security"], Verdict.YELLOW) risky = review_data(risk="high") outcome, _ = all_run(FakeJev(error=OSError("down")), model=FakeModel(data=risky, prompt_files=[BUILD, APP, APP_TEST])) codes = reason_codes(outcome) self.assertIs(codes["judged_paths_not_judged"], Verdict.YELLOW) self.assertIs(codes["model_high_risk"], Verdict.YELLOW) def test_without_a_model_jev_is_enough_for_green(self): model = FakeModel(prompt_files=[BUILD, APP, APP_TEST]) outcome, _ = all_run(FakeJev(), model=model, mode="classifier") self.assertIs(outcome.result.verdict, Verdict.GREEN) self.assertEqual(model.calls, []) self.assertNotIn("no_model_review", reason_codes(outcome)) # Without the judgment of every file, a PR of code still needs the model or a human. outcome, _ = run(all_gitea(APP), FakeModel(prompt_files=[APP]), cfg=config(engine={"mode": "classifier"}, gate={"yellow_gate": "block"}), jev=FakeJev()) self.assertIs(outcome.result.verdict, Verdict.YELLOW) class PathJudgeConfigTest(unittest.TestCase): def host(self, **path_judge): with open(os.path.join(ROOT, "config", "werkprobe.toml"), "rb") as f: data = tomllib.load(f) if path_judge: data["path_judge"] = path_judge return data def test_repository_may_enable_the_judge_only_if_the_host_allows_it(self): repo = {"hard_yellow": {"judged_paths": ["server/*.gradle.kts"]}, "path_judge": {"judge": "jev"}} cfg = build_config(self.host(), repo) self.assertEqual((cfg.path_judge.judge, cfg.hard_yellow.judged_paths), ("jev", ("server/*.gradle.kts",))) with self.assertRaisesRegex(ConfigError, r"must not set \[path_judge\] judge = 'jev': the host config does not"): build_config(self.host(repository_may_enable=False), repo) # Switching the judge off only makes the review stricter. cfg = build_config(self.host(judge="jev", repository_may_enable=False), {"path_judge": {"judge": "never"}}) self.assertEqual(cfg.path_judge.judge, "never") def test_thresholds_and_the_switch_are_host_only(self): # risk_template is no longer host only since PR#63; the switch that allows choosing it is. for key in ("repository_may_enable", "orange_weakens_min", "green_risk_max", "scope", "yellow_security_min", "repository_may_tune_risks"): with self.subTest(key=key): with self.assertRaisesRegex(ConfigError, rf"must not set \[path_judge\] {key}: only the host config"): build_config(self.host(), {"path_judge": {key: 1}}) def test_values_are_validated(self): for values, message in (({"judge": "maybe"}, r"\[path_judge\] judge must be one of never, jev"), ({"orange_risk_min": 4}, r"orange_risk_min must be a number between 0 and 3"), ({"green_weakens_max": 1.5}, r"green_weakens_max must be a number between 0 and 1"), ({"repository_may_enable": "yes"}, r"repository_may_enable must be true or false"), ({"repository_may_tune_risks": "yes"}, r"repository_may_tune_risks must be true or false"), ({"judge": "jev", "scope": "some"}, r"scope must be one of judged_paths, all"), ({"scope": "all"}, r"scope = 'all' needs judge = 'jev'"), ({"yellow_security_min": 2}, r"yellow_security_min must be a number between 0 and 1")): with self.subTest(values=values): with self.assertRaisesRegex(ConfigError, message): config(path_judge=values) host = self.host(judge="jev") del host["classifier"] host["engine"]["mode"] = "llm" with self.assertRaisesRegex(ConfigError, r"\[path_judge\] judge = 'jev' needs a \[classifier\] section"): build_config(host) self.assertEqual((config().path_judge.judge, config().path_judge.repository_may_enable), ("never", True)) if __name__ == "__main__": unittest.main()