import io import json import os import tomllib import unittest from tests.fakes import HEAD, ROOT, 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" 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(os.path.join(ROOT, "config", "werkprobe.toml"), "rb") as f: return tomllib.load(f) def test_examples_have_default_questions_and_keep_llm_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, "llm") 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", "writing_rules_violated"]) 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") self.assertEqual(classifier.rate(q, 0.9, 0.9, 0.3), (Verdict.RED, "")) self.assertEqual(classifier.rate(q, 0.9, 0.2, 0.3), (Verdict.YELLOW, "low confidence 0.20")) self.assertEqual(classifier.rate(q, 0.1, 0.2, 0.3), (Verdict.YELLOW, "low confidence 0.20")) self.assertEqual(classifier.rate(q, 0.6, 0.9, 0.3), (Verdict.YELLOW, "")) 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.RED, "")) 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) self.assertTrue(any(r.code == "classifier_incomplete" and "max_state_chars" in r.text for r in result.reasons)) 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 more units beyond 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_reasons_are_capped(self): result = classifier.judge_units(self.units(), classifier_config().classifier, FakeJev( {"vague_then": 0.7, "writing_rules_violated": 0.7}, confidence=0.1)) self.assertEqual(len(result.reasons), classifier.MAX_LISTED_REASONS + 1) self.assertEqual(result.reasons[-1].text, "spec: 6 more classifier findings") class ClassifierFlowTest(unittest.TestCase): def test_docs_only_can_pass_without_llm(self): 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_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_red_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.RED) self.assertIn(f"spec: {DE} Scenario#900.01 more_than_one_aspect (0.93)", [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 | – | RED |", comment) self.assertIn("5 of 6 units passed all questions", comment) def test_low_confidence_is_yellow(self): outcome, _ = run_with_jev(doc_gitea(), FakeJev({"hard_to_read": 0.9}, confidence=0.1), classifier_config()) self.assertIs(outcome.result.verdict, Verdict.YELLOW) self.assertIn("low confidence 0.10", outcome.result.reasons[0].text) 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 (RED)", section) self.assertNotIn("Das Problem", section) self.assertNotIn("Mandat", section) self.assertNotIn(classifier.HINTS_START, model.calls[0]) self.assertIs(outcome.result.verdict, Verdict.RED) def test_combined_still_needs_the_llm(self): model = FakeModel(data=review_data("changes_required"), prompt_files=[DE]) outcome, _ = run_with_jev(doc_gitea(), FakeJev(), classifier_config(mode="combined"), model) self.assertIs(outcome.result.verdict, Verdict.RED) 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: doc/PR/2026-x.md (not UTF-8 text)", [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(os.path.join(ROOT, "config", "werkprobe.toml"), encoding="utf-8") as f: toml = f.read().replace('mode = "llm"', '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", os.path.join(ROOT, "config", "werkprobe.toml")] + 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+RED") self.assertIn("totals: 10 units", text) self.assertIn("calls 10, input tokens 10000, cost 0.000420 USD", text) def test_eval_without_key_is_a_configuration_error(self): code = jeveval.main(["--config", os.path.join(ROOT, "config", "werkprobe.toml")] + self.files(), environ={}, out=io.StringIO()) self.assertEqual(code, 30) if __name__ == "__main__": unittest.main()