"""Reconciler role binding for post-merge moot-lease cleanup (#745). ``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase: released`` lease marker but had no capability-map entry and no role gate: entry required only ``gitea.read``, apply required only ``gitea.pr.comment``, so any profile holding the comment permission reached the mutation while the reconciler could not satisfy resolve-exact-task -> mutation. These tests pin the fixed contract: - the canonical task and its tool-name alias resolve identically (``gitea.pr.comment`` + ``reconciler``); - only a reconciler profile satisfies permission AND role; - ``apply=false`` assessment stays reachable under ``gitea.read`` for any role and mutates nothing (documented, deliberate divergence from the apply path); - ``apply=true`` requires the exact resolved task, the reconciler role, the comment permission, a validated repository binding and matching dry-run evidence; - live/non-moot/superseded/mismatched/malformed leases fail closed; - the dry-run ledger is append-only and cleanup stays idempotent. Every fixture is synthetic. No production PR, lease session or marker is used anywhere in this module (see ``TestNoProductionLeaseTouched``). """ import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent.parent)) import os # noqa: E402 import unittest # noqa: E402 from datetime import datetime, timezone # noqa: E402 from unittest.mock import patch # noqa: E402 import mcp_server # noqa: E402 import post_merge_moot_lease_gate as gate # noqa: E402 import reviewer_pr_lease as leases # noqa: E402 from mcp_server import gitea_cleanup_post_merge_moot_lease # noqa: E402 from role_session_router import RECONCILER_TASKS, TASK_REQUIRED_ROLE # noqa: E402 from task_capability_map import required_permission, required_role # noqa: E402 FAKE_AUTH = "Basic dGVzdDp0ZXN0" SLUG = "Scaled-Tech-Consulting/Gitea-Tools" PR = 487 ISSUE = 485 SESSION = "97274-676d20a825c4" HEAD_A = "a" * 40 HEAD_B = "d" * 40 LEASE_COMMENT_ID = 6603 CANONICAL_TASK = "cleanup_post_merge_moot_lease" TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease" _BASE_OPS = "gitea.read,gitea.pr.comment" RECONCILER_ENV = { "GITEA_PROFILE_NAME": "prgs-reconciler", "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.close,gitea.branch.delete", } AUTHOR_ENV = { "GITEA_PROFILE_NAME": "prgs-author", "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.create,gitea.issue.create", } REVIEWER_ENV = { "GITEA_PROFILE_NAME": "prgs-reviewer", "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.review,gitea.pr.approve", } MERGER_ENV = { "GITEA_PROFILE_NAME": "prgs-merger", "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.merge", } # Permission shape of the configured role profiles (mirrors # tests/test_task_capability_role_invariants.py CANONICAL_ROLE_PROFILES). ROLE_PROFILE_PERMISSIONS = { "author": {"gitea.read", "gitea.pr.comment", "gitea.pr.create", "gitea.issue.create", "gitea.issue.comment", "gitea.issue.close", "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit"}, "reviewer": {"gitea.read", "gitea.pr.comment", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.request_changes", "gitea.issue.comment"}, "merger": {"gitea.read", "gitea.pr.comment", "gitea.pr.merge", "gitea.issue.comment"}, "reconciler": {"gitea.read", "gitea.pr.comment", "gitea.pr.close", "gitea.issue.close", "gitea.branch.delete"}, } def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed", candidate_head=HEAD_A, comment_id=LEASE_COMMENT_ID): body = leases.format_lease_body( repo=SLUG, pr_number=pr_number, issue_number=ISSUE, reviewer_identity="sysadmin", profile="prgs-reviewer", session_id=session_id, worktree="branches/review-pr487", phase=phase, candidate_head=candidate_head, target_branch="master", target_branch_sha="b" * 40, last_activity=datetime.now(timezone.utc), ) return {"id": comment_id, "body": body, "user": {"login": "sysadmin"}} def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999): """api_request side effect keyed on method + url; records POSTs.""" calls = {"post": []} def _side(method, url, auth=None, payload=None, *a, **k): if (method or "").upper() == "POST": calls["post"].append({"url": url, "payload": payload}) return {"id": posted_id} if "/comments" in url: return list(comments) if "/pulls/" in url: pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40} if pr_merged: pr["merged"] = True pr["merged_at"] = "2026-07-08T07:46:04Z" return pr if "/issues/" in url: return {"state": "closed" if pr_merged else "open", "number": ISSUE} return {} return _side, calls def _assessment(comments, *, pr_merged=True, pr_state="closed"): return leases.assess_post_merge_moot_lease( comments, pr_number=PR, pr_merged=pr_merged, pr_state=pr_state, merge_commit_sha="c" * 40, ) # --------------------------------------------------------------------------- # # 1-5. Capability map / router contract # --------------------------------------------------------------------------- # class TestCleanupTaskContract(unittest.TestCase): def test_canonical_task_is_reconciler_owned(self): """1. The reconciler is the role that can resolve the cleanup task.""" self.assertEqual(required_permission(CANONICAL_TASK), "gitea.pr.comment") self.assertEqual(required_role(CANONICAL_TASK), "reconciler") def test_tool_alias_resolves_to_identical_contract(self): """5. Alias and canonical task must not diverge.""" self.assertEqual( (required_permission(TOOL_ALIAS), required_role(TOOL_ALIAS)), (required_permission(CANONICAL_TASK), required_role(CANONICAL_TASK)), ) def test_author_reviewer_merger_cannot_resolve_the_task(self): """2-4. No non-reconciler role satisfies permission AND role.""" for role in ("author", "reviewer", "merger"): with self.subTest(role=role): self.assertNotEqual(required_role(CANONICAL_TASK), role) # They hold the permission — which is exactly why the role gate # is required rather than optional. self.assertIn( "gitea.pr.comment", ROLE_PROFILE_PERMISSIONS[role], "test premise: non-reconciler roles do hold pr.comment", ) def test_reconciler_profile_satisfies_permission_and_role(self): self.assertIn( required_permission(CANONICAL_TASK), ROLE_PROFILE_PERMISSIONS[required_role(CANONICAL_TASK)], ) def test_router_agrees_with_capability_map(self): for task in (CANONICAL_TASK, TOOL_ALIAS): with self.subTest(task=task): self.assertIn(task, RECONCILER_TASKS) self.assertEqual(TASK_REQUIRED_ROLE[task], required_role(task)) def test_unknown_alias_still_rejected(self): for bogus in ("cleanup_post_merge_moot_leases", "cleanup_moot_lease", ""): with self.subTest(task=bogus): with self.assertRaises(KeyError): required_role(bogus) with self.assertRaises(KeyError): required_permission(bogus) # --------------------------------------------------------------------------- # # Authorization gate unit tests # --------------------------------------------------------------------------- # class TestApplyAuthorizationGate(unittest.TestCase): def setUp(self): gate._reset_for_testing() self.addCleanup(gate._reset_for_testing) self.assessment = _assessment([_lease_comment()]) def _evidence(self, **over): base = dict( pr_number=PR, repository_slug=SLUG, lease_moot=True, cleanup_allowed=True, session_id=SESSION, candidate_head=HEAD_A, lease_comment_id=LEASE_COMMENT_ID, ) base.update(over) return gate.record_dry_run(**base) def _assess(self, **over): kwargs = dict( pr_number=PR, repository_slug=SLUG, resolved_task=CANONICAL_TASK, active_role_kind="reconciler", assessment=self.assessment, evidence=self._evidence(), ) kwargs.update(over) return gate.assess_apply_authorization(**kwargs) def test_reconciler_with_matching_evidence_is_authorized(self): result = self._assess() self.assertTrue(result["allowed"], result["reasons"]) self.assertTrue(result["evidence_matched"]) def test_apply_without_exact_task_resolution_fails(self): """8. Apply without exact task resolution fails preflight.""" result = self._assess(resolved_task=None) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") def test_resolving_another_task_does_not_authorize_cleanup(self): """9. A sibling reconciler task is not a substitute.""" for other in ("delete_branch", "reconcile_already_landed_pr", "cleanup_merged_pr_branch", "comment_pr"): with self.subTest(task=other): result = self._assess(resolved_task=other) self.assertFalse(result["allowed"]) self.assertEqual( result["blocker_kind"], "unresolved_cleanup_task") def test_non_reconciler_roles_fail_closed(self): """10. Author/reviewer/merger cannot apply despite pr.comment.""" for role in ("author", "reviewer", "merger", None, ""): with self.subTest(role=role): result = self._assess(active_role_kind=role) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "wrong_role") def test_missing_repository_identity_fails_closed(self): result = self._assess(repository_slug=None) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "repository_binding") def test_missing_dry_run_evidence_fails_closed(self): result = self._assess(evidence=None) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence") def test_dry_run_that_disallowed_cleanup_fails_closed(self): result = self._assess(evidence=self._evidence(cleanup_allowed=False)) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "dry_run_not_allowed") def test_dry_run_for_another_pr_or_repo_fails_closed(self): """11. Wrong PR / repository fails closed.""" for over in ({"pr_number": PR + 1}, {"repository_slug": "Other/Repo"}): with self.subTest(**over): result = self._assess(evidence=self._evidence(**over)) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "dry_run_mismatch") def test_superseded_lease_fails_closed(self): """12. Head/session/marker drift since the dry run fails closed.""" for over in ({"session_id": "other-session"}, {"candidate_head": HEAD_B}, {"lease_comment_id": 7777}): with self.subTest(**over): result = self._assess(evidence=self._evidence(**over)) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "superseded_lease") def test_expectation_mismatch_fails_closed(self): """11. Wrong session / head / marker expectations fail closed.""" for over in ({"expected_session_id": "nope"}, {"expected_candidate_head": HEAD_B}, {"expected_lease_comment_id": 7777}): with self.subTest(**over): result = self._assess(**over) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "lease_mismatch") def test_matching_expectations_are_authorized(self): result = self._assess( expected_session_id=SESSION, expected_candidate_head=HEAD_A, expected_lease_comment_id=LEASE_COMMENT_ID, ) self.assertTrue(result["allowed"], result["reasons"]) def test_non_moot_lease_fails_closed(self): """12. A live lease on an open PR is never cleanable.""" open_pr = _assessment( [_lease_comment()], pr_merged=False, pr_state="open") result = self._assess(assessment=open_pr) self.assertFalse(result["allowed"]) self.assertIn( result["blocker_kind"], ("lease_not_moot", "malformed_lease")) def test_malformed_lease_fails_closed(self): malformed = dict(self.assessment) malformed["active_lease"] = { "session_id": "", "candidate_head": None, "comment_id": None} result = self._assess(assessment=malformed) self.assertFalse(result["allowed"]) self.assertEqual(result["blocker_kind"], "malformed_lease") def test_ledger_is_append_only(self): """14. Recording never rewrites or drops prior entries.""" first = self._evidence() second = self._evidence(candidate_head=HEAD_B, lease_comment_id=7777) history = gate.dry_run_history() self.assertEqual(len(history), 2) self.assertEqual(history[0]["candidate_head"], first["candidate_head"]) self.assertEqual(history[1]["candidate_head"], HEAD_B) # Newest-wins for lookup, but the older entry survives in history. latest = gate.latest_dry_run(pr_number=PR, repository_slug=SLUG) self.assertEqual(latest["lease_comment_id"], second["lease_comment_id"]) self.assertEqual(gate.dry_run_history()[0]["lease_comment_id"], LEASE_COMMENT_ID) def test_history_view_cannot_mutate_the_ledger(self): self._evidence() snapshot = gate.dry_run_history() snapshot[0]["pr_number"] = 999999 self.assertEqual( gate.dry_run_history()[0]["pr_number"], PR, "dry_run_history must hand out copies, not live rows") # --------------------------------------------------------------------------- # # Tool-level behavior # --------------------------------------------------------------------------- # class _ToolCase(unittest.TestCase): def setUp(self): leases.clear_session_lease() gate._reset_for_testing() self.addCleanup(gate._reset_for_testing) def _run(self, env, *, apply, comments, pr_state="closed", pr_merged=True, resolved_task=CANONICAL_TASK, slug=SLUG, **kwargs): side, calls = _api_side_effect( pr_state=pr_state, pr_merged=pr_merged, comments=comments) with patch("mcp_server.api_request", side_effect=side), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", return_value=None), \ patch("mcp_server._bound_repository_slug", return_value=slug), \ patch("mcp_server._repository_binding_block", return_value=None), \ patch.object(mcp_server, "_preflight_resolved_task", resolved_task), \ patch.dict(os.environ, env, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=apply, remote="prgs", **kwargs) return result, calls class TestDryRunOpenToEveryRole(_ToolCase): """7. Dry run performs no mutation and stays under the read capability.""" def test_dry_run_reports_moot_and_mutates_nothing_for_every_role(self): for name, env in (("reconciler", RECONCILER_ENV), ("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)): with self.subTest(role=name): gate._reset_for_testing() result, calls = self._run( env, apply=False, comments=[_lease_comment()], resolved_task=None) self.assertTrue(result["success"]) self.assertTrue(result["lease_moot"]) self.assertTrue(result["cleanup_allowed"]) self.assertFalse(result["cleanup_performed"]) self.assertEqual(result["mode"], "read_only") self.assertEqual(calls["post"], [], "dry run must not mutate") def test_dry_run_records_evidence(self): result, _ = self._run( RECONCILER_ENV, apply=False, comments=[_lease_comment()]) evidence = result["dry_run_evidence"] self.assertEqual(evidence["pr_number"], PR) self.assertEqual(evidence["repository_slug"], SLUG) self.assertEqual(evidence["session_id"], SESSION) self.assertEqual(evidence["candidate_head"], HEAD_A) self.assertEqual(evidence["lease_comment_id"], LEASE_COMMENT_ID) self.assertTrue(evidence["lease_moot"]) self.assertTrue(evidence["cleanup_allowed"]) class TestApplyRequiresReconciler(_ToolCase): def test_reconciler_apply_succeeds_after_matching_dry_run(self): """7 (apply). Allowed dry run then apply posts exactly one marker.""" comments = [_lease_comment()] dry, dry_calls = self._run( RECONCILER_ENV, apply=False, comments=comments) self.assertTrue(dry["cleanup_allowed"]) self.assertEqual(dry_calls["post"], []) result, calls = self._run( RECONCILER_ENV, apply=True, comments=comments) self.assertTrue(result["success"], result.get("reasons")) self.assertTrue(result["cleanup_performed"]) self.assertEqual(result["released_comment_id"], 9999) self.assertEqual(len(calls["post"]), 1) body = calls["post"][0]["payload"]["body"] self.assertIn("phase: released", body) self.assertIn("post-merge-moot", body) def test_apply_without_dry_run_fails_closed(self): """6. Apply must be preceded by a matching dry run.""" result, calls = self._run( RECONCILER_ENV, apply=True, comments=[_lease_comment()]) self.assertFalse(result["success"]) self.assertFalse(result["cleanup_performed"]) self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence") self.assertEqual(calls["post"], []) def test_apply_without_exact_task_resolution_fails_closed(self): """8. No resolved cleanup task -> no mutation.""" comments = [_lease_comment()] self._run(RECONCILER_ENV, apply=False, comments=comments) result, calls = self._run( RECONCILER_ENV, apply=True, comments=comments, resolved_task=None) self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") self.assertEqual(calls["post"], []) def test_resolving_a_different_task_does_not_authorize_apply(self): """9. Another resolved task is not a substitute.""" comments = [_lease_comment()] self._run(RECONCILER_ENV, apply=False, comments=comments) result, calls = self._run( RECONCILER_ENV, apply=True, comments=comments, resolved_task="delete_branch") self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task") self.assertEqual(calls["post"], []) def test_author_reviewer_merger_cannot_apply(self): """10. Permission-only roles are refused at the role gate.""" for name, env in (("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)): with self.subTest(role=name): gate._reset_for_testing() comments = [_lease_comment()] self._run(env, apply=False, comments=comments) result, calls = self._run(env, apply=True, comments=comments) self.assertFalse(result["success"]) self.assertFalse(result["cleanup_performed"]) self.assertEqual(result["blocker_kind"], "wrong_role") self.assertEqual( calls["post"], [], f"{name} must not post a terminal lease marker") class TestApplyFailsClosedOnLeaseState(_ToolCase): def test_open_pr_lease_is_never_force_cleaned(self): """12. Non-moot: an active lease on an open PR stays untouched.""" comments = [_lease_comment()] result, calls = self._run( RECONCILER_ENV, apply=True, comments=comments, pr_state="open", pr_merged=False) self.assertFalse(result["cleanup_performed"]) self.assertFalse(result["pr_merged_or_closed"]) self.assertEqual(calls["post"], [], "never force-clean an open PR lease") self.assertTrue(any( "still open" in r for r in result.get("cleanup_skipped_reason", []))) def test_superseded_lease_between_dry_run_and_apply_fails_closed(self): """12. The lease moved on after the dry run -> refuse.""" self._run(RECONCILER_ENV, apply=False, comments=[_lease_comment()]) moved = [_lease_comment(session_id="fresh-session", candidate_head=HEAD_B, comment_id=7777)] result, calls = self._run(RECONCILER_ENV, apply=True, comments=moved) self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "superseded_lease") self.assertEqual(calls["post"], []) def test_expectation_mismatch_fails_closed(self): """11. Wrong session / head / marker expectations refuse the apply.""" comments = [_lease_comment()] for kwargs in ({"expected_session_id": "wrong-session"}, {"expected_candidate_head": HEAD_B}, {"expected_lease_comment_id": 7777}): with self.subTest(**kwargs): gate._reset_for_testing() self._run(RECONCILER_ENV, apply=False, comments=comments) result, calls = self._run( RECONCILER_ENV, apply=True, comments=comments, **kwargs) self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "lease_mismatch") self.assertEqual(calls["post"], []) def test_already_terminal_cleanup_is_idempotent(self): """13. A released lease reports nothing to clean and posts nothing.""" first = _assessment([_lease_comment()]) released = {"id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}} comments = [_lease_comment(), released] self._run(RECONCILER_ENV, apply=False, comments=comments) result, calls = self._run(RECONCILER_ENV, apply=True, comments=comments) self.assertFalse(result["cleanup_performed"]) self.assertFalse(result["lease_moot"]) self.assertEqual(calls["post"], [], "no second terminal marker") self.assertTrue(any( "already released/terminal" in r for r in result.get("cleanup_skipped_reason", []))) def test_apply_is_append_only_never_deletes(self): """14. The only write is a POST; nothing is edited or deleted.""" comments = [_lease_comment()] self._run(RECONCILER_ENV, apply=False, comments=comments) side, _calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=comments) seen = [] def _recording(method, url, auth=None, payload=None, *a, **k): seen.append((method or "").upper()) return side(method, url, auth, payload, *a, **k) with patch("mcp_server.api_request", side_effect=_recording), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", return_value=None), \ patch("mcp_server._bound_repository_slug", return_value=SLUG), \ patch("mcp_server._repository_binding_block", return_value=None), \ patch.object(mcp_server, "_preflight_resolved_task", CANONICAL_TASK), \ patch.dict(os.environ, RECONCILER_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs") self.assertTrue(result["cleanup_performed"]) self.assertNotIn("DELETE", seen) self.assertNotIn("PATCH", seen) self.assertNotIn("PUT", seen) self.assertEqual(seen.count("POST"), 1) class TestRepositoryBinding(_ToolCase): """11. Foreign-repository targets fail closed before any mutation.""" def test_explicit_foreign_repository_is_rejected(self): comments = [_lease_comment()] side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=comments) with patch("mcp_server.api_request", side_effect=side), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", return_value=None), \ patch("mcp_server._canonical_repository_slug", return_value=(None, [])), \ patch("mcp_server._workspace_repository_slug", return_value=SLUG), \ patch.object(mcp_server, "_preflight_resolved_task", CANONICAL_TASK), \ patch.dict(os.environ, RECONCILER_ENV, clear=True): gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=False, remote="prgs") result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs", org="Some-Other-Org", repo="Some-Other-Repo") self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "repository_binding") self.assertEqual(calls["post"], []) def test_unresolvable_canonical_root_fails_closed(self): comments = [_lease_comment()] side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=comments) with patch("mcp_server.api_request", side_effect=side), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", return_value=None), \ patch("mcp_server._canonical_repository_slug", return_value=(None, ["canonical root unresolvable"])), \ patch.object(mcp_server, "_preflight_resolved_task", CANONICAL_TASK), \ patch.dict(os.environ, RECONCILER_ENV, clear=True): gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=False, remote="prgs") result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs") self.assertFalse(result["success"]) self.assertEqual(result["blocker_kind"], "repository_binding") self.assertEqual(calls["post"], []) class TestPreflightBinding(_ToolCase): """4. The apply path binds the shared preflight to the exact task.""" def test_apply_forwards_task_and_target_to_preflight(self): comments = [_lease_comment()] self._run(RECONCILER_ENV, apply=False, comments=comments) side, _calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=comments) seen = {} def _purity(remote=None, worktree_path=None, task=None, **kw): seen.update({"remote": remote, "worktree_path": worktree_path, "task": task, **kw}) return None with patch("mcp_server.api_request", side_effect=side), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", _purity), \ patch("mcp_server._bound_repository_slug", return_value=SLUG), \ patch("mcp_server._repository_binding_block", return_value=None), \ patch.object(mcp_server, "_preflight_resolved_task", CANONICAL_TASK), \ patch.dict(os.environ, RECONCILER_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=True, remote="prgs", worktree_path="/tmp/branches/reconciler-745", org="Scaled-Tech-Consulting", repo="Gitea-Tools") self.assertTrue(result["cleanup_performed"]) self.assertEqual(seen["task"], CANONICAL_TASK) self.assertEqual(seen["worktree_path"], "/tmp/branches/reconciler-745") self.assertEqual(seen["org"], "Scaled-Tech-Consulting") self.assertEqual(seen["repo"], "Gitea-Tools") def test_dry_run_does_not_require_preflight(self): def _boom(*a, **k): raise AssertionError("dry run must not run mutation preflight") side, calls = _api_side_effect( pr_state="closed", pr_merged=True, comments=[_lease_comment()]) with patch("mcp_server.api_request", side_effect=side), \ patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \ patch("mcp_server.verify_preflight_purity", _boom), \ patch("mcp_server._bound_repository_slug", return_value=SLUG), \ patch.dict(os.environ, AUTHOR_ENV, clear=True): result = gitea_cleanup_post_merge_moot_lease( pr_number=PR, apply=False, remote="prgs") self.assertTrue(result["success"]) self.assertEqual(calls["post"], []) class TestNoProductionLeaseTouched(unittest.TestCase): """16. No real production lease or PR is referenced by these tests.""" PRODUCTION_PR = 744 PRODUCTION_SESSION = "33673-1d54887a0415" PRODUCTION_MARKER = 12452 def test_fixtures_are_synthetic(self): self.assertNotEqual(PR, self.PRODUCTION_PR) self.assertNotEqual(SESSION, self.PRODUCTION_SESSION) self.assertNotEqual(LEASE_COMMENT_ID, self.PRODUCTION_MARKER) def test_module_source_never_names_the_production_lease(self): source = _Path(__file__).read_text() for token in (self.PRODUCTION_SESSION, str(self.PRODUCTION_MARKER)): self.assertEqual( source.count(token), 1, f"{token!r} must appear only in this guard's own constants", ) if __name__ == "__main__": unittest.main()