feat: add prgs-reconciler profile and gated already-landed PR close (Closes #310)

Introduce a dedicated reconciler role with gitea_reconcile_already_landed_pr, which closes open PRs only after live fetch and ancestor proof against a fresh target branch. Document the gitea-reconciler namespace and extend task routing.
2026-07-07 09:20:50 -04:00
parent 0fe8f57dda
commit 9cf1b41b77
12 changed files with 750 additions and 2 deletions
@@ -0,0 +1,131 @@
+"""Tests for already-landed PR reconciliation gates (#310)."""
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import already_landed_reconcile
+
+
+class TestAssessAlreadyLandedReconciliation(unittest.TestCase):
+    def test_open_pr_already_landed_allows_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=True,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 278,
+                    "state": "open",
+                    "title": "Fix thing (Closes #263)",
+                    "body": "",
+                    "head": {"ref": "feat/x", "sha": "abc123"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                    "git_fetch_command": "git fetch prgs master",
+                },
+            )
+        self.assertTrue(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_ALREADY_LANDED,
+        )
+        self.assertEqual(assessment["linked_issue"], 263)
+        self.assertTrue(assessment["ancestor_proof"])
+
+    def test_not_landed_pr_denies_close(self):
+        with patch(
+            "already_landed_reconcile.is_head_ancestor_of_ref",
+            return_value=False,
+        ):
+            assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+                pr={
+                    "number": 99,
+                    "state": "open",
+                    "title": "WIP",
+                    "body": "",
+                    "head": {"ref": "feat/y", "sha": "fff111"},
+                    "base": {"ref": "master"},
+                },
+                project_root="/tmp/repo",
+                remote="prgs",
+                target_branch="master",
+                target_fetch={
+                    "success": True,
+                    "target_branch": "master",
+                    "target_ref": "prgs/master",
+                    "target_branch_sha": "deadbeef",
+                },
+            )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_NOT_LANDED,
+        )
+
+    def test_stale_target_branch_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 99,
+                "state": "open",
+                "title": "WIP",
+                "body": "",
+                "head": {"ref": "feat/y", "sha": "fff111"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": False,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": None,
+                "reasons": ["git fetch failed"],
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_STALE_TARGET,
+        )
+
+    def test_closed_pr_denies_close(self):
+        assessment = already_landed_reconcile.assess_already_landed_reconciliation(
+            pr={
+                "number": 50,
+                "state": "closed",
+                "title": "Done",
+                "body": "",
+                "head": {"ref": "feat/z", "sha": "aaa"},
+                "base": {"ref": "master"},
+            },
+            project_root="/tmp/repo",
+            remote="prgs",
+            target_branch="master",
+            target_fetch={
+                "success": True,
+                "target_branch": "master",
+                "target_ref": "prgs/master",
+                "target_branch_sha": "deadbeef",
+            },
+        )
+        self.assertFalse(assessment["close_allowed"])
+        self.assertEqual(
+            assessment["eligibility_class"],
+            already_landed_reconcile.ELIGIBILITY_PR_NOT_OPEN,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()