diff --git a/tests/test_validation_ledger.py b/tests/test_validation_ledger.py new file mode 100644 index 0000000..0048820 --- /dev/null +++ b/tests/test_validation_ledger.py @@ -0,0 +1,155 @@ +"""Tests for validation ledger and final-report verifier (#271).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from validation_ledger import ( # noqa: E402 + assess_full_suite_claim, + assess_ledger_entry, + new_ledger_entry, + verify_final_report, +) + + +class TestLedgerEntry(unittest.TestCase): + def test_valid_entry_is_claimable(self): + entry = new_ledger_entry( + command="pytest tests/ -q", + cwd="/repo/branches/feat-issue-271", + exit_code=0, + output_summary="994 passed, 6 skipped", + ) + result = assess_ledger_entry(entry) + self.assertTrue(result["claimable"]) + self.assertEqual(result["verdict"], "strong") + + def test_missing_command_output_is_invalid(self): + entry = new_ledger_entry( + command="pytest tests/ -q", + cwd="/repo/branches/task", + exit_code=0, + output_summary="", + ) + result = assess_ledger_entry(entry) + self.assertFalse(result["claimable"]) + self.assertEqual(result["verdict"], "invalid") + + +class TestFullSuiteClaim(unittest.TestCase): + def _entry(self, **kwargs): + base = new_ledger_entry( + command="pytest tests/ -q", + cwd="/repo/branches/task", + exit_code=0, + output_summary="10 passed", + ) + base.update(kwargs) + return base + + def test_full_suite_overclaim_with_ignored_paths_fails(self): + entry = self._entry( + ignored_paths=[{"path": "tests/integration/", "justification": ""}], + ) + result = assess_full_suite_claim( + [entry], + report_text="Validation: full suite passed.", + ) + self.assertFalse(result["claimable"]) + self.assertTrue( + any("full suite" in r.lower() for r in result["reasons"]) + ) + + def test_full_suite_except_note_allows_ignored_paths(self): + entry = self._entry( + ignored_paths=[{ + "path": "tests/integration/", + "justification": "network tests require GITEA_INTEGRATION=1", + }], + ) + result = assess_full_suite_claim( + [entry], + report_text="Validation: full suite except integration tests.", + ) + self.assertTrue(result["claimable"]) + + +class TestFinalReportVerifier(unittest.TestCase): + def _base_report(self, **overrides): + report = { + "identity": "jcwalker3", + "profile": "prgs-author", + "capability_resolved": True, + "worktree_path": "/repo/branches/feat-issue-271", + "changed_files": ["validation_ledger.py"], + "validation_ledger": [ + new_ledger_entry( + command="pytest tests/test_validation_ledger.py -q", + cwd="/repo/branches/feat-issue-271", + exit_code=0, + output_summary="12 passed", + ) + ], + "validation_claim": "passed", + "review_submitted": False, + "review_visible_approved": False, + "review_pending": False, + "merge_performed": False, + "merge_blocker": "not a reviewer task", + "workspace_clean": True, + } + report.update(overrides) + return report + + def test_complete_report_is_proven(self): + result = verify_final_report(self._base_report()) + self.assertTrue(result["proven"]) + self.assertFalse(result["block"]) + + def test_validation_overclaim_fails(self): + report = self._base_report( + validation_claim="passed", + validation_ledger=[ + new_ledger_entry( + command="pytest tests/ -q", + cwd="/repo/branches/task", + exit_code=1, + output_summary="3 failed", + ) + ], + ) + result = verify_final_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(result["contradictions"]) + + def test_pending_approval_misreported_as_approved_fails(self): + report = self._base_report( + claims_approved_review=True, + review_visible_approved=False, + review_pending=True, + ) + result = verify_final_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("approved" in c.lower() for c in result["contradictions"]) + ) + + def test_missing_command_output_in_ledger_fails(self): + report = self._base_report( + validation_ledger=[ + new_ledger_entry( + command="pytest tests/ -q", + cwd="/repo/branches/task", + exit_code=0, + output_summary="", + ) + ], + ) + result = verify_final_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(result["reasons"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/validation_ledger.py b/validation_ledger.py new file mode 100644 index 0000000..a78dd6d --- /dev/null +++ b/validation_ledger.py @@ -0,0 +1,239 @@ +"""Validation command ledger and final-report verifier (#271). + +Sessions may only claim validation passed when the exact command, exit code, +and output summary are recorded. Final reports are checked against the +ledger and workflow proof fields before submission. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +LEDGER_REQUIRED_FIELDS = ( + "command", + "cwd", + "exit_code", + "output_summary", + "timestamp", +) + +OPTIONAL_LEDGER_FIELDS = ( + "skipped_tests", + "ignored_paths", +) + + +def _clean(value: Any) -> str: + return (value or "").strip() if isinstance(value, str) else str(value or "").strip() + + +def normalize_ledger_entry(entry: dict | None) -> dict: + """Return a normalized ledger entry with required keys present.""" + data = dict(entry or {}) + normalized = { + "command": _clean(data.get("command")), + "cwd": _clean(data.get("cwd")), + "exit_code": data.get("exit_code"), + "output_summary": _clean(data.get("output_summary")), + "timestamp": _clean(data.get("timestamp")), + "skipped_tests": data.get("skipped_tests") or [], + "ignored_paths": data.get("ignored_paths") or [], + } + return normalized + + +def assess_ledger_entry(entry: dict | None) -> dict: + """Fail closed when a ledger entry cannot support a validation claim.""" + normalized = normalize_ledger_entry(entry) + reasons: list[str] = [] + + for field in LEDGER_REQUIRED_FIELDS: + value = normalized.get(field) + if field == "exit_code": + if value is None or not isinstance(value, int): + reasons.append("exit_code missing or not an integer") + continue + if not _clean(value): + reasons.append(f"ledger field '{field}' missing or empty") + + exit_code = normalized.get("exit_code") + if isinstance(exit_code, int) and exit_code != 0: + reasons.append(f"exit_code {exit_code} indicates failure") + + if not _clean(normalized.get("output_summary")): + reasons.append("output_summary missing; command output was not summarized") + + claimable = not reasons and isinstance(exit_code, int) and exit_code == 0 + return { + "claimable": claimable, + "entry": normalized, + "reasons": reasons, + "verdict": "strong" if claimable else "invalid", + } + + +def assess_full_suite_claim( + ledger_entries: list[dict] | None, + *, + report_text: str = "", +) -> dict: + """'Full suite passed' requires zero ignored tests unless explicitly noted.""" + entries = [normalize_ledger_entry(e) for e in (ledger_entries or [])] + reasons: list[str] = [] + lower = (report_text or "").lower() + + if not entries: + return { + "claimable": False, + "verdict": "invalid", + "reasons": ["no validation ledger entries recorded"], + } + + entry_assessments = [assess_ledger_entry(e) for e in entries] + invalid = [a for a in entry_assessments if not a["claimable"]] + if invalid: + reasons.extend(invalid[0]["reasons"]) + + ignored_total = 0 + for entry in entries: + ignored = entry.get("ignored_paths") or [] + ignored_total += len(ignored) + for item in ignored: + path = (item or {}).get("path") if isinstance(item, dict) else item + justification = (item or {}).get("justification", "") if isinstance(item, dict) else "" + if path and not _clean(justification): + reasons.append( + f"ignored path '{path}' lacks justification for full-suite claim" + ) + + claims_full_suite = "full suite passed" in lower or "full test suite passed" in lower + if claims_full_suite and ignored_total: + except_phrases = ( + "full suite except", + "except ignored", + "ignored paths:", + "ignored test", + ) + if not any(phrase in lower for phrase in except_phrases): + reasons.append( + "report claims full suite passed but ledger records ignored " + "tests/paths without an explicit 'full suite except …' note" + ) + + claimable = not reasons and not invalid + return { + "claimable": claimable, + "verdict": "strong" if claimable else ("invalid" if invalid else "weak"), + "reasons": reasons, + "entry_assessments": entry_assessments, + "ignored_path_count": ignored_total, + } + + +def new_ledger_entry( + *, + command: str, + cwd: str, + exit_code: int, + output_summary: str, + skipped_tests: list[str] | None = None, + ignored_paths: list[dict] | None = None, + timestamp: str | None = None, +) -> dict: + """Build a ledger entry with an ISO-8601 UTC timestamp.""" + ts = timestamp or datetime.now(timezone.utc).replace(microsecond=0).isoformat() + return normalize_ledger_entry({ + "command": command, + "cwd": cwd, + "exit_code": exit_code, + "output_summary": output_summary, + "timestamp": ts, + "skipped_tests": skipped_tests or [], + "ignored_paths": ignored_paths or [], + }) + + +def verify_final_report(report: dict | None) -> dict: + """Verify a structured final report against ledger and workflow proofs.""" + data = dict(report or {}) + reasons: list[str] = [] + contradictions: list[str] = [] + + identity = _clean(data.get("identity")) + profile = _clean(data.get("profile")) + capability_resolved = data.get("capability_resolved") + worktree_path = _clean(data.get("worktree_path")) + changed_files = data.get("changed_files") or [] + ledger_entries = data.get("validation_ledger") or data.get("ledger_entries") or [] + + if not identity: + reasons.append("identity missing from final report") + if not profile: + reasons.append("profile missing from final report") + if capability_resolved is not True: + reasons.append("capability_resolved must be true with exact resolver evidence") + if not worktree_path: + reasons.append("worktree_path missing from final report") + if not isinstance(changed_files, list): + reasons.append("changed_files must be a list (use [] when none)") + + ledger_verdict = assess_full_suite_claim( + ledger_entries, + report_text=_clean(data.get("report_text")), + ) + if not ledger_verdict["claimable"]: + reasons.extend(ledger_verdict["reasons"]) + + validation_claim = _clean(data.get("validation_claim")).lower() + if validation_claim in ("passed", "pass", "full suite passed"): + if ledger_verdict["verdict"] == "invalid": + contradictions.append( + "validation_claim says passed but ledger entries are invalid" + ) + + review_submitted = data.get("review_submitted") + review_visible_approved = data.get("review_visible_approved") + review_pending = data.get("review_pending") + if review_visible_approved is True and review_pending is True: + contradictions.append( + "review_visible_approved and review_pending cannot both be true" + ) + if data.get("claims_approved_review") is True: + if review_visible_approved is not True: + contradictions.append( + "report claims APPROVED review but review_visible_approved is not true" + ) + if review_pending is True: + contradictions.append( + "report claims APPROVED review while review_pending is true" + ) + + merge_performed = data.get("merge_performed") + merge_blocker = _clean(data.get("merge_blocker")) + if merge_performed is True and merge_blocker: + contradictions.append("merge_performed true but merge_blocker also set") + if merge_performed is not True and not merge_blocker and data.get("claims_merged") is True: + contradictions.append("report claims merge but merge_performed is not true") + + workspace_clean = data.get("workspace_clean") + workspace_classification = _clean(data.get("workspace_classification")) + if workspace_clean is False and not workspace_classification: + reasons.append( + "workspace is not clean but workspace_classification is missing" + ) + + if review_submitted is False and data.get("claims_review_submitted") is True: + contradictions.append( + "claims_review_submitted true but review_submitted is false" + ) + + all_reasons = reasons + contradictions + proven = not all_reasons + return { + "proven": proven, + "block": not proven, + "reasons": all_reasons, + "contradictions": contradictions, + "ledger_verdict": ledger_verdict, + } \ No newline at end of file