import sys as _sys from pathlib import Path as _Path _sys.path.insert(0, str(_Path(__file__).resolve().parent)) from mutation_profile_fixture import shared_mutation_env # noqa: E402,F401 """Acquired merger-lease provenance + owner finalization (#742). Covers the two confirmed defects behind the PR #740 failure: 1. a lease minted by ``gitea_acquire_merger_pr_lease`` was listed as a sanctioned provenance source but rejected by ``is_sanctioned_session_lease`` and reported as unsanctioned by ``describe_session_lease_proof``, so the merge gate refused the merger's own freshly acquired lease; 2. no merger-role owner-session operation existed to terminally release or abandon that lease, leaving natural expiry as the only exit. """ import os import sys import unittest from datetime import datetime, timedelta, timezone from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import gitea_mcp_server as mcp_server # noqa: E402 import merger_lease_adoption as mla # noqa: E402 import pr_work_lease as pwl # noqa: E402 import reviewer_pr_lease as leases # noqa: E402 import task_capability_map as tcm # noqa: E402 HEAD = "c" * 40 OTHER_HEAD = "d" * 40 REPO = "Scaled-Tech-Consulting/Gitea-Tools" PR = 740 def _lease_comment( pr_number: int = PR, session_id: str = "merger-session", *, phase: str = "claimed", profile: str = "prgs-merger", identity: str = "merger-user", candidate_head: str = HEAD, comment_id: int = 12354, ) -> dict: body = leases.format_lease_body( repo=REPO, pr_number=pr_number, issue_number=742, reviewer_identity=identity, profile=profile, session_id=session_id, worktree="branches/merger-pr740", phase=phase, candidate_head=candidate_head, target_branch="master", target_branch_sha="e" * 40, last_activity=datetime.now(timezone.utc), ) return {"id": comment_id, "body": body, "user": {"login": identity}} def _acquired_session(**overrides) -> dict: session = { "pr_number": PR, "issue_number": 742, "session_id": "merger-session", "reviewer_identity": "merger-user", "profile": "prgs-merger", "worktree": "branches/merger-pr740", "phase": "claimed", "candidate_head": HEAD, "target_branch": "master", "target_branch_sha": "e" * 40, "repo": REPO, "comment_id": 12354, } session.update(overrides) return session def _record_acquired(session: dict | None = None, **provenance_kwargs) -> dict: provenance = mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE_MERGER, comment_id=provenance_kwargs.pop("comment_id", 12354), **provenance_kwargs, ) return leases.record_session_lease( session if session is not None else _acquired_session(), lease_provenance=provenance, ) class TestAcquiredMergerProvenance(unittest.TestCase): """AC1/AC2/AC4/AC5/AC6 — provenance sanctioning and proof description.""" def setUp(self): leases.clear_session_lease() def tearDown(self): leases.clear_session_lease() def test_acquired_merger_lease_is_sanctioned(self): session = _record_acquired() self.assertTrue(mla.is_sanctioned_session_lease(session)) self.assertEqual(mla.assess_acquired_merger_lease_integrity(session), []) def test_proof_description_is_explicit_acquired_merger_kind(self): session = _record_acquired() proof = mla.describe_session_lease_proof(session) self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ACQUIRE_MERGER) self.assertEqual(proof["lease_proof_kind"], "sanctioned_acquire_merger") self.assertTrue(proof["lease_proof_sanctioned"]) self.assertEqual(proof["lease_proof_comment_id"], 12354) def test_manual_session_seed_without_provenance_still_rejected(self): session = leases.record_session_lease(_acquired_session()) self.assertFalse(mla.is_sanctioned_session_lease(session)) proof = mla.describe_session_lease_proof(session) self.assertEqual(proof["lease_proof_kind"], "unsanctioned_manual_seed") def test_unknown_provenance_still_rejected(self): session = leases.record_session_lease( _acquired_session(), lease_provenance={"source": "totally_made_up_tool", "comment_id": 1}, ) self.assertFalse(mla.is_sanctioned_session_lease(session)) self.assertEqual( mla.describe_session_lease_proof(session)["lease_proof_kind"], "unsanctioned", ) def test_forged_acquired_provenance_without_comment_marker_rejected(self): session = leases.record_session_lease( _acquired_session(comment_id=None), lease_provenance={"source": mla.SOURCE_ACQUIRE_MERGER}, ) self.assertFalse(mla.is_sanctioned_session_lease(session)) self.assertTrue( any( "comment marker" in reason for reason in mla.assess_acquired_merger_lease_integrity(session) ) ) def test_comment_id_mismatch_between_provenance_and_session_rejected(self): session = _record_acquired(_acquired_session(comment_id=999)) self.assertFalse(mla.is_sanctioned_session_lease(session)) def test_reviewer_profile_lease_is_not_acquired_merger_proof(self): session = _record_acquired(_acquired_session(profile="prgs-reviewer")) self.assertFalse(mla.is_sanctioned_session_lease(session)) self.assertTrue( any( "not a merger profile" in reason for reason in mla.assess_acquired_merger_lease_integrity(session) ) ) def test_incomplete_fields_fail_closed(self): for field, value in ( ("session_id", ""), ("reviewer_identity", ""), ("profile", ""), ("repo", ""), ("pr_number", None), ("candidate_head", None), ("candidate_head", "not-a-sha"), ): with self.subTest(field=field, value=value): leases.clear_session_lease() session = _record_acquired(_acquired_session(**{field: value})) self.assertFalse( mla.is_sanctioned_session_lease(session), f"{field}={value!r} must not be sanctioned", ) def test_existing_reviewer_acquire_and_heartbeat_kinds_unchanged(self): for source, kind in ( (mla.SOURCE_ACQUIRE, "sanctioned_acquire"), (mla.SOURCE_HEARTBEAT, "sanctioned_heartbeat"), ): with self.subTest(source=source): leases.clear_session_lease() session = leases.record_session_lease( { "pr_number": PR, "session_id": "reviewer-session", "candidate_head": HEAD, "comment_id": 101, }, lease_provenance=mla.build_lease_provenance( source=source, comment_id=101 ), ) self.assertTrue(mla.is_sanctioned_session_lease(session)) self.assertEqual( mla.describe_session_lease_proof(session)["lease_proof_kind"], kind, ) def test_adoption_provenance_unchanged(self): session = leases.record_session_lease( { "pr_number": PR, "session_id": "merger-session", "candidate_head": HEAD, "comment_id": 202, }, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ADOPT, comment_id=202, adopted_from_session_id="reviewer-session", ), ) self.assertTrue(mla.is_sanctioned_session_lease(session)) self.assertEqual( mla.describe_session_lease_proof(session)["lease_proof_kind"], "sanctioned_adoption", ) def test_adoption_without_source_session_still_rejected(self): session = leases.record_session_lease( {"pr_number": PR, "session_id": "merger-session", "comment_id": 202}, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ADOPT, comment_id=202 ), ) self.assertFalse(mla.is_sanctioned_session_lease(session)) class TestMergeAuthorizationGate(unittest.TestCase): """AC3 — the merge gate accepts a valid freshly acquired merger lease.""" def setUp(self): leases.clear_session_lease() def tearDown(self): leases.clear_session_lease() def test_acquired_merger_lease_passes_mutation_lease_gate(self): comments = [_lease_comment()] _record_acquired() result = leases.assess_mutation_lease_gate( pr_number=PR, comments=comments, reviewer_identity="merger-user", session_id="merger-session", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertFalse(result["block"], result["reasons"]) def test_manual_seed_still_blocked_by_mutation_lease_gate(self): comments = [_lease_comment()] leases.record_session_lease(_acquired_session()) result = leases.assess_mutation_lease_gate( pr_number=PR, comments=comments, reviewer_identity="merger-user", session_id="merger-session", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertTrue(result["block"]) self.assertTrue( any("sanctioned provenance" in r for r in result["reasons"]) ) def test_head_mismatch_still_blocked_with_acquired_lease(self): comments = [_lease_comment()] _record_acquired() result = leases.assess_mutation_lease_gate( pr_number=PR, comments=comments, reviewer_identity="merger-user", session_id="merger-session", mutation="merge", live_head_sha=OTHER_HEAD, pinned_head_sha=HEAD, ) self.assertTrue(result["block"]) class TestFinalizationAssessment(unittest.TestCase): """AC7/AC8/AC9/AC10 — owner-only, append-only, idempotent finalization.""" def setUp(self): leases.clear_session_lease() def tearDown(self): leases.clear_session_lease() def _assess(self, comments, session, **overrides): kwargs = { "pr_number": PR, "session": session, "actor_identity": "merger-user", "actor_profile": "prgs-merger", "actor_session_id": "merger-session", "repo": REPO, "worktree": "branches/merger-pr740", "candidate_head": HEAD, "live_head_sha": HEAD, } kwargs.update(overrides) return mla.assess_merger_lease_finalization(comments, **kwargs) def test_owner_merger_may_release_its_own_acquired_lease(self): session = _record_acquired() result = self._assess([_lease_comment()], session) self.assertTrue(result["finalize_allowed"], result["reasons"]) self.assertFalse(result["already_terminal"]) self.assertIn(mla.MERGER_FINALIZATION_MARKER, result["finalization_body"]) self.assertIn("phase: released", result["finalization_body"]) def test_abandon_outcome_supported(self): session = _record_acquired() result = self._assess( [_lease_comment()], session, outcome=mla.OUTCOME_ABANDONED, reason="non-retryable-merge-failure", ) self.assertTrue(result["finalize_allowed"], result["reasons"]) self.assertIn("phase: abandoned", result["finalization_body"]) self.assertIn( "finalization_reason: non-retryable-merge-failure", result["finalization_body"], ) def test_unknown_outcome_rejected(self): session = _record_acquired() result = self._assess([_lease_comment()], session, outcome="deleted") self.assertFalse(result["finalize_allowed"]) def test_foreign_session_release_rejected(self): session = _record_acquired() comments = [_lease_comment(session_id="someone-else")] result = self._assess(comments, session, actor_session_id="someone-else") self.assertFalse(result["finalize_allowed"]) self.assertTrue( any("does not match" in r or "owned by" in r for r in result["reasons"]) ) def test_active_foreign_lease_on_thread_rejected(self): session = _record_acquired() comments = [ _lease_comment(), _lease_comment(session_id="other-merger", comment_id=12400), ] result = self._assess(comments, session) self.assertFalse(result["finalize_allowed"]) self.assertTrue(any("owned by session_id" in r for r in result["reasons"])) def test_reviewer_profile_cannot_finalize_merger_lease(self): session = _record_acquired() result = self._assess( [_lease_comment()], session, actor_profile="prgs-reviewer" ) self.assertFalse(result["finalize_allowed"]) self.assertTrue( any("not a merger profile" in r for r in result["reasons"]) ) def test_reviewer_sourced_session_lease_is_not_merger_owned(self): session = leases.record_session_lease( _acquired_session(profile="prgs-reviewer"), lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE, comment_id=12354 ), ) result = self._assess([_lease_comment(profile="prgs-reviewer")], session) self.assertFalse(result["finalize_allowed"]) self.assertTrue( any("merger-owned" in r for r in result["reasons"]), result["reasons"] ) def test_no_session_lease_rejected(self): result = self._assess([_lease_comment()], None) self.assertFalse(result["finalize_allowed"]) def test_wrong_pr_rejected(self): session = _record_acquired() result = self._assess( [_lease_comment(pr_number=741, session_id="merger-session")], session, pr_number=741, ) self.assertFalse(result["finalize_allowed"]) def test_wrong_repository_rejected(self): session = _record_acquired() result = self._assess( [_lease_comment()], session, repo="Scaled-Tech-Consulting/Other" ) self.assertFalse(result["finalize_allowed"]) self.assertTrue(any("repository" in r for r in result["reasons"])) def test_wrong_head_rejected(self): session = _record_acquired() result = self._assess( [_lease_comment()], session, candidate_head=OTHER_HEAD ) self.assertFalse(result["finalize_allowed"]) def test_wrong_identity_rejected(self): session = _record_acquired() result = self._assess( [_lease_comment()], session, actor_identity="someone-else" ) self.assertFalse(result["finalize_allowed"]) self.assertTrue(any("identity" in r for r in result["reasons"])) def test_token_fingerprint_mismatch_rejected(self): session = _record_acquired(native_token_fingerprint="fp-acquire") result = self._assess( [_lease_comment()], session, runtime_token_fingerprint="fp-different" ) self.assertFalse(result["finalize_allowed"]) self.assertTrue(any("fingerprint" in r for r in result["reasons"])) def test_matching_token_fingerprint_allowed(self): session = _record_acquired(native_token_fingerprint="fp-acquire") result = self._assess( [_lease_comment()], session, runtime_token_fingerprint="fp-acquire" ) self.assertTrue(result["finalize_allowed"], result["reasons"]) def test_missing_lease_marker_on_thread_rejected(self): session = _record_acquired() result = self._assess([_lease_comment(comment_id=99999)], session) self.assertFalse(result["finalize_allowed"]) self.assertTrue(any("lease marker comment" in r for r in result["reasons"])) def test_already_terminal_lease_is_idempotent(self): session = _record_acquired() comments = [ _lease_comment(), _lease_comment(phase="released", comment_id=12360), ] result = self._assess(comments, session) self.assertTrue(result["already_terminal"]) self.assertFalse(result["finalize_allowed"]) self.assertIsNone(result["finalization_body"]) self.assertEqual(result["reasons"], []) def test_abandoned_marker_ends_the_lease(self): comments = [ _lease_comment(), _lease_comment(phase="abandoned", comment_id=12361), ] self.assertIsNone( leases.find_active_reviewer_lease(comments, pr_number=PR) ) def test_finalization_is_append_only(self): session = _record_acquired() original = _lease_comment() comments = [original] result = self._assess(comments, session) self.assertTrue(result["finalize_allowed"], result["reasons"]) # The assessment never mutates or removes prior ledger entries. self.assertEqual(comments, [original]) self.assertEqual(result["lease_comment_id"], 12354) class TestCrossModuleTerminalPhaseAgreement(unittest.TestCase): """#742 review 460 — reviewer_pr_lease and pr_work_lease must agree. Both modules parse the same append-only lease markers. A phase that is terminal in one and active in the other produces two conflicting truths for the same comment, so an abandoned finalization would still read as an active lease to the conflict-fix acquire and PR-sync inventory paths. """ TERMINAL_PHASES = ("released", "blocked", "done", "abandoned") NONTERMINAL_PHASES = ("claimed", "validating") def _both_active(self, phase: str) -> tuple[bool, bool]: comments = [_lease_comment(phase=phase)] return ( leases.find_active_reviewer_lease(comments, pr_number=PR) is not None, pwl.find_active_reviewer_lease(comments, pr_number=PR) is not None, ) def test_abandoned_marker_is_not_active_in_pr_work_lease(self): _, pwl_active = self._both_active("abandoned") self.assertFalse(pwl_active) def test_terminal_phases_agree_across_modules(self): for phase in self.TERMINAL_PHASES: with self.subTest(phase=phase): rpl_active, pwl_active = self._both_active(phase) self.assertFalse(rpl_active) self.assertFalse(pwl_active) self.assertEqual(rpl_active, pwl_active) def test_nonterminal_phases_remain_active_in_both_modules(self): for phase in self.NONTERMINAL_PHASES: with self.subTest(phase=phase): rpl_active, pwl_active = self._both_active(phase) self.assertTrue(rpl_active) self.assertTrue(pwl_active) def test_terminal_phase_sets_are_mirrored(self): self.assertEqual( set(leases._TERMINAL_PHASES), set(pwl._TERMINAL_REVIEWER_PHASES) ) def test_default_released_finalization_behavior_unchanged(self): # The default outcome is still 'released', and a released marker ends # the lease for both modules exactly as before this change. leases.clear_session_lease() try: session = _record_acquired() result = mla.assess_merger_lease_finalization( [_lease_comment()], pr_number=PR, session=session, actor_identity="merger-user", actor_profile="prgs-merger", actor_session_id="merger-session", repo=REPO, worktree="branches/merger-pr740", candidate_head=HEAD, live_head_sha=HEAD, ) self.assertTrue(result["finalize_allowed"], result["reasons"]) self.assertEqual(result["outcome"], mla.OUTCOME_RELEASED) self.assertIn("phase: released", result["finalization_body"]) self.assertEqual( result["finalization_reason"], mla.DEFAULT_MERGER_FINALIZATION_REASON, ) finally: leases.clear_session_lease() def test_finalization_appends_and_never_rewrites_prior_markers(self): claimed = _lease_comment() original_body = claimed["body"] ledger = [claimed] # A terminal marker is appended alongside the original claim; the # earlier marker is left byte-for-byte intact and is never removed. terminal = _lease_comment(phase="abandoned", comment_id=12361) ledger.append(terminal) self.assertEqual(len(ledger), 2) self.assertEqual(ledger[0]["body"], original_body) self.assertEqual(ledger[0]["id"], 12354) # Newest-wins (#577): the appended terminal marker ends the lease. self.assertIsNone(leases.find_active_reviewer_lease(ledger, pr_number=PR)) def test_full_ledger_terminal_phases_end_the_lease_in_both_modules(self): for phase in self.TERMINAL_PHASES: with self.subTest(phase=phase): ledger = [ _lease_comment(), _lease_comment(phase=phase, comment_id=12361), ] for module in (leases, pwl): self.assertIsNone( module.find_active_reviewer_lease(ledger, pr_number=PR), f"{module.__name__} still active after {phase}", ) class TestFullLedgerNewestWins(unittest.TestCase): """#742 second remediation — chain-scoped newest-wins in pr_work_lease. On a realistic append-only ledger (claim -> heartbeat -> terminal), pr_work_lease.find_active_reviewer_lease skipped the terminal marker and walked backward to the older claim of that same chain, resurrecting it. A terminal marker must end its own chain, while a foreign, forged, or malformed terminal marker must never cancel someone else's active lease. """ OTHER_SESSION = "other-session" def _claim(self, **overrides): return _lease_comment(**overrides) def _terminal(self, phase="released", comment_id=12361, **overrides): return _lease_comment(phase=phase, comment_id=comment_id, **overrides) def _pwl_active(self, ledger): return pwl.find_active_reviewer_lease(ledger, pr_number=PR) # --- the chain must end ------------------------------------------------- def test_claim_then_each_terminal_phase_leaves_no_active_lease(self): for phase in ("released", "blocked", "done", "abandoned"): with self.subTest(phase=phase): self.assertIsNone( self._pwl_active([self._claim(), self._terminal(phase=phase)]) ) def test_claim_heartbeat_terminal_leaves_no_active_lease(self): for phase in ("released", "blocked", "done", "abandoned"): with self.subTest(phase=phase): ledger = [ self._claim(), self._lease_validating(), self._terminal(phase=phase, comment_id=12362), ] self.assertIsNone(self._pwl_active(ledger)) self.assertIsNone( leases.find_active_reviewer_lease(ledger, pr_number=PR) ) def _lease_validating(self): return _lease_comment(phase="validating", comment_id=12355) # --- foreign / forged terminal markers must not cancel ------------------ def test_foreign_session_terminal_does_not_cancel_active_claim(self): ledger = [ self._claim(), self._terminal(session_id=self.OTHER_SESSION), ] active = self._pwl_active(ledger) self.assertIsNotNone(active) self.assertEqual(active["session_id"], "merger-session") def test_mismatched_chain_fields_do_not_cancel_active_claim(self): cases = { "identity": {"identity": "someone-else"}, "profile": {"profile": "prgs-reviewer"}, "candidate_head": {"candidate_head": OTHER_HEAD}, } for label, override in cases.items(): with self.subTest(field=label): ledger = [self._claim(), self._terminal(**override)] self.assertIsNotNone( self._pwl_active(ledger), f"terminal with wrong {label} must not cancel the claim", ) def test_terminal_for_another_pr_does_not_cancel_active_claim(self): ledger = [ self._claim(), self._terminal(pr_number=741, comment_id=12363), ] self.assertIsNotNone(self._pwl_active(ledger)) def test_terminal_for_another_repository_does_not_cancel_active_claim(self): foreign = self._terminal() foreign["body"] = foreign["body"].replace( REPO, "Scaled-Tech-Consulting/Other-Repo" ) self.assertIsNotNone(self._pwl_active([self._claim(), foreign])) def test_malformed_terminal_marker_does_not_hide_active_claim(self): malformed = self._terminal() # Strip the session id line: no provable chain, so it cancels nothing. malformed["body"] = "\n".join( line for line in malformed["body"].splitlines() if not line.startswith("session_id:") ) active = self._pwl_active([self._claim(), malformed]) self.assertIsNotNone(active) self.assertEqual(active["session_id"], "merger-session") # --- multi-chain ledgers ------------------------------------------------ def test_newer_active_chain_survives_older_terminated_chain(self): ledger = [ self._claim(), self._terminal(comment_id=12361), _lease_comment(session_id=self.OTHER_SESSION, comment_id=12370), ] active = self._pwl_active(ledger) self.assertIsNotNone(active) self.assertEqual(active["session_id"], self.OTHER_SESSION) def test_newest_valid_chain_selected_across_multiple_histories(self): ledger = [ _lease_comment(session_id="chain-1", comment_id=12300), _lease_comment(session_id="chain-1", phase="released", comment_id=12301), _lease_comment(session_id="chain-2", comment_id=12310), _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311), _lease_comment(session_id="chain-3", comment_id=12320), ] active = self._pwl_active(ledger) self.assertIsNotNone(active) self.assertEqual(active["session_id"], "chain-3") def test_all_chains_terminated_leaves_no_active_lease(self): ledger = [ _lease_comment(session_id="chain-1", comment_id=12300), _lease_comment(session_id="chain-1", phase="released", comment_id=12301), _lease_comment(session_id="chain-2", comment_id=12310), _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311), ] self.assertIsNone(self._pwl_active(ledger)) # --- no regression in existing behavior --------------------------------- def test_single_marker_behavior_matches_reviewer_pr_lease(self): for phase in ( "claimed", "validating", "approved", "merging", "released", "blocked", "done", "abandoned", ): with self.subTest(phase=phase): ledger = [_lease_comment(phase=phase)] self.assertEqual( pwl.find_active_reviewer_lease(ledger, pr_number=PR) is not None, leases.find_active_reviewer_lease(ledger, pr_number=PR) is not None, ) def test_expired_claim_stays_inactive_and_freshness_unchanged(self): past = datetime.now(timezone.utc) - timedelta(hours=4) expired = { "id": 12399, "user": {"login": "merger-user"}, "body": leases.format_lease_body( repo=REPO, pr_number=PR, issue_number=742, reviewer_identity="merger-user", profile="prgs-merger", session_id="expired-session", worktree="branches/merger-pr740", phase="claimed", candidate_head=HEAD, target_branch="master", target_branch_sha="e" * 40, last_activity=past, expires_at=past, ), } self.assertIsNone(self._pwl_active([expired])) self.assertIsNone( leases.find_active_reviewer_lease([expired], pr_number=PR) ) def test_active_claim_without_any_terminal_marker_remains_active(self): self.assertIsNotNone(self._pwl_active([self._claim()])) def test_production_readers_agree_after_release_tool_finalization(self): """The ledger the release tool actually writes must read as terminal.""" leases.clear_session_lease() try: session = _record_acquired() result = mla.assess_merger_lease_finalization( [_lease_comment()], pr_number=PR, session=session, actor_identity="merger-user", actor_profile="prgs-merger", actor_session_id="merger-session", repo=REPO, worktree="branches/merger-pr740", candidate_head=HEAD, live_head_sha=HEAD, outcome=mla.OUTCOME_ABANDONED, ) self.assertTrue(result["finalize_allowed"], result["reasons"]) posted = { "id": 12370, "user": {"login": "merger-user"}, "body": result["finalization_body"], } ledger = [_lease_comment(), posted] for module in (leases, pwl): with self.subTest(module=module.__name__): self.assertIsNone( module.find_active_reviewer_lease(ledger, pr_number=PR) ) # Append-only: the claim marker is still present and unmodified. self.assertEqual(len(ledger), 2) self.assertEqual(ledger[0]["id"], 12354) finally: leases.clear_session_lease() class TestReleaseMergerLeaseTool(unittest.TestCase): """AC7/AC9/AC10/AC11 — the native tool contract end to end.""" def setUp(self): mcp_server._preflight_whoami_called = True mcp_server._preflight_capability_called = True mcp_server._preflight_resolved_role = "merger" mcp_server._preflight_resolved_task = "release_merger_pr_lease" leases.clear_session_lease() def tearDown(self): leases.clear_session_lease() def _merger_profile(self, **extra): p = { "username": "merger-user", "profile_name": "prgs-merger", "role": "merger", "allowed_operations": [ "gitea.read", "gitea.pr.comment", "gitea.pr.merge", ], "forbidden_operations": [ "gitea.pr.approve", "gitea.pr.review", "gitea.pr.request_changes", ], } p.update(extra) return p def _reviewer_profile(self): return { "username": "reviewer-user", "profile_name": "prgs-reviewer", "role": "reviewer", "allowed_operations": [ "gitea.read", "gitea.pr.comment", "gitea.pr.review", "gitea.pr.approve", ], "forbidden_operations": ["gitea.pr.merge"], } def _call(self, **kwargs): params = { "pr_number": PR, "worktree": "branches/merger-pr740", "candidate_head": HEAD, "remote": "prgs", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", } params.update(kwargs) return mcp_server.gitea_release_merger_pr_lease(**params) @patch("gitea_mcp_server.api_request") @patch("gitea_mcp_server._authenticated_username", return_value="merger-user") @patch("gitea_mcp_server._auth", return_value="token pass") @patch( "gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), ) @patch("gitea_mcp_server.get_profile") @patch("gitea_mcp_server._fetch_pr_comments") @patch("gitea_mcp_server._verify_role_mutation_workspace") def test_merger_releases_own_acquired_lease( self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api ): mock_profile.return_value = self._merger_profile() mock_comments.return_value = [_lease_comment()] mock_api.side_effect = [ {"state": "open", "head": {"sha": HEAD}}, {"id": 12370}, ] _record_acquired() with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): res = self._call() self.assertTrue(res["success"], res) self.assertTrue(res["finalized"]) self.assertEqual(res["finalization_comment_id"], 12370) self.assertEqual(res["outcome"], mla.OUTCOME_RELEASED) # AC11: no active in-session lease survives finalization. self.assertIsNone(leases.get_session_lease()) @patch("gitea_mcp_server.api_request") @patch("gitea_mcp_server._authenticated_username", return_value="merger-user") @patch("gitea_mcp_server._auth", return_value="token pass") @patch( "gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), ) @patch("gitea_mcp_server.get_profile") @patch("gitea_mcp_server._fetch_pr_comments") @patch("gitea_mcp_server._verify_role_mutation_workspace") def test_merger_cannot_release_foreign_session_lease( self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api ): mock_profile.return_value = self._merger_profile() mock_comments.return_value = [_lease_comment(session_id="other-merger")] mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}] _record_acquired() with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): res = self._call() self.assertFalse(res["success"], res) self.assertFalse(res["finalized"]) self.assertTrue(res["reasons"]) # Fail closed: no comment POST was attempted. self.assertEqual(mock_api.call_count, 1) self.assertIsNotNone(leases.get_session_lease()) @patch("gitea_mcp_server.api_request") @patch("gitea_mcp_server._authenticated_username", return_value="merger-user") @patch("gitea_mcp_server._auth", return_value="token pass") @patch( "gitea_mcp_server._resolve", return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), ) @patch("gitea_mcp_server.get_profile") @patch("gitea_mcp_server._fetch_pr_comments") @patch("gitea_mcp_server._verify_role_mutation_workspace") def test_already_terminal_release_is_idempotent( self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api ): mock_profile.return_value = self._merger_profile() mock_comments.return_value = [ _lease_comment(), _lease_comment(phase="released", comment_id=12360), ] mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}] _record_acquired() with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}): res = self._call() self.assertTrue(res["success"], res) self.assertFalse(res["finalized"]) self.assertTrue(res["already_terminal"]) # No second marker posted. self.assertEqual(mock_api.call_count, 1) self.assertIsNone(leases.get_session_lease()) @patch("gitea_mcp_server.get_profile") def test_reviewer_profile_cannot_invoke_merger_release(self, mock_profile): mock_profile.return_value = self._reviewer_profile() _record_acquired() with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-reviewer"}): res = self._call() self.assertFalse(res["success"], res) self.assertFalse(res["finalized"]) self.assertTrue(res["reasons"]) # The lease is untouched by a rejected cross-role call. self.assertIsNotNone(leases.get_session_lease()) class TestReleaseMergerLeaseCapability(unittest.TestCase): """AC9 — capability map binds the task to gitea.pr.comment + merger role.""" def test_task_and_tool_alias_are_merger_scoped(self): for task in ("release_merger_pr_lease", "gitea_release_merger_pr_lease"): with self.subTest(task=task): self.assertEqual( tcm.required_permission(task), "gitea.pr.comment" ) self.assertEqual(tcm.required_role(task), "merger") def test_reviewer_release_tool_remains_reviewer_scoped(self): self.assertTrue(hasattr(mcp_server, "gitea_release_reviewer_pr_lease")) self.assertNotEqual( mcp_server.gitea_release_reviewer_pr_lease, mcp_server.gitea_release_merger_pr_lease, ) if __name__ == "__main__": unittest.main()