Spaces:
Running
Running
| import pytest | |
| from mouh.jev.audit import AuditLog | |
| from mouh.jev.types import ( | |
| JevMessageType as T, JevProofObligation, TaskPayload, VERIFICATION_THRESHOLD, | |
| ) | |
| from mouh.verification import ( | |
| CallableGenerator, CorrectionStore, NullGenerator, TestTimeVerificationLoop, parse_files, | |
| ) | |
| BUGGY = "def add(a, b):\n return a - b\n" | |
| FIXED = "def add(a, b):\n return a + b\n" | |
| TESTS = "from calc import add\n\ndef test_add():\n assert add(2, 3) == 5\n\ndef test_zero():\n assert add(0, 0) == 0\n" | |
| def make_task(code=None, tests=TESTS, proofs=()): | |
| return TaskPayload(instruction="écrire add(a, b)", candidate_files={"calc.py": code} if code else {}, | |
| test_files={"test_calc.py": tests} if tests else {}, proofs=tuple(proofs)) | |
| def types_of(outcome): | |
| return [m.type for m in outcome.messages] | |
| def test_threshold_constant(): | |
| assert VERIFICATION_THRESHOLD == 0.85 | |
| def test_certified_on_first_attempt_emits_certificate(tmp_path): | |
| audit = AuditLog(tmp_path / "audit.jsonl") | |
| loop = TestTimeVerificationLoop(audit=audit) | |
| out = loop.run(make_task(FIXED)) | |
| assert out.status == "certified" and out.attempts == 1 | |
| assert out.final.certificate.score >= 0.85 and "tests" in out.final.certificate.evidence | |
| assert T.FINAL_ANSWER in types_of(out) and out.abort is None | |
| assert audit.verify() == (True, None) | |
| # chaque saut a été journalisé | |
| kinds = [r["event"]["kind"] for r in audit.records()] | |
| assert kinds.count("accepted") == len(out.messages) | |
| def test_correction_loop_repairs_bug_and_remembers_fix(tmp_path): | |
| calls = [] | |
| def gen(req): | |
| calls.append(req) | |
| assert req.message_type is T.CORRECTION_REQUEST and req.attempt == 2 | |
| assert req.experts[0] == 5 # expert n°5 (refactoring) prioritaire | |
| assert any(d.error_type.value == "test_failure" for d in req.diagnostics) | |
| return {"calc.py": FIXED} | |
| store = CorrectionStore(tmp_path / "fixes.jsonl") | |
| loop = TestTimeVerificationLoop(CallableGenerator(gen), corrections=store) | |
| out = loop.run(make_task(BUGGY)) | |
| assert out.status == "certified" and out.attempts == 2 and len(calls) == 1 | |
| assert [a["certified"] for a in out.attempts_log] == [False, True] | |
| assert out.attempts_log[0]["score"] < 1.0 | |
| assert T.CORRECTION_REQUEST in types_of(out) | |
| assert len(store) == 1 | |
| # la correction mémorisée devient un indice et une donnée d'affinage | |
| again = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": FIXED} if r.hints else pytest.fail("pas d'indice")), | |
| corrections=CorrectionStore(tmp_path / "fixes.jsonl")) | |
| assert again.run(make_task(BUGGY)).status == "certified" | |
| n = store.export_sft(tmp_path / "sft.jsonl") | |
| assert n == 1 and "<diagnostics>" in (tmp_path / "sft.jsonl").read_text() | |
| def test_exhausted_budget_emits_no_final_text(): | |
| loop = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": BUGGY + f"# v{r.attempt}\n"}), max_attempts=3) | |
| out = loop.run(make_task(BUGGY)) | |
| assert out.status == "exhausted" and out.final is None and out.attempts == 3 | |
| assert T.FINAL_ANSWER not in types_of(out) and out.abort.reason.startswith("budget de tentatives épuisé") | |
| def test_no_progress_detected(): | |
| loop = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": BUGGY}), max_attempts=5) | |
| out = loop.run(make_task(BUGGY)) | |
| assert out.status == "aborted" and "aucun progrès" in out.abort.reason and out.attempts == 1 | |
| def test_null_generator_only_verifies_given_code(): | |
| out = TestTimeVerificationLoop(NullGenerator()).run(make_task(BUGGY)) | |
| assert out.status == "aborted" and out.final is None | |
| def test_no_oracle_means_nothing_is_certified(): | |
| out = TestTimeVerificationLoop().run(make_task(FIXED, tests="")) | |
| assert out.status == "aborted" and "aucun oracle" in out.abort.reason and out.final is None | |
| def test_syntax_error_then_repaired(): | |
| seq = iter([{"calc.py": FIXED}]) | |
| out = TestTimeVerificationLoop(CallableGenerator(lambda r: next(seq))).run(make_task("def add(a, b:\n return\n")) | |
| assert out.status == "certified" and out.attempts == 2 | |
| assert out.attempts_log[0]["components"]["static_analysis"] == 0.0 | |
| def test_policy_violation_blocks_certification(): | |
| evil = "import subprocess\ndef add(a, b):\n return a + b\n" | |
| out = TestTimeVerificationLoop(NullGenerator()).run(make_task(evil)) | |
| assert out.status == "aborted" and out.final is None | |
| assert out.attempts_log[0]["components"]["static_analysis"] == 0.0 | |
| def test_formal_proof_obligations_participate(): | |
| good = JevProofObligation(name="monotone", declarations={"x": "Int"}, assumptions=([">", "x", 0],), | |
| goal=[">", ["+", "x", 1], 1]) | |
| bad = JevProofObligation(name="faux", declarations={"x": "Int"}, assumptions=([">", "x", 0],), goal=[">", "x", 5]) | |
| ok = TestTimeVerificationLoop().run(make_task(FIXED, proofs=[good])) | |
| assert ok.status == "certified" and "proofs" in ok.final.certificate.evidence | |
| ko = TestTimeVerificationLoop().run(make_task(FIXED, proofs=[good, bad])) | |
| assert ko.final is None and ko.attempts_log[0]["components"]["proofs"] == 0.5 | |
| def test_proofs_alone_can_certify_without_tests(): | |
| p = JevProofObligation(name="p", declarations={"a": "Int"}, goal=["=", ["+", "a", 0], "a"]) | |
| out = TestTimeVerificationLoop().run(TaskPayload(instruction="x", candidate_files={"m.py": "x = 1\n"}, proofs=(p,))) | |
| assert out.status == "certified" | |
| def test_generator_crash_is_contained_and_reported(): | |
| def boom(req): | |
| raise RuntimeError("générateur en panne") | |
| out = TestTimeVerificationLoop(CallableGenerator(boom)).run(make_task(BUGGY)) | |
| assert out.final is None and out.status == "aborted" | |
| assert out.abort.diagnostics and out.abort.diagnostics[0].error_type.value == "handler_failure" | |
| def test_generator_proposes_first_attempt_when_no_candidate(): | |
| out = TestTimeVerificationLoop(CallableGenerator(lambda r: {"calc.py": FIXED})).run(make_task(None)) | |
| assert out.status == "certified" and out.attempts == 1 | |
| code_msgs = [m for m in out.messages if m.type is T.CODE_PROPOSAL] | |
| assert code_msgs[0].payload.origin == "model" | |
| def test_all_hops_are_schema_validated_and_routed_through_bus(): | |
| out = TestTimeVerificationLoop().run(make_task(FIXED)) | |
| seq = types_of(out) | |
| for needed in (T.TASK_SUBMISSION, T.ROUTE_REQUEST, T.ROUTE_DECISION, T.CODE_PROPOSAL, T.TOOL_CALL, | |
| T.TOOL_RESULT, T.VERIFICATION_RESULT, T.FINAL_ANSWER): | |
| assert needed in seq, needed | |
| assert seq.index(T.TOOL_CALL) < seq.index(T.TOOL_RESULT) < seq.index(T.VERIFICATION_RESULT) < seq.index(T.FINAL_ANSWER) | |
| def test_parse_files_formats(): | |
| txt = "### FILE: a.py\n```python\nx = 1\n```\n### END\n### FILE: b/c.py\ny = 2\n### END\n" | |
| assert parse_files(txt) == {"a.py": "x = 1\n", "b/c.py": "y = 2\n"} | |
| assert parse_files("```python\nz = 3\n```") == {"solution.py": "z = 3\n"} | |
| assert parse_files(" ") == {} | |