werkator build succeeded after 00:22
werkrevisor 0.14.0: the severity of the model findings decides between ORANGE and YELLOW. - The model marks every finding as `[high]`, `[medium]` or `[low]`; werkrevisor reads and strips the mark. - ORANGE needs one high finding or `[review] orange_min_medium` (default 2) medium ones; other findings give YELLOW. - `[review] orange_min_medium` replaces `red_min_findings`. Details: the PR document under `docs/prs/`. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Reviewed-on: #14 Co-authored-by: claude-code <claude-code@javagil.de>
222 lines
12 KiB
Python
222 lines
12 KiB
Python
import math
|
||
import os
|
||
import tempfile
|
||
import unittest
|
||
|
||
from tests.fakes import ROOT, config, example_config
|
||
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.verdict import Verdict
|
||
|
||
|
||
class ConfigTest(unittest.TestCase):
|
||
def test_examples_load(self):
|
||
for name, gate in (("werkprobe.toml", "approval"), ("hsadmin-ng.toml", "block")):
|
||
with self.subTest(name=name):
|
||
cfg = load_config(os.path.join(ROOT, "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", 80000), ("werkrevisor.toml", 80000), ("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", "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", "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_werkrevisor_host_config_needs_the_repository_config(self):
|
||
with self.assertRaisesRegex(ConfigError, "sections missing"):
|
||
load_config(os.path.join(ROOT, "config", "werkrevisor.toml"))
|
||
|
||
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_examples_ask_for_a_severity_per_finding(self):
|
||
for name in ("werkprobe.toml", "werkrevisor.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(80000, 0.01), cfg.model, cfg.budget))
|
||
self.assertEqual(budget.precheck(counted(80001, 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()
|