Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
7ae5f3a5414deaf5b297031f9ac01642323420b6
Gitea-Tools/tests/test_merger_lease_finalization.py
T
sysadminandClaude Opus 4.8 7ae5f3a541 fix(mcp): mirror abandoned terminal phase into pr_work_lease (#742 review 460) 
Addresses REQUEST_CHANGES review 460 on PR #743 @ 22d0fdd.

reviewer_pr_lease._TERMINAL_PHASES gained "abandoned" for owner-session merger
finalization, but pr_work_lease._TERMINAL_REVIEWER_PHASES did not. The two
modules parse the same append-only lease markers, so the same comment read as
terminal through reviewer_pr_lease and active through pr_work_lease, leaving
the conflict-fix acquire and PR-sync inventory readers with a stale active
lease after an abandoned finalization.

Fix: add "abandoned" to pr_work_lease._TERMINAL_REVIEWER_PHASES, with a comment
recording that the two sets must stay mirrored.

Reproduced before the fix (single abandoned marker):
  reviewer_pr_lease active=False, pr_work_lease active=True
After: both False; released/blocked/done unchanged in both modules.

Tests: new TestCrossModuleTerminalPhaseAgreement in
tests/test_merger_lease_finalization.py proves the abandoned marker is inactive
in pr_work_lease, that both modules agree for released/blocked/done/abandoned,
that claimed/validating stay active in both, that the two phase sets are
mirrored, that the default 'released' finalization outcome and reason are
unchanged, and that finalization appends without rewriting or deleting the
prior marker.

Separately pinned, NOT fixed here: on a claim-then-terminal ledger,
pr_work_lease.find_active_reviewer_lease skips the terminal marker and walks
back to the older claim, so it still reports an active lease. That newest-wins
gap is pre-existing on clean baseline a8d2087 for released, blocked, and done
alike — the #577 fix landed in reviewer_pr_lease only — and is out of scope for
this bounded remediation. test_abandoned_matches_preexisting_terminal_phases_on_full_ledger
pins that "abandoned" introduces no behavior of its own, so all four phases can
be reconciled in one separate change.

Validation: focused TestCrossModuleTerminalPhaseAgreement 7 passed / 9 subtests;
tests/test_merger_lease_finalization.py 45 passed / 20 subtests; related
pr_work_lease + reviewer/merger lease + adoption + provenance + capability-map +
anti-stomp + report-validator suites 261 passed / 41 subtests; full suite 3312
passed, 6 skipped, 2 failed. Both failures (test_issue_702_review_findings_f1_f6
F1 recovery, reconciler supersession close) reproduce identically on clean
baseline master a8d2087 and are pre-existing #737 org/repo-forwarding drift.
git diff --check clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-18 09:53:34 -04:00

756 lines
29 KiB
Python
Raw Blame History

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 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_abandoned_matches_preexisting_terminal_phases_on_full_ledger(self):
"""'abandoned' must behave exactly like the other terminal phases.
On a claim-then-terminal ledger, pr_work_lease.find_active_reviewer_lease
skips the terminal marker and walks back to the older claim, so it still
reports an active lease. That newest-wins gap is pre-existing on clean
baseline a8d2087 for 'released', 'blocked', and 'done' alike (the #577
fix landed in reviewer_pr_lease only) and is out of scope for #742; this
test pins that 'abandoned' introduces no behavior of its own, so the two
modules can be reconciled for all four phases in one separate change.
"""
def pwl_active(phase):
ledger = [
_lease_comment(),
_lease_comment(phase=phase, comment_id=12361),
]
return pwl.find_active_reviewer_lease(ledger, pr_number=PR) is not None
baseline = pwl_active("released")
for phase in ("blocked", "done", "abandoned"):
with self.subTest(phase=phase):
self.assertEqual(pwl_active(phase), baseline)
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()
Reference in New Issue View Git Blame Copy Permalink