471 lines
24 KiB
Python
471 lines
24 KiB
Python
import contextlib
|
||
import io
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import tomllib
|
||
import unittest
|
||
|
||
from tests.fakes import (HEAD, ROOT, TEST_CONFIG, FakeGitea, FakeJev, FakeModel, config, diff_for, gitea_file,
|
||
review_data)
|
||
from tests.test_flow import run
|
||
from tests.test_specdocs import DE, EN, FIXTURES
|
||
from werkrevisor import classifier, fullfiles, jeveval
|
||
from werkrevisor.config import ConfigError, QuestionConfig, load_config, parse_config
|
||
from werkrevisor.jev import JevError
|
||
from werkrevisor.specdocs import extract_units
|
||
from werkrevisor.verdict import Verdict
|
||
|
||
MAIN = "src/main/java/A.java"
|
||
WERKPROBE = TEST_CONFIG
|
||
REPO_QUESTION = """
|
||
[[classifier.section_questions]]
|
||
name = "{name}"
|
||
type = "noul"
|
||
instructions = "Is `section` bad?"
|
||
red_above = 0.9
|
||
yellow_above = 0.6
|
||
"""
|
||
|
||
|
||
def fixture_bytes(path):
|
||
with open(os.path.join(FIXTURES, os.path.basename(path)), "rb") as f:
|
||
return f.read()
|
||
|
||
|
||
def classifier_config(mode="classifier", **classifier_overrides):
|
||
overrides = {"engine": {"mode": mode}, "gate": {"yellow_gate": "block"}}
|
||
if classifier_overrides:
|
||
overrides["classifier"] = classifier_overrides
|
||
return config(**overrides)
|
||
|
||
|
||
def doc_gitea(*extra):
|
||
files = [gitea_file(DE)] + [gitea_file(p) for p in extra]
|
||
diff = diff_for(DE, added=["Neuer Text."]) + "".join(diff_for(p, added=["x"]) for p in extra)
|
||
gitea = FakeGitea(files=files, diff=diff)
|
||
gitea.raw_files[DE] = fixture_bytes(DE)
|
||
return gitea
|
||
|
||
|
||
def run_with_jev(gitea, jev, cfg, model=None, **kw):
|
||
return run(gitea, model, cfg=cfg, jev=jev, **kw)
|
||
|
||
|
||
def question(kind, **kw):
|
||
values = {"noul": {}, "choice": {"criteria": {"ok": "fine", "bad": "broken", "worse": "very broken"},
|
||
"problem_options": ("bad", "worse")},
|
||
"score": {"criteria": ("fine", "meh", "bad")}}[kind]
|
||
values.update(kw)
|
||
return QuestionConfig(name="q", type=kind, instructions="?", red_above=0.8, yellow_above=0.5, **values)
|
||
|
||
|
||
class ConfigTest(unittest.TestCase):
|
||
def raw(self):
|
||
with open(TEST_CONFIG, "rb") as f:
|
||
return tomllib.load(f)
|
||
|
||
def test_examples_have_default_questions_and_combined_mode(self):
|
||
for name in ("werkprobe.toml", "hsadmin-ng.toml"):
|
||
with self.subTest(name=name):
|
||
cfg = load_config(os.path.join(ROOT, "config", name))
|
||
self.assertEqual(cfg.engine.mode, "combined")
|
||
self.assertEqual([q.name for q in cfg.classifier.scenario_questions],
|
||
["hard_to_read", "more_than_one_aspect", "implementation_details",
|
||
"given_when_then_misused", "vague_then"])
|
||
self.assertEqual([q.name for q in cfg.classifier.section_questions],
|
||
["hard_to_understand", "slop"])
|
||
|
||
def test_engine_and_classifier_are_optional(self):
|
||
data = self.raw()
|
||
del data["engine"], data["classifier"]
|
||
cfg = parse_config(data)
|
||
self.assertEqual((cfg.engine.mode, cfg.classifier), ("llm", None))
|
||
|
||
def test_mode_needs_a_classifier(self):
|
||
data = self.raw()
|
||
del data["classifier"]
|
||
data["engine"]["mode"] = "combined"
|
||
with self.assertRaisesRegex(ConfigError, "needs a \\[classifier\\]"):
|
||
parse_config(data)
|
||
|
||
def test_every_classifier_key_is_required(self):
|
||
for key in ("endpoint", "min_confidence", "pr_doc_globs", "max_units", "section_questions"):
|
||
with self.subTest(key=key):
|
||
data = self.raw()
|
||
del data["classifier"][key]
|
||
with self.assertRaisesRegex(ConfigError, f"{key} missing"):
|
||
parse_config(data)
|
||
|
||
def test_invalid_values(self):
|
||
cases = (
|
||
("engine", None, {"mode": "both"}),
|
||
("classifier", None, {"endpoint": "http://api.typesafe.ai/v1/systemone"}),
|
||
("classifier", None, {"min_confidence": 1.5}),
|
||
("classifier", None, {"max_state_chars": 0}),
|
||
("classifier", None, {"pr_doc_globs": []}),
|
||
("classifier", None, {"surprise": 1}),
|
||
("classifier", 0, {"red_above": 0.4, "yellow_above": 0.6}),
|
||
("classifier", 0, {"type": "rank"}),
|
||
("classifier", 0, {"criteria": ["only one"]}),
|
||
("classifier", 1, {"criteria": {"yes": "a", "no": "b"}}),
|
||
("classifier", 1, {"type": "choice", "criteria": {"a": "x", "b": "y"}, "problem_options": ["a", "b"]}),
|
||
("classifier", 1, {"type": "choice", "criteria": {"a": "x", "b": "y"}}),
|
||
("classifier", 1, {"problem_options": ["a"]}),
|
||
("classifier", 1, {"name": "Bad Name"}),
|
||
("classifier", 1, {"name": "hard_to_read"}),
|
||
("classifier", 1, {"typo": 1}),
|
||
)
|
||
for section, index, values in cases:
|
||
with self.subTest(values=values):
|
||
data = self.raw()
|
||
target = data[section] if index is None else data[section]["scenario_questions"][index]
|
||
target.update(values)
|
||
with self.assertRaises(ConfigError):
|
||
parse_config(data)
|
||
|
||
def test_payload(self):
|
||
self.assertEqual(question("score").payload(),
|
||
{"type": "score", "instructions": "?", "criteria": ["fine", "meh", "bad"]})
|
||
self.assertEqual(question("noul").payload(), {"type": "noul", "instructions": "?"})
|
||
self.assertEqual(question("choice").payload()["criteria"]["bad"], "broken")
|
||
|
||
|
||
class MappingTest(unittest.TestCase):
|
||
def test_problem_values(self):
|
||
pc = classifier.problem_and_confidence
|
||
self.assertEqual(pc(question("noul"), {"type": "noul", "noul": 0.7}), (0.7, None))
|
||
self.assertEqual(pc(question("score"), {"type": "score", "score": 1.5, "confidence": 0.6}), (0.75, 0.6))
|
||
choice = {"type": "choice", "choice": "ok", "confidence": 0.5,
|
||
"probabilities": {"ok": 0.4, "bad": 0.35, "worse": 0.25}}
|
||
self.assertAlmostEqual(pc(question("choice"), choice)[0], 0.6)
|
||
|
||
def test_invalid_answers(self):
|
||
for q, answer in ((question("noul"), {"type": "score", "score": 1}),
|
||
(question("noul"), {"type": "noul", "noul": 1.2}),
|
||
(question("noul"), {"type": "noul", "noul": True}),
|
||
(question("score"), {"type": "score", "score": 3, "confidence": 0.9}),
|
||
(question("score"), {"type": "score", "score": 1}),
|
||
(question("choice"), {"type": "choice", "confidence": 0.9})):
|
||
with self.subTest(answer=answer):
|
||
with self.assertRaises(JevError):
|
||
classifier.problem_and_confidence(q, answer)
|
||
|
||
def test_rate(self):
|
||
q = question("score")
|
||
low = "low confidence 0.20 < [classifier] min_confidence = 0.3"
|
||
self.assertEqual(classifier.rate(q, 0.9, 0.9, 0.3), (Verdict.ORANGE, "0.90 > red_above = 0.8"))
|
||
self.assertEqual(classifier.rate(q, 0.9, 0.2, 0.3), (None, low))
|
||
self.assertEqual(classifier.rate(q, 0.1, 0.2, 0.3), (None, low))
|
||
self.assertEqual(classifier.rate(q, 0.6, 0.9, 0.3), (Verdict.YELLOW, "0.60 > yellow_above = 0.5"))
|
||
self.assertEqual(classifier.rate(q, 0.5, 0.9, 0.3), (Verdict.GREEN, ""))
|
||
self.assertEqual(classifier.rate(question("noul"), 0.81, None, 0.99),
|
||
(Verdict.ORANGE, "0.81 > red_above = 0.8"))
|
||
|
||
|
||
class JudgeUnitsTest(unittest.TestCase):
|
||
def units(self):
|
||
with open(os.path.join(FIXTURES, os.path.basename(DE)), encoding="utf-8") as f:
|
||
return extract_units(f.read(), DE)
|
||
|
||
def test_one_call_per_unit_with_all_its_questions(self):
|
||
jev = FakeJev()
|
||
cfg = classifier_config().classifier
|
||
result = classifier.judge_units(self.units(), cfg, jev)
|
||
self.assertEqual(len(jev.requests), 6)
|
||
self.assertEqual((result.calls, result.input_tokens, result.cost_usd), (6, 6000, 0.000252))
|
||
self.assertEqual(result.judged_units, result.passed_units)
|
||
scenario_requests = [q for state, q in jev.requests if "scenario" in state]
|
||
self.assertEqual(list(scenario_requests[0]), [q.name for q in cfg.scenario_questions])
|
||
self.assertEqual(result.reasons, [])
|
||
|
||
def test_oversized_unit_is_incomplete_and_not_sent(self):
|
||
jev = FakeJev()
|
||
result = classifier.judge_units(self.units(), classifier_config(max_state_chars=900).classifier, jev)
|
||
incomplete = [r.text for r in result.reasons if r.code == "classifier_incomplete"]
|
||
self.assertEqual(len(incomplete), 1)
|
||
self.assertIn("not judged by the classifier (each larger than [classifier] max_state_chars = 900): ",
|
||
incomplete[0])
|
||
self.assertTrue(all(len(json.dumps(state, ensure_ascii=False)) <= 900 for state, _ in jev.requests))
|
||
|
||
def test_max_units(self):
|
||
jev = FakeJev()
|
||
result = classifier.judge_units(self.units(), classifier_config(max_units=2).classifier, jev)
|
||
self.assertEqual(len(jev.requests), 2)
|
||
self.assertIn("4 units not judged by the classifier (beyond [classifier] max_units = 2)",
|
||
result.reasons[0].text)
|
||
self.assertIs(result.reasons[0].verdict, Verdict.YELLOW)
|
||
|
||
def test_error_is_error(self):
|
||
result = classifier.judge_units(self.units(), classifier_config().classifier,
|
||
FakeJev(error=JevError("POST /v1/systemone failed with HTTP 529")))
|
||
self.assertEqual([r.verdict for r in result.reasons], [Verdict.ERROR])
|
||
self.assertIn("HTTP 529 and 5 more", result.reasons[0].text)
|
||
|
||
def test_low_confidence_answers_are_ignored(self):
|
||
result = classifier.judge_units(self.units(), classifier_config().classifier, FakeJev(
|
||
{"hard_to_read": 0.9, "slop": 0.9}, confidences={"hard_to_read": 0.1, "slop": 0.29}))
|
||
self.assertEqual(result.reasons, [])
|
||
self.assertEqual(result.flagged, [])
|
||
self.assertEqual(result.passed_units, result.judged_units)
|
||
self.assertEqual(sorted({r.question for r in result.ignored}), ["hard_to_read", "slop"])
|
||
self.assertEqual(len(result.ignored), 6)
|
||
self.assertEqual({r.note for r in result.ignored},
|
||
{"low confidence 0.10 < [classifier] min_confidence = 0.3",
|
||
"low confidence 0.29 < [classifier] min_confidence = 0.3"})
|
||
self.assertTrue(all(r.verdict is Verdict.GREEN for r in result.ignored))
|
||
|
||
def test_reasons_are_capped(self):
|
||
result = classifier.judge_units(self.units(), classifier_config().classifier, FakeJev(
|
||
{"vague_then": 0.7, "hard_to_read": 0.6, "hard_to_understand": 0.6, "slop": 0.6}))
|
||
self.assertEqual(len(result.reasons), classifier.MAX_LISTED_REASONS + 1)
|
||
self.assertEqual(result.reasons[-1].text, "spec: 2 more classifier findings")
|
||
|
||
|
||
class ClassifierFlowTest(unittest.TestCase):
|
||
def test_docs_only_can_pass_without_llm(self):
|
||
"""Scenario#64.02"""
|
||
gitea = doc_gitea()
|
||
model = FakeModel()
|
||
outcome, d = run_with_jev(gitea, FakeJev(), classifier_config(), model)
|
||
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
||
self.assertEqual(model.calls, [])
|
||
self.assertEqual(outcome.result.llm_status, "skipped (engine classifier)")
|
||
self.assertEqual(gitea.raw_requests, [(DE, HEAD)])
|
||
record = json.loads(d.out.getvalue())
|
||
self.assertEqual((record["engine"], record["classifier_calls"], record["classifier_input_tokens"]),
|
||
("classifier", 6, 6000))
|
||
self.assertAlmostEqual(record["classifier_cost_usd"], 0.000252)
|
||
self.assertIn("6 of 6 units passed all questions (1 PR document(s))", gitea.created_comments[0])
|
||
|
||
def test_no_section_without_units(self):
|
||
"""Scenario#64.01"""
|
||
gitea = FakeGitea(files=[gitea_file(MAIN)], diff=diff_for(MAIN, added=["x"]))
|
||
outcome, _ = run_with_jev(gitea, FakeJev(), classifier_config("combined"), FakeModel(prompt_files=[MAIN]))
|
||
self.assertIsNot(outcome.result.verdict, Verdict.ERROR)
|
||
self.assertNotIn("### Classifier", gitea.created_comments[0])
|
||
self.assertNotIn("units passed all questions", gitea.created_comments[0])
|
||
|
||
def test_code_is_never_green_without_llm(self):
|
||
outcome, _ = run_with_jev(doc_gitea(MAIN), FakeJev(), classifier_config())
|
||
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
||
self.assertIn("no_model_review", [r.code for r in outcome.result.reasons])
|
||
|
||
def test_other_docs_are_never_green_without_llm(self):
|
||
outcome, _ = run_with_jev(doc_gitea("README.md"), FakeJev(), classifier_config())
|
||
self.assertIs(outcome.result.verdict, Verdict.YELLOW)
|
||
|
||
def test_spec_problem_is_orange_without_llm_finding(self):
|
||
jev = FakeJev({("900.01", "more_than_one_aspect"): 0.93})
|
||
gitea = doc_gitea()
|
||
outcome, _ = run_with_jev(gitea, jev, classifier_config())
|
||
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
||
self.assertIn(f"spec: {DE} Scenario#900.01 more_than_one_aspect (0.93 > red_above = 0.85)",
|
||
[r.text for r in outcome.result.reasons])
|
||
comment = gitea.created_comments[0]
|
||
self.assertIn(f"| {DE} Scenario#900.01 | more_than_one_aspect | 0.93 | – | ORANGE |", comment)
|
||
self.assertIn("5 of 6 units passed all questions", comment)
|
||
|
||
def test_spec_problem_in_approval_mode_waits_for_a_developer(self):
|
||
jev = FakeJev({("Das Problem", "slop"): 0.9})
|
||
gitea = doc_gitea()
|
||
cfg = config(engine={"mode": "classifier"}, gate={"yellow_gate": "approval", "human_reviewers": ["alice"]})
|
||
outcome, _ = run_with_jev(gitea, jev, cfg)
|
||
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
||
self.assertEqual(outcome.result.exit_code, 0)
|
||
self.assertEqual(gitea.created_reviews, [])
|
||
self.assertEqual(gitea.requested, ["alice"])
|
||
self.assertEqual(gitea.statuses[-1][1], "warning")
|
||
|
||
def test_low_confidence_gives_no_reason_and_no_yellow(self):
|
||
gitea = doc_gitea()
|
||
jev = FakeJev({"hard_to_read": 0.9}, confidences={"hard_to_read": 0.1})
|
||
outcome, d = run_with_jev(gitea, jev, classifier_config())
|
||
kinds = ["scenario" if "scenario" in state else "section" for state, _ in jev.requests]
|
||
self.assertEqual((kinds.count("scenario"), kinds.count("section")), (2, 4))
|
||
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
||
self.assertNotIn("hard_to_read", " ".join(r.text for r in outcome.result.reasons))
|
||
self.assertIn("6 of 6 units passed all questions", gitea.created_comments[0])
|
||
self.assertIn("2 answers ignored because of a confidence below [classifier] min_confidence.",
|
||
gitea.created_comments[0])
|
||
record = json.loads(d.out.getvalue())
|
||
self.assertEqual((record["classifier_flagged"], record["classifier_ignored"]), (0, 2))
|
||
|
||
def test_jev_error_is_error_and_redacted(self):
|
||
from werkrevisor.redact import Redactor
|
||
jev = FakeJev(error=JevError("failed with typesafe-secret-key-0123456789"))
|
||
gitea = doc_gitea()
|
||
outcome, _ = run_with_jev(gitea, jev, classifier_config(mode="combined"),
|
||
redactor=Redactor(["typesafe-secret-key-0123456789"]))
|
||
self.assertIs(outcome.result.verdict, Verdict.ERROR)
|
||
self.assertEqual(outcome.result.llm_status, "skipped (classifier failed)")
|
||
self.assertNotIn("typesafe-secret-key", gitea.created_comments[0])
|
||
|
||
def test_combined_passes_hints_without_pr_text_to_the_llm(self):
|
||
jev = FakeJev({("900.01", "vague_then"): 0.7, ("Das Problem", "slop"): 0.9})
|
||
model = FakeModel(prompt_files=[DE])
|
||
outcome, _ = run_with_jev(doc_gitea(), jev, classifier_config(mode="combined"), model)
|
||
self.assertEqual(len(model.calls), 1)
|
||
section = model.sections[0].split(fullfiles.SECTION_END)[-1]
|
||
self.assertIn(classifier.HINTS_START, section)
|
||
self.assertIn(f"- `{DE}` line 20, Scenario#900.01: vague then, 0.70 (YELLOW)", section)
|
||
self.assertIn(f"- `{DE}` line 7, section: slop, 0.90 (ORANGE)", section)
|
||
self.assertNotIn("Das Problem", section)
|
||
self.assertNotIn("Mandat", section)
|
||
self.assertNotIn(classifier.HINTS_START, model.calls[0])
|
||
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
||
|
||
def test_combined_passes_low_confidence_answers_as_uncertain_hints(self):
|
||
jev = FakeJev({("900.01", "hard_to_read"): 0.7}, confidences={("900.01", "hard_to_read"): 0.1})
|
||
model = FakeModel(prompt_files=[DE])
|
||
outcome, _ = run_with_jev(doc_gitea(), jev, classifier_config(mode="combined"), model)
|
||
section = model.sections[0].split(fullfiles.SECTION_END)[-1]
|
||
self.assertIn(f"- `{DE}` line 20, Scenario#900.01: hard to read, 0.70 (uncertain, confidence 0.10)", section)
|
||
self.assertIn("An uncertain answer had a low confidence and does not count for the verdict", section)
|
||
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
||
|
||
def test_combined_still_needs_the_llm(self):
|
||
model = FakeModel(data=review_data(issues=[DE], header="[high] Possible Bug"), prompt_files=[DE])
|
||
jev = FakeJev(problems={"real_defect": 0.9, "severity": 0.9})
|
||
outcome, _ = run_with_jev(doc_gitea(), jev, classifier_config(mode="combined"), model)
|
||
self.assertIs(outcome.result.verdict, Verdict.ORANGE)
|
||
self.assertNotIn(classifier.HINTS_START, model.sections[0])
|
||
|
||
def test_llm_mode_does_not_call_jev(self):
|
||
jev = FakeJev()
|
||
gitea = doc_gitea()
|
||
outcome, _ = run_with_jev(gitea, jev, classifier_config(mode="llm"))
|
||
self.assertEqual(jev.requests, [])
|
||
self.assertIs(outcome.result.verdict, Verdict.GREEN)
|
||
self.assertNotIn("### Classifier", gitea.created_comments[0])
|
||
self.assertNotIn("classifier `", gitea.created_comments[0])
|
||
|
||
def test_secrets_keep_the_pr_away_from_jev(self):
|
||
gitea = doc_gitea()
|
||
gitea.diff = diff_for(DE, added=["-----BEGIN PRIVATE KEY-----"])
|
||
jev = FakeJev()
|
||
outcome, _ = run_with_jev(gitea, jev, classifier_config())
|
||
self.assertEqual(jev.requests, [])
|
||
self.assertEqual(outcome.result.classifier_status, "skipped (possible secrets)")
|
||
|
||
def test_no_llm_skips_jev(self):
|
||
jev = FakeJev()
|
||
run_with_jev(doc_gitea(), jev, classifier_config(), use_llm=False)
|
||
self.assertEqual(jev.requests, [])
|
||
|
||
def test_templates_and_non_utf8_docs(self):
|
||
gitea = doc_gitea("doc/PR/TEMPLATE.md", "doc/PR/2026-x.md")
|
||
gitea.raw_files["doc/PR/2026-x.md"] = b"\xff\xfe"
|
||
jev = FakeJev()
|
||
outcome, _ = run_with_jev(gitea, jev, classifier_config())
|
||
self.assertEqual({state["pr_document"] for state, _ in jev.requests}, {DE})
|
||
self.assertIn("classifier context incomplete: 1 PR document not judged by the classifier "
|
||
"(not UTF-8 text): doc/PR/2026-x.md",
|
||
[r.text for r in outcome.result.reasons])
|
||
|
||
|
||
class MainKeyTest(unittest.TestCase):
|
||
def test_classifier_mode_needs_the_typesafe_key_but_no_openrouter_key(self):
|
||
import contextlib
|
||
from werkrevisor.cli import main
|
||
with open(TEST_CONFIG, encoding="utf-8") as f:
|
||
toml = f.read().replace('mode = "combined"', 'mode = "classifier"')
|
||
environ = {"WERKREVISOR_CONFIG_TOML": toml, "GITEA_BOT_TOKEN": "gitea-token-123",
|
||
"WERKATOR_GITEA_BASE_URL": "https://git.example.org", "WERKATOR_GITEA_OWNER": "o",
|
||
"WERKATOR_GITEA_REPO": "r", "WERKATOR_PR_NUMBER": "1", "WERKATOR_PR_HEAD_SHA": HEAD}
|
||
err = io.StringIO()
|
||
with contextlib.redirect_stderr(err):
|
||
self.assertEqual(main([], environ), 30)
|
||
self.assertIn("TYPESAFE_API_KEY is not set", err.getvalue())
|
||
self.assertNotIn("OPENROUTER_API_KEY", err.getvalue())
|
||
|
||
|
||
class ReportTest(unittest.TestCase):
|
||
def test_table_cells_are_escaped(self):
|
||
units = extract_units("## A | B\n\nText.\n", "doc/PR/x.md")
|
||
result = classifier.judge_units(units, classifier_config().classifier, FakeJev({"slop": 0.9}))
|
||
table = "\n".join(classifier.render_table(result))
|
||
self.assertIn("section 'A \\| B'", table)
|
||
|
||
|
||
class JevEvalTest(unittest.TestCase):
|
||
def files(self):
|
||
return [os.path.join(FIXTURES, os.path.basename(p)) for p in (DE, EN)]
|
||
|
||
def test_units_only_needs_no_key_and_no_config(self):
|
||
out = io.StringIO()
|
||
self.assertEqual(jeveval.main(["--units-only"] + self.files(), environ={}, out=out), 0)
|
||
text = out.getvalue()
|
||
self.assertIn("Scenario#900.01", text)
|
||
self.assertIn(" Given ein Mandat, dessen Gültigkeitsende in 10 Tagen liegt", text)
|
||
self.assertIn("10 units: 4 scenarios, 6 sections", text)
|
||
|
||
def test_eval_prints_rows_and_totals(self):
|
||
out = io.StringIO()
|
||
jev = FakeJev({("901.02", "vague_then"): 0.95})
|
||
code = jeveval.main(["--config", WERKPROBE, "--no-repo-config"] + self.files(),
|
||
environ={}, out=out, jev=jev)
|
||
self.assertEqual(code, 0)
|
||
text = out.getvalue()
|
||
self.assertRegex(text, r"PR#901-build-status-badge\.md\s+Scenario#901\.02\s+vague_then\s+0\.95\s+-\s+ORANGE")
|
||
self.assertIn("totals: 10 units", text)
|
||
self.assertIn("calls 10, input tokens 10000, cost 0.000420 USD", text)
|
||
|
||
def test_eval_shows_low_confidence_answers_as_ignored(self):
|
||
out = io.StringIO()
|
||
jev = FakeJev({("901.02", "hard_to_read"): 0.9}, confidences={("901.02", "hard_to_read"): 0.1})
|
||
code = jeveval.main(["--config", WERKPROBE, "--no-repo-config"] + self.files(),
|
||
environ={}, out=out, jev=jev)
|
||
self.assertEqual(code, 0)
|
||
text = out.getvalue()
|
||
self.assertRegex(text, r"Scenario#901\.02\s+hard_to_read\s+0\.90\s+0\.10\s+"
|
||
r"IGNORED \(low confidence 0\.10 < \[classifier\] min_confidence = 0\.3\)\n")
|
||
self.assertRegex(text, r"answers \(GREEN \d+, YELLOW 0, ORANGE 0, IGNORED 1\)")
|
||
|
||
def test_eval_without_key_is_a_configuration_error(self):
|
||
code = jeveval.main(["--config", WERKPROBE, "--no-repo-config"] + self.files(),
|
||
environ={}, out=io.StringIO())
|
||
self.assertEqual(code, 30)
|
||
|
||
def section_questions(self, jev):
|
||
return {name for state, questions in jev.requests if "section" in state for name in questions}
|
||
|
||
def eval_with(self, *args):
|
||
jev = FakeJev()
|
||
err = io.StringIO()
|
||
with contextlib.redirect_stderr(err):
|
||
code = jeveval.main(["--config", WERKPROBE, *args] + self.files(), environ={}, out=io.StringIO(), jev=jev)
|
||
return code, jev, err.getvalue()
|
||
|
||
def test_repo_config_is_layered_on_top_of_the_host_config(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = os.path.join(tmp, "repo.toml")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write(REPO_QUESTION.format(name="only_this"))
|
||
code, jev, _ = self.eval_with("--repo-config", path)
|
||
self.assertEqual(code, 0)
|
||
self.assertEqual(self.section_questions(jev), {"only_this"})
|
||
|
||
def test_werkrevisor_toml_in_the_current_directory_is_the_default(self):
|
||
with tempfile.TemporaryDirectory() as tmp, contextlib.chdir(tmp):
|
||
with open(".werkrevisor.toml", "w", encoding="utf-8") as f:
|
||
f.write(REPO_QUESTION.format(name="from_cwd"))
|
||
_, by_default, _ = self.eval_with()
|
||
_, host_alone, _ = self.eval_with("--no-repo-config")
|
||
self.assertEqual(self.section_questions(by_default), {"from_cwd"})
|
||
self.assertEqual(self.section_questions(host_alone), {"hard_to_understand", "slop"})
|
||
|
||
def test_repo_config_follows_the_rules_of_the_review(self):
|
||
with tempfile.TemporaryDirectory() as tmp:
|
||
path = os.path.join(tmp, "repo.toml")
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
f.write('[classifier]\nendpoint = "https://classifier.example.org"\n')
|
||
code, jev, err = self.eval_with("--repo-config", path)
|
||
self.assertEqual(code, 30)
|
||
self.assertIn(f"{path} must not set [classifier] endpoint: only the host config sets it", err)
|
||
self.assertEqual(jev.requests, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|