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, 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 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 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()