Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 75 Pull Requests 2 Actions Packages Projects Releases Wiki Activity
Files
fde95b9266e8d610c8400e0290b04e0e0201a822
Gitea-Tools/tests/test_merger_lease_finalization.py
T
sysadminandClaude Opus 4.8 fde95b9266 fix(mcp): chain-scoped newest-wins reviewer lease reads in pr_work_lease (#742) 
Second author remediation on PR #743. Fixes the full-ledger defect surfaced by
the first remediation.

Root cause: pr_work_lease.find_active_reviewer_lease iterated the reviewer
markers newest-first and used `continue` on a terminal marker. On a realistic
append-only ledger (claimed -> validating -> terminal) it therefore stepped
over the terminal marker and returned the older claim of the very chain that
marker had just ended. reviewer_pr_lease got strict newest-wins under #577;
pr_work_lease never did, so the two modules disagreed for released, blocked,
done, and abandoned alike, and merger owner finalization did not leave "no
active lease" for pr_work_lease consumers (conflict-fix acquire, PR sync
inventory).

Fix: a claim is skipped when a later marker terminates its own chain. Chain
identity is (repo, pr_number, candidate_head, session_id, reviewer_identity,
profile) via the new _reviewer_chain_key; _chain_terminated_after scans only
markers appended after the candidate. Consequences:

- a valid terminal marker ends its matching earlier claim (no resurrection);
- a foreign-session, wrong-repo, wrong-PR, wrong-head, wrong-identity, or
  wrong-profile terminal marker cannot cancel another session's active lease,
  which strict newest-wins alone would have allowed;
- a malformed terminal marker has no provable chain key, so it cancels nothing
  and cannot hide a valid active claim;
- expiry, freshness, phase sets, ownership and integrity checks, both parsers,
  and find_active_conflict_fix_lease are untouched;
- history stays append-only; no marker is deleted or rewritten.

Reviewer lease markers carry no token field, so token-fingerprint validation
remains where it already lives (session provenance, merger finalization) and is
not weakened here.

Tests: new TestFullLedgerNewestWins in tests/test_merger_lease_finalization.py
covers claimed->released/blocked/done/abandoned, claimed->validating->terminal,
foreign-session and mismatched repo/PR/head/identity/profile terminals, the
malformed terminal marker, a newer active chain surviving an older terminated
chain, newest-valid-chain selection across multiple histories, all-chains-
terminated, single-marker parity between the two modules across eight phases,
expired-claim/freshness non-regression, and the real ledger written by
gitea_release_merger_pr_lease reading as terminal in both modules with the
prior marker preserved. TestCrossModuleTerminalPhaseAgreement's scope-limited
placeholder test is replaced by a real both-modules full-ledger assertion.

Verified 10 of the new tests fail at the prior head 7ae5f3a and pass here.

Validation: focused TestFullLedgerNewestWins 14 passed / 21 subtests;
tests/test_merger_lease_finalization.py 59 passed / 42 subtests; targeted
pr_work_lease + reviewer/merger lease + adoption + provenance + acquire/release
+ anti-stomp + capability-map + decision-lock + report-validator suites 309
passed / 41 subtests; full suite 3326 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 10:12:49 -04:00

957 lines
37 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, 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()
Reference in New Issue View Git Blame Copy Permalink