322 lines
19 KiB
Python
322 lines
19 KiB
Python
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()
|