Merge pull request 'feat: require pagination proof in reconciliation PR inventory (Closes #308)' (#376) from feat/issue-308-reconcile-inventory-pagination into master

2026-07-07 09:26:02 -05:00
parent 4243b60ca3 f5953549aa
commit 7fbb1bf34a
4 changed files with 297 additions and 0 deletions
@@ -201,6 +201,12 @@ def test_validation_worktree_edit_verifier_exported():
    assert callable(assess_validation_worktree_edit_report)


+def test_reconcile_inventory_verifier_exported():
+    from review_proofs import assess_reconcile_inventory_report
+
+    assert callable(assess_reconcile_inventory_report)
+
+
 def test_reconcile_linked_issue_verifier_exported():
    from review_proofs import assess_reconcile_linked_issue_report

@@ -0,0 +1,102 @@
+"""Tests for reconciliation inventory pagination verifier (#308)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_reconcile_inventory import assess_reconcile_inventory_report  # noqa: E402
+
+
+def _good_report() -> str:
+    return "\n".join([
+        "## Already-landed reconciliation",
+        "Open PR inventory:",
+        "- Requested page size: 50",
+        "- Pages fetched: 2",
+        "- Returned PR count per page: page1=50, page2=6",
+        "- Inventory pagination proof: is_final_page: true, has_more: false",
+        "All already-landed PRs in the scanned open queue were found.",
+        "",
+        "## Controller Handoff",
+        "- Selected PR: #278",
+        "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED",
+    ])
+
+
+class TestReconcileInventoryPagination(unittest.TestCase):
+    def test_complete_metadata_passes(self):
+        result = assess_reconcile_inventory_report(_good_report())
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["pagination_proven"])
+
+    def test_no_inventory_claim_skips_checks(self):
+        report = "## Controller Handoff\n- Selected PR: #278"
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"])
+        self.assertFalse(result["inventory_claimed"])
+
+    def test_first_page_below_limit_with_finality_passes(self):
+        report = "\n".join([
+            "Open PRs listed for reconciliation.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=6",
+            "- Inventory pagination proof: is_final_page: true",
+        ])
+        result = assess_reconcile_inventory_report(report)
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_exactly_full_first_page_without_finality_blocks(self):
+        report = "\n".join([
+            "Complete queue scan of open PRs.",
+            "- Requested page size: 50",
+            "- Pages fetched: 1",
+            "- Returned PR count per page: page1=50",
+            "Listed 50 open PRs on the first page.",
+        ])
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={"requested_page_size": 50, "returned_page_size": 50},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("full first" in r.lower() for r in result["reasons"]))
+
+    def test_missing_pagination_metadata_blocks(self):
+        report = "All open PRs were inventoried. Inventory is complete."
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("metadata" in r.lower() for r in result["reasons"]))
+
+    def test_default_page_size_assumption_blocks(self):
+        report = (
+            "Exhaustive PR inventory: fewer than the default Gitea page size returned."
+        )
+        result = assess_reconcile_inventory_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("page-size" in r.lower() or "page size" in r.lower()
+                            for r in result["reasons"]))
+
+    def test_session_pagination_complete_passes(self):
+        report = "All already-landed PRs were found after scanning open PRs."
+        result = assess_reconcile_inventory_report(
+            report,
+            inventory_session={
+                "pagination_complete": True,
+                "requested_page_size": 50,
+                "pages_fetched": 2,
+                "returned_per_page": [50, 3],
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_reconcile_inventory_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()