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()