285 lines
17 KiB
Python
285 lines
17 KiB
Python
import math
|
||
import os
|
||
import tempfile
|
||
import unittest
|
||
|
||
from tests.fakes import ROOT, config, example_config, gitea_file
|
||
from werkrevisor import budget
|
||
from werkrevisor.config import ConfigError, load_config, pr_ref_from_env, pr_ref_from_url
|
||
from werkrevisor.redact import Redactor
|
||
from werkrevisor.rules import evaluate_rules, files_from_gitea
|
||
from werkrevisor.verdict import Verdict
|
||
|
||
|
||
class ConfigTest(unittest.TestCase):
|
||
def test_examples_load(self):
|
||
for name, gate in (("werkprobe.toml", "approval"), ("hsadmin-ng.toml", "block"), ("werkator.toml", "approval"),
|
||
("werkcrappit.toml", "approval"), ("werkregie.toml", "approval"),
|
||
("werkbaum.toml", "approval"), ("werkkompass.toml", "approval"),
|
||
("werkdock.toml", "approval")):
|
||
with self.subTest(name=name):
|
||
cfg = example_config(name)
|
||
self.assertEqual(cfg.gate.yellow_gate, gate)
|
||
|
||
def test_example_input_limits_and_their_worst_case_fit_the_budget(self):
|
||
for name, tokens in (("werkprobe.toml", 100000), ("werkrevisor.toml", 150000), ("werkator.toml", 150000), ("werkcrappit.toml", 150000),
|
||
("werkregie.toml", 150000), ("werkbaum.toml", 150000), ("werkkompass.toml", 150000),
|
||
("werkdock.toml", 150000),
|
||
("hsadmin-ng.toml", 150000)):
|
||
with self.subTest(name=name):
|
||
cfg = example_config(name)
|
||
limit = budget.input_token_limit(cfg.model, cfg.budget)
|
||
self.assertEqual(limit.tokens, tokens)
|
||
m = cfg.model
|
||
worst = m.attempts * (limit.tokens * m.input_price_usd_per_mtok
|
||
+ m.max_output_tokens * m.output_price_usd_per_mtok) / 1e6
|
||
self.assertLessEqual(worst, cfg.budget.max_usd_per_review)
|
||
|
||
def test_example_full_files_leave_room_for_the_diff_of_a_pr_at_the_size_limit(self):
|
||
for name in ("werkprobe.toml", "werkrevisor.toml", "werkator.toml", "werkcrappit.toml", "werkregie.toml",
|
||
"werkbaum.toml", "werkkompass.toml", "werkdock.toml", "hsadmin-ng.toml"):
|
||
with self.subTest(name=name):
|
||
cfg = example_config(name)
|
||
diff_chars = cfg.size.max_changed_lines * 80
|
||
est = budget.estimate(diff_chars, cfg.review.max_full_files_chars, cfg.model, cfg.budget)
|
||
self.assertLessEqual(est.input_tokens, budget.input_token_limit(cfg.model, cfg.budget).tokens)
|
||
|
||
def test_examples_leave_the_countable_writing_rules_to_werkrevisor(self):
|
||
for name in ("werkprobe.toml", "werkrevisor.toml", "werkator.toml", "werkcrappit.toml", "werkregie.toml",
|
||
"werkbaum.toml", "werkkompass.toml", "werkdock.toml", "hsadmin-ng.toml"):
|
||
with self.subTest(name=name):
|
||
instructions = example_config(name).review.extra_instructions
|
||
self.assertIn("werkrevisor itself checks line length, sentences per line and sentences per paragraph",
|
||
instructions)
|
||
self.assertIn("never report these rules", instructions)
|
||
self.assertIn("report only bloated or mannered prose", instructions)
|
||
self.assertIn("unclear statements, and contradictions between the PR document and the code",
|
||
instructions)
|
||
self.assertNotIn("Judge documentation changes against the writing style", instructions)
|
||
self.assertIn("documentation appears only in the diff", instructions)
|
||
|
||
def test_the_plan_of_each_repository_is_documentation(self):
|
||
from werkrevisor.rules import categorize
|
||
for name, plan in (("werkrevisor.toml", "docs/werkrevisor.werkbaum"), ("werkator.toml", "docs/werkator.werkbaum"),
|
||
("werkcrappit.toml", "docs/werkcrappit.werkbaum"),
|
||
("werkregie.toml", "doc/werkregie.werkbaum")):
|
||
with self.subTest(name=name):
|
||
self.assertEqual(categorize(plan, example_config(name).categories), "docs")
|
||
|
||
def test_werkator_rules_with_its_host_config(self):
|
||
from werkrevisor.globs import first_match
|
||
cfg = example_config("werkator.toml")
|
||
self.assertEqual((cfg.pr_doc.globs, cfg.review.repo_context_files), (("docs/prs/20*-PR#*.md",), ("AGENTS.md",)))
|
||
self.assertEqual(cfg.classifier.pr_doc_globs, ("docs/prs/20*-PR#*.md",))
|
||
self.assertIsNone(first_match("docs/prs/TEAMPLATE.md", cfg.pr_doc.globs))
|
||
self.assertIsNotNone(first_match("docs/prs/2026-10-06-PR#127-agents-pr-workflow.md", cfg.pr_doc.globs))
|
||
self.assertEqual(cfg.pr_doc.open_question_headings, ("Open Questions",))
|
||
# The isolation of the builds stays hard, Jev judges the build and CI files and, through scope "all", the rest.
|
||
for path in ("werkdock/build-rootfs", "src/main/kotlin/de/hoennig/werkator/build/PullRequestTargets.kt"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNotNone(first_match(path, cfg.hard_yellow.paths))
|
||
for path in ("packaging/werkator.service", "build.gradle.kts", ".werkator.yml"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNone(first_match(path, cfg.hard_yellow.paths))
|
||
self.assertIsNotNone(first_match(path, cfg.hard_yellow.judged_paths))
|
||
for path in ("src/main/kotlin/de/hoennig/werkator/config/ConfigLoader.kt",
|
||
"src/main/kotlin/de/hoennig/werkator/server/UiFormats.kt"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNone(first_match(path, cfg.hard_yellow.paths + cfg.hard_yellow.judged_paths))
|
||
self.assertEqual((cfg.path_judge.judge, cfg.path_judge.scope), ("jev", "all"))
|
||
|
||
def test_werkcrappit_judges_its_builds_but_not_the_fixture_builds(self):
|
||
from werkrevisor.globs import first_match
|
||
cfg = example_config("werkcrappit.toml")
|
||
for path in ("build.gradle.kts", "werkcrappit-pit/build.gradle.kts", "settings.gradle.kts", "tools/publish",
|
||
".werkator.yml"):
|
||
with self.subTest(path=path):
|
||
self.assertIsNotNone(first_match(path, cfg.hard_yellow.judged_paths))
|
||
fixture = "werkcrappit-pit/src/functionalTest/resources/fixtures/java-basic/build.gradle.kts"
|
||
self.assertIsNone(first_match(fixture, cfg.hard_yellow.paths + cfg.hard_yellow.judged_paths))
|
||
self.assertIsNotNone(first_match(fixture, cfg.categories.tests))
|
||
self.assertEqual((cfg.path_judge.scope, cfg.engine.model_role), ("all", "advises"))
|
||
|
||
def test_host_configs_with_only_the_host_part_need_the_repository_config(self):
|
||
for name in ("werkrevisor.toml", "werkator.toml", "werkcrappit.toml", "werkdock.toml"):
|
||
with self.subTest(name=name):
|
||
with self.assertRaisesRegex(ConfigError, "sections missing"):
|
||
load_config(os.path.join(ROOT, "config", name))
|
||
|
||
def test_budget_that_cannot_pay_the_output_is_rejected(self):
|
||
with self.assertRaisesRegex(ConfigError, "does not even pay for"):
|
||
config(budget={"max_usd_per_review": 0.08})
|
||
|
||
def test_old_max_input_tokens_is_an_unknown_key(self):
|
||
with self.assertRaisesRegex(ConfigError, "unknown keys: max_input_tokens"):
|
||
config(model={"max_input_tokens": 25000})
|
||
|
||
def test_old_red_min_findings_is_an_unknown_key(self):
|
||
with self.assertRaisesRegex(ConfigError, "unknown keys: red_min_findings"):
|
||
config(review={"red_min_findings": 2})
|
||
|
||
def test_orange_min_medium_must_be_a_non_negative_integer(self):
|
||
self.assertEqual(config(review={"orange_min_medium": 0}).review.orange_min_medium, 0)
|
||
for value in (-1, "two", True):
|
||
with self.subTest(value=value):
|
||
with self.assertRaisesRegex(ConfigError, r"\[review\] orange_min_medium must be a non-negative integer"):
|
||
config(review={"orange_min_medium": value})
|
||
|
||
def test_werkdock_holds_the_levels_of_its_required_checks_for_the_human(self):
|
||
# werkrevisor checks hard_yellow.paths only outside the docs, so the risk analysis must not count as docs.
|
||
cfg = example_config("werkdock.toml")
|
||
for path in ("docs/risk-analysis.md", "tools/vulnerability-exceptions.txt", "tools/allowed-licenses.txt"):
|
||
with self.subTest(path=path):
|
||
outcome = evaluate_rules(files_from_gitea([gitea_file(path)]), {}, cfg)
|
||
self.assertIn("risky_paths", [r.code for r in outcome.reasons])
|
||
self.assertIs(outcome.verdict, Verdict.YELLOW)
|
||
|
||
def test_examples_ask_for_a_severity_per_finding(self):
|
||
for name in ("werkprobe.toml", "werkrevisor.toml", "werkator.toml", "werkcrappit.toml", "werkregie.toml",
|
||
"werkbaum.toml", "werkkompass.toml", "werkdock.toml", "hsadmin-ng.toml"):
|
||
with self.subTest(name=name):
|
||
instructions = example_config(name).review.extra_instructions
|
||
self.assertIn("Start every issue_header with the severity of the issue: [high], [medium] or [low].",
|
||
instructions)
|
||
self.assertEqual(example_config(name).review.orange_min_medium, 2)
|
||
|
||
def test_hsadmin_ng_budget(self):
|
||
self.assertEqual(load_config(os.path.join(ROOT, "config", "hsadmin-ng.toml")).budget.max_usd_per_review, 1.0)
|
||
|
||
def test_unknown_key_is_rejected(self):
|
||
with self.assertRaisesRegex(ConfigError, "unknown keys: max_finding"):
|
||
config(review={"max_finding": 3})
|
||
|
||
def test_invalid_values(self):
|
||
for overrides in ({"gate": {"yellow_gate": "maybe"}},
|
||
{"gate": {"yellow_gate": "pass_with_review_request", "human_reviewers": []}},
|
||
{"gate": {"yellow_gate": "approval", "human_reviewers": []}},
|
||
{"model": {"max_retries": 3}},
|
||
{"model": {"reasoning_effort": "huge"}},
|
||
{"categories": {"auto_approvable": ["everything"]}},
|
||
{"budget": {"max_usd_per_review": 0}}):
|
||
with self.subTest(overrides=overrides):
|
||
with self.assertRaises(ConfigError):
|
||
config(**overrides)
|
||
|
||
def test_missing_file(self):
|
||
with self.assertRaises(ConfigError):
|
||
load_config(os.path.join(tempfile.mkdtemp(), "missing.toml"))
|
||
|
||
def test_env(self):
|
||
env = {"WERKATOR_GITEA_BASE_URL": "https://git.example.org/", "WERKATOR_GITEA_OWNER": "o",
|
||
"WERKATOR_GITEA_REPO": "r", "WERKATOR_PR_NUMBER": "12", "WERKATOR_PR_HEAD_SHA": "abc"}
|
||
ref = pr_ref_from_env(env)
|
||
self.assertEqual(ref.url, "https://git.example.org/o/r/pulls/12")
|
||
with self.assertRaisesRegex(ConfigError, "WERKATOR_PR_HEAD_SHA"):
|
||
pr_ref_from_env({k: v for k, v in env.items() if k != "WERKATOR_PR_HEAD_SHA"})
|
||
with self.assertRaisesRegex(ConfigError, "https"):
|
||
pr_ref_from_env(dict(env, WERKATOR_GITEA_BASE_URL="http://git.example.org"))
|
||
|
||
def test_pr_url(self):
|
||
ref = pr_ref_from_url("https://git.example.org/gitea/o/r/pulls/5")
|
||
self.assertEqual((ref.base_url, ref.owner, ref.repo, ref.number), ("https://git.example.org/gitea", "o", "r", 5))
|
||
with self.assertRaises(ConfigError):
|
||
pr_ref_from_url("https://git.example.org/o/r/issues/5")
|
||
|
||
|
||
def pinned_config():
|
||
# The example config's prices change with the market; these tests pin their own.
|
||
return config(model={"input_price_usd_per_mtok": 0.30, "output_price_usd_per_mtok": 2.50, "max_output_tokens": 4000},
|
||
budget={"max_usd_per_review": 0.10})
|
||
|
||
|
||
def counted(tokens, worst_case_usd):
|
||
return budget.TokenCount(tokens, worst_case_usd, "o200k_base", round(tokens / 1.15), 1.15)
|
||
|
||
|
||
class BudgetTest(unittest.TestCase):
|
||
def test_estimate_counts_all_attempts(self):
|
||
cfg = pinned_config()
|
||
est = budget.estimate(3500, 0, cfg.model, cfg.budget)
|
||
expected_tokens = math.ceil((3500 * 1.5 + 16000) / 3.5)
|
||
self.assertEqual(est.input_tokens, expected_tokens)
|
||
per_attempt = (expected_tokens * 0.30 + 4000 * 2.50) / 1e6
|
||
self.assertAlmostEqual(est.worst_case_usd, 2 * per_attempt, places=6)
|
||
|
||
def test_input_limit_is_the_lower_of_budget_and_context(self):
|
||
cfg = config(budget={"max_usd_per_review": 0.40})
|
||
self.assertEqual(budget.input_token_limit(cfg.model, cfg.budget),
|
||
budget.InputLimit(80000, "the limit of 80000 derived from [budget] max_usd_per_review = 0.40"))
|
||
cfg = config(budget={"max_usd_per_review": 1.00})
|
||
self.assertEqual(budget.input_token_limit(cfg.model, cfg.budget),
|
||
budget.InputLimit(150000, "[model] max_context_tokens = 150000"))
|
||
|
||
def test_input_limit_without_input_price_is_the_context(self):
|
||
cfg = config(model={"input_price_usd_per_mtok": 0})
|
||
self.assertEqual(budget.input_token_limit(cfg.model, cfg.budget).tokens, 150000)
|
||
|
||
def test_precheck(self):
|
||
cfg = config()
|
||
self.assertIsNone(budget.precheck(counted(100000, 0.01), cfg.model, cfg.budget))
|
||
self.assertEqual(budget.precheck(counted(100001, 0.01), cfg.model, cfg.budget).code, "too_large_tokens")
|
||
reason = budget.precheck(counted(1000, 0.5), cfg.model, cfg.budget)
|
||
self.assertEqual((reason.code, reason.verdict), ("too_large_cost", Verdict.YELLOW))
|
||
|
||
def test_precheck_reasons_name_the_count_and_the_limit(self):
|
||
cfg = config(budget={"max_usd_per_review": 0.40})
|
||
reason = budget.precheck(counted(90000, 0.01), cfg.model, cfg.budget)
|
||
self.assertEqual(reason.text, "no AI review: the PR needs 90000 input tokens "
|
||
"(78261 counted with o200k_base × [model] token_count_factor = 1.15), "
|
||
"more than the limit of 80000 derived from [budget] max_usd_per_review = 0.40")
|
||
cfg = config(budget={"max_usd_per_review": 1.00})
|
||
reason = budget.precheck(budget.TokenCount(210000, 0.01, "estimated"), cfg.model, cfg.budget)
|
||
self.assertEqual(reason.text, "no AI review: the PR needs about 210000 input tokens "
|
||
"(token count estimated: the tokenizer could not load), "
|
||
"more than [model] max_context_tokens = 150000")
|
||
reason = budget.precheck(counted(1000, 1.4), cfg.model, cfg.budget)
|
||
self.assertEqual(reason.text, "no AI review: it would cost up to 1.40 USD, "
|
||
"more than [budget] max_usd_per_review = 1.00")
|
||
|
||
def test_precheck_cost_just_above_the_limit_stays_distinguishable(self):
|
||
cfg = config(budget={"max_usd_per_review": 0.20})
|
||
reason = budget.precheck(counted(1000, 0.2004), cfg.model, cfg.budget)
|
||
self.assertIn("up to 0.2004 USD, more than [budget] max_usd_per_review = 0.20", reason.text)
|
||
|
||
def test_key_check_names_the_deciding_limit(self):
|
||
reason = budget.key_check({"limit_remaining": 0.3}, config(budget={"max_usd_per_review": 0.40,
|
||
"min_key_remaining_usd": 0.50}).budget)
|
||
self.assertEqual(reason.text, "budget: the key has 0.3000 USD left, "
|
||
"less than [budget] min_key_remaining_usd = 0.50")
|
||
reason = budget.key_check({"limit_remaining": 0.1}, config(budget={"max_usd_per_review": 0.40,
|
||
"min_key_remaining_usd": 0}).budget)
|
||
self.assertIn("less than [budget] max_usd_per_review = 0.40", reason.text)
|
||
|
||
def test_key_check(self):
|
||
b = pinned_config().budget
|
||
self.assertIsNone(budget.key_check({"limit_remaining": 5.0}, b))
|
||
self.assertIsNone(budget.key_check({"limit_remaining": None}, b))
|
||
reason = budget.key_check({"limit_remaining": 0.3}, b)
|
||
self.assertEqual((reason.code, reason.verdict), ("budget", Verdict.ERROR))
|
||
self.assertIsNotNone(budget.key_check({"limit_remaining": 0.08},
|
||
config(budget={"min_key_remaining_usd": 0, "max_usd_per_review": 0.10}).budget))
|
||
|
||
def test_actual_cost(self):
|
||
m = pinned_config().model
|
||
self.assertEqual(budget.actual_cost_usd({"cost_status": "complete", "total_cost_usd": 0.02}, m),
|
||
(0.02, "reported"))
|
||
self.assertEqual(budget.actual_cost_usd({"cost_status": "unavailable", "prompt_tokens": 1_000_000,
|
||
"completion_tokens": 0}, m), (0.3, "estimated"))
|
||
self.assertEqual(budget.actual_cost_usd({}, m), (None, "unknown"))
|
||
|
||
|
||
class RedactTest(unittest.TestCase):
|
||
def test_known_secrets_and_patterns(self):
|
||
r = Redactor(["gitea-token-123456"])
|
||
text = r("token gitea-token-123456 key sk-or-v1-abcdef1234567890 Authorization: token xyz123")
|
||
self.assertNotIn("gitea-token-123456", text)
|
||
self.assertNotIn("sk-or-v1-abcdef1234567890", text)
|
||
self.assertNotIn("xyz123", text)
|
||
|
||
def test_short_values_are_ignored(self):
|
||
self.assertEqual(Redactor(["", "ab"])("ab cd"), "ab cd")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|