Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  jcwalker3andClaude Opus 5
|
f49e781102 |
+83
-44
@@ -2781,36 +2781,6 @@ def _collect_issue_duplicate_context(
|
||||
return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
|
||||
|
||||
|
||||
def _owning_pr_continuation_from_lock(lock_record: dict | None) -> dict | None:
|
||||
"""Owning-PR continuation evidence a persisted lock still proves (#945).
|
||||
|
||||
``gitea_lock_issue`` grants the duplicate-work waiver from either a
|
||||
sanctioned dead-session recovery (#755) or a sanctioned exact-owner renewal
|
||||
(#760), in that precedence. Every later enforcement path — commit,
|
||||
create-PR, push-ownership, and the read-only duplicate assessor — re-derives
|
||||
ownership from the durable lock instead of that live assessment.
|
||||
|
||||
Until #945 only the recovery half was rebuilt there, so an ordinary
|
||||
exact-owner renewal lost its waiver the moment ``gitea_lock_issue``
|
||||
returned: the author renewed successfully and was then refused
|
||||
``duplicate_commit_prevented`` with ``owning_pr_recovery_exempted: false``
|
||||
on the very PR the renewal had just proved it owned.
|
||||
|
||||
Resolving both halves here, in the same precedence the lock path applies,
|
||||
keeps the answer from drifting between the gate that grants the waiver and
|
||||
the gates that enforce it. This only decides which server-written block the
|
||||
token is rebuilt from — the token is still re-validated against live PR
|
||||
state by ``issue_work_duplicate_gate._assess_owning_pr_exemption``, which
|
||||
remains the single authoritative policy for whether an exemption applies.
|
||||
"""
|
||||
if not lock_record:
|
||||
return None
|
||||
recovered = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
|
||||
if recovered:
|
||||
return recovered
|
||||
return issue_lock_renewal.owning_pr_renewal_from_lock(lock_record)
|
||||
|
||||
|
||||
def _assess_issue_duplicate_gate(
|
||||
issue_number: int,
|
||||
*,
|
||||
@@ -2863,11 +2833,6 @@ def _enforce_locked_issue_duplicate_recheck(
|
||||
commit and create-PR phases run in their own calls, long after the recovery
|
||||
assessment ended, so without this they re-block the very PR the recovery
|
||||
already proved belongs to this author.
|
||||
|
||||
#945: an exact-owner *renewal* (#760) owns its open PR for exactly the same
|
||||
reason, and ``gitea_lock_issue`` already waives the blocker for both. Both
|
||||
halves are resolved together here so the renewal waiver survives past the
|
||||
lock call instead of expiring with it.
|
||||
"""
|
||||
lock_data = _load_existing_issue_lock()
|
||||
if not lock_data:
|
||||
@@ -2891,7 +2856,9 @@ def _enforce_locked_issue_duplicate_recheck(
|
||||
auth=auth,
|
||||
locked_branch=locked_branch,
|
||||
phase=phase,
|
||||
recovered_owning_pr=_owning_pr_continuation_from_lock(lock_data),
|
||||
recovered_owning_pr=issue_lock_recovery.recovered_owning_pr_from_lock(
|
||||
lock_data
|
||||
),
|
||||
)
|
||||
if gate.get("block"):
|
||||
return gate
|
||||
@@ -3613,6 +3580,64 @@ def _authenticated_username(host: str):
|
||||
return user
|
||||
|
||||
|
||||
# #943: process-local session identifier. The pre-existing call sites that mint a
|
||||
# session id (workflow dashboard, lease adopt, lease reclaim) all build the same
|
||||
# "<profile>-<pid>-<hex>" shape when the caller supplies none. Binding it once per
|
||||
# process keeps lease-ownership comparisons stable for the life of the session
|
||||
# instead of minting a fresh identifier — and therefore a fresh owner — on every
|
||||
# call. Process-local only, never shared to a file (same rationale as the
|
||||
# immutable session context in session_context_binding).
|
||||
_ACTIVE_SESSION_ID: str | None = None
|
||||
|
||||
|
||||
def _active_username() -> str | None:
|
||||
"""Authenticated identity bound to this session, or None when unbound.
|
||||
|
||||
Reads the immutable #714 session context that ``gitea_whoami`` seeds; it is
|
||||
the authoritative identity pin every other mutation gate already consults.
|
||||
Never re-derives or fabricates an identity: an unbound context returns None
|
||||
so callers fail closed instead of acting as an unverified actor.
|
||||
"""
|
||||
ctx = session_ctx.get_session_context() or {}
|
||||
return ((ctx.get("identity") or "").strip()) or None
|
||||
|
||||
|
||||
def _active_profile_name() -> str | None:
|
||||
"""Active runtime profile name, or None when it cannot be determined.
|
||||
|
||||
The live profile is authoritative (``get_profile``); the bound session
|
||||
context is consulted only when the profile cannot be read, so the reported
|
||||
name always describes the profile actually serving this process.
|
||||
"""
|
||||
try:
|
||||
profile = get_profile() or {}
|
||||
except Exception:
|
||||
profile = {}
|
||||
name = (profile.get("profile_name") or "").strip()
|
||||
if name:
|
||||
return name
|
||||
ctx = session_ctx.get_session_context() or {}
|
||||
return ((ctx.get("profile_name") or "").strip()) or None
|
||||
|
||||
|
||||
def _current_session_id() -> str | None:
|
||||
"""Session identifier for this process, or None when the profile is unknown.
|
||||
|
||||
Uses the same "<profile>-<pid>-<hex>" shape as the existing lease call
|
||||
sites. Bound once per process so repeated calls describe one session; fails
|
||||
soft to None when the profile is undeterminable, letting callers fail closed
|
||||
rather than inventing an owner.
|
||||
"""
|
||||
global _ACTIVE_SESSION_ID
|
||||
if _ACTIVE_SESSION_ID:
|
||||
return _ACTIVE_SESSION_ID
|
||||
profile_name = _active_profile_name()
|
||||
if not profile_name:
|
||||
return None
|
||||
_ACTIVE_SESSION_ID = f"{profile_name}-{os.getpid()}-{uuid.uuid4().hex[:8]}"
|
||||
return _ACTIVE_SESSION_ID
|
||||
|
||||
|
||||
def _authenticated_actor(host: str) -> dict:
|
||||
"""Resolve the authenticated actor's stable identity (#709 F7 review 438).
|
||||
|
||||
@@ -5173,10 +5198,9 @@ def gitea_assess_work_issue_duplicate(
|
||||
recovered_owning_pr = None
|
||||
lock_data = _load_existing_issue_lock()
|
||||
if lock_data and int(lock_data.get("issue_number") or 0) == int(issue_number):
|
||||
# #945: rebuilt from a sanctioned recovery *or* a sanctioned exact-owner
|
||||
# renewal, so this read-only assessor reports the same disposition the
|
||||
# commit and create-PR gates will enforce.
|
||||
recovered_owning_pr = _owning_pr_continuation_from_lock(lock_data)
|
||||
recovered_owning_pr = issue_lock_recovery.recovered_owning_pr_from_lock(
|
||||
lock_data
|
||||
)
|
||||
gate = _assess_issue_duplicate_gate(
|
||||
issue_number,
|
||||
h=h,
|
||||
@@ -9936,6 +9960,24 @@ def gitea_commit_files(
|
||||
}
|
||||
|
||||
|
||||
def _author_mutation_block(reasons: list[str], **extra) -> dict:
|
||||
"""Uniform fail-closed shape for an author mutation refused after a reviewer stop.
|
||||
|
||||
#943: referenced by ``gitea_bootstrap_author_issue_worktree`` and never
|
||||
defined, so the reviewer-stop refusal path raised ``NameError`` instead of
|
||||
returning its refusal. Mirrors the inline shape the other author mutations
|
||||
return for the same ``check_author_mutation_after_reviewer_stop`` block.
|
||||
"""
|
||||
payload = {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"outcome": "REFUSED",
|
||||
"reasons": reasons,
|
||||
}
|
||||
payload.update(extra)
|
||||
return payload
|
||||
|
||||
|
||||
def _publication_block(reasons: list[str], **extra) -> dict:
|
||||
"""Uniform fail-closed shape for publication refusals (#812 AC20)."""
|
||||
payload = {
|
||||
@@ -19458,10 +19500,7 @@ def _prove_author_ownership_for_pr(
|
||||
# advance is the one recovery already sanctioned, not a foreign head.
|
||||
recovered_owning_pr = None
|
||||
if proven and lock_record:
|
||||
# #945: a sanctioned exact-owner renewal proves the same ownership of
|
||||
# the same PR, so the push gate resolves both halves rather than seeing
|
||||
# only the recovery one.
|
||||
candidate = _owning_pr_continuation_from_lock(lock_record)
|
||||
candidate = issue_lock_recovery.recovered_owning_pr_from_lock(lock_record)
|
||||
if candidate and int(candidate.get("pr_number") or 0) == int(pr_number):
|
||||
recovered_owning_pr = candidate
|
||||
return {
|
||||
|
||||
@@ -436,90 +436,6 @@ def owning_pr_renewal_evidence(
|
||||
}
|
||||
|
||||
|
||||
def owning_pr_renewal_from_lock(
|
||||
lock_record: Mapping[str, Any] | None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""Rebuild owning-PR renewal evidence from a persisted lock (#945).
|
||||
|
||||
The renewal mirror of ``issue_lock_recovery.recovered_owning_pr_from_lock``.
|
||||
``owning_pr_renewal_evidence`` supplies the waiver for the duration of the
|
||||
``gitea_lock_issue`` call only. The commit, push, create-PR, and
|
||||
duplicate-assessment gates run later in their own calls and re-derive
|
||||
ownership from the durable lock instead — so without this the open PR that
|
||||
renewal already proved belongs to this author reappears there as competing
|
||||
duplicate work, and the exact owner is refused with
|
||||
``duplicate_commit_prevented`` despite complete matching evidence.
|
||||
|
||||
This reads only the ``lease_renewal`` block that the server itself writes,
|
||||
on a lock the caller must already own. Like the recovery mirror it is a
|
||||
re-read of server-derived state, never a fresh assertion: a caller able to
|
||||
forge it could equally forge the lock file every other ownership gate
|
||||
already treats as authoritative.
|
||||
|
||||
Renewal has no descendant case — the assessor required the local, remote and
|
||||
PR heads to be equal — so that equality is re-checked here, and the record
|
||||
must still name the claimant the lock records.
|
||||
"""
|
||||
if not isinstance(lock_record, Mapping):
|
||||
return None
|
||||
record = lock_record.get("lease_renewal")
|
||||
if not isinstance(record, Mapping) or not record.get("renewed"):
|
||||
return None
|
||||
|
||||
branch_name = _text(record.get("branch_name")) or _text(
|
||||
lock_record.get("branch_name")
|
||||
)
|
||||
pr_head = _text(record.get("pr_head_sha"))
|
||||
local_head = _text(record.get("head_sha"))
|
||||
remote_head = _text(record.get("remote_head_sha"))
|
||||
raw_pr_number = record.get("pr_number")
|
||||
raw_issue_number = lock_record.get("issue_number")
|
||||
|
||||
if raw_pr_number is None or raw_issue_number is None:
|
||||
return None
|
||||
if not branch_name or not pr_head:
|
||||
return None
|
||||
# The assessor required all three heads to agree before it granted renewal.
|
||||
# Re-check, so a truncated, drifted, or hand-built record cannot widen the
|
||||
# exemption past the single head the renewal disposition actually proved.
|
||||
if not local_head or not remote_head:
|
||||
return None
|
||||
if pr_head != local_head or pr_head != remote_head:
|
||||
return None
|
||||
# Renewal is refused outright unless the durable lock records both a
|
||||
# claimant username and profile, so a sanctioned record always carries them.
|
||||
# Requiring them to still agree keeps a renewal block from being reused
|
||||
# under an identity or profile the lock no longer names.
|
||||
claimant = lock_record.get("claimant")
|
||||
if not isinstance(claimant, Mapping):
|
||||
lease = lock_record.get("work_lease")
|
||||
claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
|
||||
if not isinstance(claimant, Mapping):
|
||||
return None
|
||||
identity = _text(record.get("identity"))
|
||||
profile = _text(record.get("profile"))
|
||||
if not identity or identity != _text(claimant.get("username")):
|
||||
return None
|
||||
if not profile or profile != _text(claimant.get("profile")):
|
||||
return None
|
||||
|
||||
try:
|
||||
pr_number = int(raw_pr_number)
|
||||
issue_number = int(raw_issue_number)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"pr_number": pr_number,
|
||||
"branch_name": branch_name,
|
||||
"head_sha": pr_head,
|
||||
"recorded_head": pr_head,
|
||||
"accepted_head": pr_head,
|
||||
"head_relation": "equal",
|
||||
}
|
||||
|
||||
|
||||
def build_renewal_record(
|
||||
assessment: Mapping[str, Any] | None,
|
||||
*,
|
||||
|
||||
@@ -0,0 +1,458 @@
|
||||
"""Regression: the author bootstrap wrapper's runtime-context helpers (#943).
|
||||
|
||||
``gitea_bootstrap_author_issue_worktree`` passed three values down to
|
||||
``author_issue_bootstrap.bootstrap_author_issue_worktree``::
|
||||
|
||||
active_identity=_active_username(),
|
||||
active_profile=_active_profile_name(),
|
||||
owner_session=_current_session_id(),
|
||||
|
||||
None of those three names was ever defined. Commit ``a942afe`` (#850) introduced
|
||||
the references and no definition, so every call — dry-run included — raised
|
||||
``NameError: name '_active_username' is not defined`` while evaluating the
|
||||
arguments, before the bootstrap service was entered.
|
||||
|
||||
The defect was unreachable until PR #942 (#941) wired the bootstrap scope into
|
||||
``workflow_scope_guard``: until then ``verify_preflight_purity`` refused first
|
||||
with ``missing_issue_worktree``, so the guard fix is what exposed this.
|
||||
|
||||
``test_every_global_referenced_by_the_wrapper_resolves`` is the test that would
|
||||
have caught the original defect: it resolves every global name the wrapper's
|
||||
body references. Asserting only that the three known helpers now exist would
|
||||
not generalise to the next missing reference.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import builtins
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
|
||||
import author_issue_bootstrap as aib
|
||||
import create_issue_bootstrap as cib
|
||||
import gitea_mcp_server as gms
|
||||
import workflow_scope_guard
|
||||
|
||||
BOOTSTRAP_TASK = "bootstrap_author_issue_worktree"
|
||||
WRAPPER_NAME = "gitea_bootstrap_author_issue_worktree"
|
||||
RUNTIME_HELPERS = ("_active_username", "_active_profile_name", "_current_session_id")
|
||||
|
||||
# "<profile>-<pid>-<hex8>", the shape the pre-existing lease call sites mint.
|
||||
SESSION_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+-\d+-[0-9a-f]{8}$")
|
||||
|
||||
|
||||
def _make_control_repo(tmp: str) -> tuple[str, str]:
|
||||
"""Create a clean control checkout on master and return (path, head)."""
|
||||
repo = os.path.join(tmp, "repo")
|
||||
os.makedirs(os.path.join(repo, "branches"))
|
||||
subprocess.check_call(
|
||||
["git", "init", "-b", "master", repo],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
subprocess.check_call(
|
||||
[
|
||||
"git", "-C", repo,
|
||||
"-c", "user.email=t@t", "-c", "user.name=t",
|
||||
"commit", "--allow-empty", "-m", "init",
|
||||
],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
head = subprocess.check_output(
|
||||
["git", "-C", repo, "rev-parse", "HEAD"], text=True
|
||||
).strip()
|
||||
return repo, head
|
||||
|
||||
|
||||
def _wrapper_ast() -> ast.FunctionDef:
|
||||
"""Return the AST of the bootstrap wrapper as it exists on disk.
|
||||
|
||||
Read from source rather than ``inspect``: the tool decorator may replace the
|
||||
callable, and the defect lived in the *source* argument expressions.
|
||||
"""
|
||||
path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"gitea_mcp_server.py")
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
tree = ast.parse(fh.read())
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.FunctionDef) and node.name == WRAPPER_NAME:
|
||||
return node
|
||||
raise AssertionError(f"{WRAPPER_NAME} not found in gitea_mcp_server.py")
|
||||
|
||||
|
||||
class RuntimeHelperResolutionTests(unittest.TestCase):
|
||||
"""AC: every runtime helper the wrapper references is defined and callable."""
|
||||
|
||||
def test_three_named_helpers_are_defined_and_callable(self):
|
||||
for name in RUNTIME_HELPERS:
|
||||
with self.subTest(helper=name):
|
||||
self.assertTrue(
|
||||
hasattr(gms, name), f"{name} is referenced but not defined"
|
||||
)
|
||||
self.assertTrue(callable(getattr(gms, name)), f"{name} not callable")
|
||||
|
||||
def test_helpers_accept_zero_arguments_as_called(self):
|
||||
"""The wrapper calls each with no arguments; the signature must allow it."""
|
||||
prior = gms._ACTIVE_SESSION_ID
|
||||
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
|
||||
for name in RUNTIME_HELPERS:
|
||||
with self.subTest(helper=name):
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None):
|
||||
gms._ACTIVE_SESSION_ID = None
|
||||
getattr(gms, name)() # must not raise TypeError
|
||||
|
||||
def test_every_global_referenced_by_the_wrapper_resolves(self):
|
||||
"""The generalised form of this defect: an unresolvable global name.
|
||||
|
||||
Collects every ``Name`` load in the wrapper body, subtracts locals
|
||||
(arguments, assignments, comprehension targets, imports), and asserts the
|
||||
remainder resolves against module globals or builtins.
|
||||
"""
|
||||
fn = _wrapper_ast()
|
||||
bound: set[str] = {a.arg for a in fn.args.args}
|
||||
bound |= {a.arg for a in fn.args.kwonlyargs}
|
||||
if fn.args.vararg:
|
||||
bound.add(fn.args.vararg.arg)
|
||||
if fn.args.kwarg:
|
||||
bound.add(fn.args.kwarg.arg)
|
||||
for node in ast.walk(fn):
|
||||
if isinstance(node, ast.Name) and isinstance(node.ctx, (ast.Store, ast.Del)):
|
||||
bound.add(node.id)
|
||||
elif isinstance(node, (ast.Import, ast.ImportFrom)):
|
||||
for alias in node.names:
|
||||
bound.add((alias.asname or alias.name).split(".")[0])
|
||||
elif isinstance(node, ast.ExceptHandler) and node.name:
|
||||
bound.add(node.name)
|
||||
|
||||
unresolved = sorted(
|
||||
node.id
|
||||
for node in ast.walk(fn)
|
||||
if isinstance(node, ast.Name)
|
||||
and isinstance(node.ctx, ast.Load)
|
||||
and node.id not in bound
|
||||
and not hasattr(gms, node.id)
|
||||
and not hasattr(builtins, node.id)
|
||||
)
|
||||
self.assertEqual(
|
||||
unresolved, [], f"{WRAPPER_NAME} references undefined globals: {unresolved}"
|
||||
)
|
||||
|
||||
def test_wrapper_still_passes_all_three_runtime_values(self):
|
||||
"""Guard the wiring itself: the fix must not be 'stop passing them'."""
|
||||
fn = _wrapper_ast()
|
||||
called = {
|
||||
node.func.id
|
||||
for node in ast.walk(fn)
|
||||
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
|
||||
}
|
||||
for name in RUNTIME_HELPERS:
|
||||
with self.subTest(helper=name):
|
||||
self.assertIn(name, called)
|
||||
|
||||
|
||||
class ActiveUsernameTests(unittest.TestCase):
|
||||
"""AC: identity comes from the authoritative pin, and fails closed."""
|
||||
|
||||
def test_returns_identity_from_bound_session_context(self):
|
||||
with mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"identity": "jcwalker3", "profile_name": "prgs-author"},
|
||||
):
|
||||
self.assertEqual(gms._active_username(), "jcwalker3")
|
||||
|
||||
def test_unbound_context_returns_none_so_callers_fail_closed(self):
|
||||
with mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None
|
||||
):
|
||||
self.assertIsNone(gms._active_username())
|
||||
|
||||
def test_blank_identity_is_not_treated_as_an_identity(self):
|
||||
for blank in ("", " ", None):
|
||||
with self.subTest(identity=blank):
|
||||
with mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"identity": blank},
|
||||
):
|
||||
self.assertIsNone(gms._active_username())
|
||||
|
||||
def test_identity_is_not_fabricated_from_the_profile(self):
|
||||
"""A profile's expected username must never stand in for a real identity."""
|
||||
with mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"identity": None, "expected_username": "jcwalker3"},
|
||||
):
|
||||
self.assertIsNone(gms._active_username())
|
||||
|
||||
|
||||
class ActiveProfileNameTests(unittest.TestCase):
|
||||
"""AC: profile name comes from the live profile, context only as fallback."""
|
||||
|
||||
def test_prefers_the_live_profile(self):
|
||||
with mock.patch.object(
|
||||
gms, "get_profile", return_value={"profile_name": "prgs-author"}
|
||||
), mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"profile_name": "prgs-reviewer"},
|
||||
):
|
||||
self.assertEqual(gms._active_profile_name(), "prgs-author")
|
||||
|
||||
def test_falls_back_to_session_context_when_profile_unreadable(self):
|
||||
with mock.patch.object(gms, "get_profile", side_effect=RuntimeError("no cfg")), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"profile_name": "prgs-author"}):
|
||||
self.assertEqual(gms._active_profile_name(), "prgs-author")
|
||||
|
||||
def test_returns_none_when_neither_source_knows(self):
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None):
|
||||
self.assertIsNone(gms._active_profile_name())
|
||||
|
||||
|
||||
class CurrentSessionIdTests(unittest.TestCase):
|
||||
"""AC: a real, stable session identifier — never a fresh owner per call."""
|
||||
|
||||
def setUp(self):
|
||||
self._prior = gms._ACTIVE_SESSION_ID
|
||||
gms._ACTIVE_SESSION_ID = None
|
||||
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", self._prior)
|
||||
|
||||
def test_shape_matches_the_existing_lease_call_sites(self):
|
||||
with mock.patch.object(
|
||||
gms, "get_profile", return_value={"profile_name": "prgs-author"}
|
||||
):
|
||||
sid = gms._current_session_id()
|
||||
self.assertRegex(sid, SESSION_ID_RE)
|
||||
self.assertTrue(sid.startswith("prgs-author-"))
|
||||
self.assertIn(str(os.getpid()), sid)
|
||||
|
||||
def test_stable_across_calls_within_one_process(self):
|
||||
"""A new id per call would make lease-ownership checks unsatisfiable."""
|
||||
with mock.patch.object(
|
||||
gms, "get_profile", return_value={"profile_name": "prgs-author"}
|
||||
):
|
||||
first = gms._current_session_id()
|
||||
second = gms._current_session_id()
|
||||
third = gms._current_session_id()
|
||||
self.assertEqual(first, second)
|
||||
self.assertEqual(second, third)
|
||||
|
||||
def test_returns_none_when_profile_undeterminable(self):
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None):
|
||||
self.assertIsNone(gms._current_session_id())
|
||||
|
||||
def test_none_result_is_not_memoised_as_a_session(self):
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None):
|
||||
self.assertIsNone(gms._current_session_id())
|
||||
with mock.patch.object(
|
||||
gms, "get_profile", return_value={"profile_name": "prgs-author"}
|
||||
):
|
||||
self.assertIsNotNone(gms._current_session_id())
|
||||
|
||||
|
||||
class BootstrapServiceReachedTests(unittest.TestCase):
|
||||
"""AC: the values the helpers produce carry a dry-run into the service."""
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.tmp = self._tmp.name
|
||||
self.repo, self.head = _make_control_repo(self.tmp)
|
||||
self.journals = os.path.join(self.tmp, "journals")
|
||||
os.makedirs(self.journals)
|
||||
|
||||
def _bootstrap(self, **over):
|
||||
kwargs = dict(
|
||||
issue_number=943,
|
||||
canonical_repo_root=self.repo,
|
||||
expected_base_sha=self.head,
|
||||
branch_name="fix/issue-943-runtime-context-helpers",
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
active_identity="jcwalker3",
|
||||
active_profile="prgs-author",
|
||||
owner_session="prgs-author-4242-abcdef12",
|
||||
lock_dir=self.journals,
|
||||
idempotency_key="test-943",
|
||||
dry_run=True,
|
||||
)
|
||||
kwargs.update(over)
|
||||
return aib.bootstrap_author_issue_worktree(**kwargs)
|
||||
|
||||
def test_dry_run_succeeds_with_helper_produced_bindings(self):
|
||||
"""Feed the service exactly what the live helpers return."""
|
||||
prior = gms._ACTIVE_SESSION_ID
|
||||
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
|
||||
with mock.patch.object(
|
||||
gms, "get_profile", return_value={"profile_name": "prgs-author"}
|
||||
), mock.patch.object(
|
||||
gms.session_ctx, "get_session_context",
|
||||
return_value={"identity": "jcwalker3", "profile_name": "prgs-author"},
|
||||
):
|
||||
gms._ACTIVE_SESSION_ID = None
|
||||
identity = gms._active_username()
|
||||
profile = gms._active_profile_name()
|
||||
session = gms._current_session_id()
|
||||
|
||||
res = self._bootstrap(
|
||||
active_identity=identity, active_profile=profile, owner_session=session
|
||||
)
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertTrue(res.get("dry_run"))
|
||||
self.assertEqual(res.get("issue_number"), 943)
|
||||
self.assertEqual(res.get("base_sha"), self.head)
|
||||
|
||||
def test_dry_run_creates_no_branch_worktree_or_lease(self):
|
||||
res = self._bootstrap()
|
||||
self.assertTrue(res.get("success"), res)
|
||||
|
||||
branches = subprocess.check_output(
|
||||
["git", "-C", self.repo, "branch", "--list"], text=True
|
||||
)
|
||||
self.assertNotIn("issue-943", branches)
|
||||
worktrees = subprocess.check_output(
|
||||
["git", "-C", self.repo, "worktree", "list"], text=True
|
||||
)
|
||||
self.assertNotIn("issue-943", worktrees)
|
||||
self.assertFalse(
|
||||
os.path.exists(os.path.join(self.repo, "branches",
|
||||
"fix-issue-943-runtime-context-helpers"))
|
||||
)
|
||||
journal = res.get("phase_journal") or {}
|
||||
self.assertFalse(journal.get("completed"))
|
||||
self.assertFalse(any((journal.get("artifacts_created") or {}).values()))
|
||||
self.assertIsNone(journal.get("lease_id"))
|
||||
self.assertIsNone(journal.get("assignment_id"))
|
||||
|
||||
def test_missing_identity_fails_closed(self):
|
||||
res = self._bootstrap(active_identity=None)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "missing_active_identity")
|
||||
|
||||
def test_missing_profile_fails_closed(self):
|
||||
res = self._bootstrap(active_profile=" ")
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "missing_active_profile")
|
||||
|
||||
def test_missing_session_fails_closed(self):
|
||||
res = self._bootstrap(owner_session=None)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "missing_owner_session")
|
||||
|
||||
def test_expected_base_mismatch_fails_closed(self):
|
||||
res = self._bootstrap(expected_base_sha="0" * 40)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertEqual(res.get("reason_code"), "stale_concurrency_pin")
|
||||
|
||||
def test_unbound_runtime_context_cannot_reach_the_service(self):
|
||||
"""With nothing bound, the helpers yield None and the service refuses."""
|
||||
prior = gms._ACTIVE_SESSION_ID
|
||||
self.addCleanup(setattr, gms, "_ACTIVE_SESSION_ID", prior)
|
||||
with mock.patch.object(gms, "get_profile", return_value={}), \
|
||||
mock.patch.object(
|
||||
gms.session_ctx, "get_session_context", return_value=None):
|
||||
gms._ACTIVE_SESSION_ID = None
|
||||
res = self._bootstrap(
|
||||
active_identity=gms._active_username(),
|
||||
active_profile=gms._active_profile_name(),
|
||||
owner_session=gms._current_session_id(),
|
||||
)
|
||||
self.assertFalse(res.get("success"))
|
||||
self.assertIn(
|
||||
res.get("reason_code"),
|
||||
{"missing_owner_session", "missing_active_identity",
|
||||
"missing_active_profile"},
|
||||
)
|
||||
|
||||
def test_apply_reaches_the_intended_transition(self):
|
||||
res = self._bootstrap(dry_run=False)
|
||||
self.assertTrue(res.get("success"), res)
|
||||
self.assertNotEqual(res.get("dry_run"), True)
|
||||
branches = subprocess.check_output(
|
||||
["git", "-C", self.repo, "branch", "--list"], text=True
|
||||
)
|
||||
self.assertIn("issue-943", branches)
|
||||
self.assertTrue(os.path.isdir(res.get("worktree_path") or ""))
|
||||
|
||||
|
||||
class Issue941ScopeGuardNotRegressedTests(unittest.TestCase):
|
||||
"""AC: PR #942's bootstrap-scope wiring still holds."""
|
||||
|
||||
def setUp(self):
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.repo, self.head = _make_control_repo(self._tmp.name)
|
||||
|
||||
def _assessment(self, task: str = BOOTSTRAP_TASK) -> dict:
|
||||
return aib.assess_author_issue_bootstrap(
|
||||
workspace_path=self.repo,
|
||||
canonical_repo_root=self.repo,
|
||||
current_branch="master",
|
||||
head_sha=self.head,
|
||||
porcelain_status="",
|
||||
remote_master_sha=self.head,
|
||||
task=task,
|
||||
)
|
||||
|
||||
def test_bootstrap_task_still_permitted_from_clean_control_checkout(self):
|
||||
res = workflow_scope_guard.assess_root_source_mutation(
|
||||
workspace_path=self.repo,
|
||||
canonical_repo_root=self.repo,
|
||||
role_kind="author",
|
||||
mutation_task=BOOTSTRAP_TASK,
|
||||
porcelain_status="",
|
||||
bootstrap_assessment=self._assessment(),
|
||||
)
|
||||
self.assertFalse(res.get("block"), res)
|
||||
self.assertNotEqual(
|
||||
res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
|
||||
)
|
||||
|
||||
def test_bootstrap_task_still_blocked_without_evidence(self):
|
||||
res = workflow_scope_guard.assess_root_source_mutation(
|
||||
workspace_path=self.repo,
|
||||
canonical_repo_root=self.repo,
|
||||
role_kind="author",
|
||||
mutation_task=BOOTSTRAP_TASK,
|
||||
porcelain_status="",
|
||||
)
|
||||
self.assertTrue(res.get("block"))
|
||||
self.assertEqual(
|
||||
res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
|
||||
)
|
||||
|
||||
def test_ordinary_author_mutation_still_blocked_from_control_checkout(self):
|
||||
res = workflow_scope_guard.assess_root_source_mutation(
|
||||
workspace_path=self.repo,
|
||||
canonical_repo_root=self.repo,
|
||||
role_kind="author",
|
||||
mutation_task="commit_files",
|
||||
porcelain_status="",
|
||||
bootstrap_assessment=self._assessment(),
|
||||
)
|
||||
self.assertTrue(res.get("block"))
|
||||
self.assertEqual(
|
||||
res.get("blocker_kind"), workflow_scope_guard.BLOCKER_MISSING_WORKTREE
|
||||
)
|
||||
|
||||
def test_create_issue_bootstrap_unchanged(self):
|
||||
self.assertTrue(cib.is_create_issue_task("create_issue"))
|
||||
self.assertFalse(cib.is_create_issue_task(BOOTSTRAP_TASK))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,465 +0,0 @@
|
||||
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: F401,E402
|
||||
"""Exact-owner renewal keeps its owning-PR waiver past lock_issue (#945).
|
||||
|
||||
#755 taught the duplicate-work gate that a sanctioned *dead-session recovery*
|
||||
owns its open PR, and #768 taught the later gates to rebuild that proof from the
|
||||
durable lock. #760 added the exact-owner *renewal* disposition and granted it
|
||||
the same waiver inside ``gitea_lock_issue`` — but never added the matching
|
||||
rebuild. So an ordinary renewal held the waiver only for the duration of the
|
||||
lock call: ``_enforce_locked_issue_duplicate_recheck`` asked
|
||||
``recovered_owning_pr_from_lock``, which reads only ``dead_session_recovery``,
|
||||
and the very next commit was refused ``duplicate_commit_prevented`` with
|
||||
``owning_pr_recovery_exempted: false`` on the PR the renewal had just proved.
|
||||
|
||||
``TestPreFixReproduction`` pins that defect directly: the recovery-only rebuild
|
||||
still returns ``None`` for a renewal lock, which is exactly why the gates lost
|
||||
the waiver. Everything else proves the renewal half now survives, that recovery
|
||||
is unchanged, and that no path grants an exemption on weaker evidence.
|
||||
|
||||
Every fixture here is an in-memory mapping. Nothing writes a branch, worktree,
|
||||
lock file, lease, comment, or PR (#945 AC18).
|
||||
"""
|
||||
import copy
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_mcp_server # noqa: E402
|
||||
import issue_lock_recovery # noqa: E402
|
||||
import issue_lock_renewal # noqa: E402
|
||||
from issue_work_duplicate_gate import ( # noqa: E402
|
||||
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
|
||||
PHASE_COMMIT,
|
||||
PHASE_CREATE_PR,
|
||||
PHASE_LOCK,
|
||||
PHASE_PUSH,
|
||||
assess_work_issue_duplicate_gate,
|
||||
)
|
||||
|
||||
ISSUE = 4945
|
||||
OWNING_PR = 4946
|
||||
OTHER_PR = 4947
|
||||
BRANCH = f"fix/issue-{ISSUE}-owning-pr-renewal"
|
||||
OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
|
||||
HEAD = "a" * 40
|
||||
OTHER_HEAD = "b" * 40
|
||||
IDENTITY = "example-user"
|
||||
PROFILE = "test-author-prgs"
|
||||
|
||||
|
||||
def renewal_record(**overrides):
|
||||
"""The ``lease_renewal`` block ``build_renewal_record`` writes on success."""
|
||||
record = {
|
||||
"renewed": True,
|
||||
"renewed_at": "2026-01-01T00:00:00Z",
|
||||
"prior_pid": 4242,
|
||||
"prior_pid_alive": True,
|
||||
"prior_expires_at": "2026-01-01T00:00:00Z",
|
||||
"replacement_pid": 4243,
|
||||
"new_expires_at": "2026-01-01T00:10:00Z",
|
||||
"identity": IDENTITY,
|
||||
"profile": PROFILE,
|
||||
"branch_name": BRANCH,
|
||||
"worktree_path": f"branches/issue-{ISSUE}-owning-pr-renewal",
|
||||
"head_sha": HEAD,
|
||||
"remote_head_sha": HEAD,
|
||||
"pr_head_sha": HEAD,
|
||||
"pr_number": OWNING_PR,
|
||||
"reason": "expired lease renewed by its exact recorded owner",
|
||||
"proof": [],
|
||||
}
|
||||
record.update(overrides)
|
||||
return record
|
||||
|
||||
|
||||
def renewal_lock(record=None, *, issue_number=ISSUE, claimant=True, **lock_overrides):
|
||||
lock = {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": BRANCH,
|
||||
"lease_renewal": renewal_record() if record is None else record,
|
||||
}
|
||||
if claimant:
|
||||
lock["claimant"] = {"username": IDENTITY, "profile": PROFILE}
|
||||
lock.update(lock_overrides)
|
||||
return lock
|
||||
|
||||
|
||||
def recovery_lock(pr_number=OWNING_PR, head=HEAD):
|
||||
"""A lock carrying sanctioned dead-session recovery evidence (#755/#768)."""
|
||||
return {
|
||||
"issue_number": ISSUE,
|
||||
"branch_name": BRANCH,
|
||||
"claimant": {"username": IDENTITY, "profile": PROFILE},
|
||||
"dead_session_recovery": {
|
||||
"recovered": True,
|
||||
"branch_name": BRANCH,
|
||||
"pr_number": pr_number,
|
||||
"pr_head": head,
|
||||
"recorded_head": head,
|
||||
"accepted_head": head,
|
||||
"head_relation": issue_lock_recovery.HEAD_RELATION_EQUAL,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
|
||||
return {
|
||||
"number": number,
|
||||
"title": f"fix: something (Closes #{issue})",
|
||||
"body": f"Closes #{issue}.",
|
||||
"head": {"ref": ref, "sha": sha},
|
||||
}
|
||||
|
||||
|
||||
def gate(phase, *, token, open_prs=None, branch_names=None, locked_branch=BRANCH):
|
||||
return assess_work_issue_duplicate_gate(
|
||||
ISSUE,
|
||||
open_prs=[owning_pr()] if open_prs is None else open_prs,
|
||||
branch_names=branch_names or [],
|
||||
claim_entry={},
|
||||
locked_branch=locked_branch,
|
||||
phase=phase,
|
||||
recovered_owning_pr=token,
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────── the defect this issue exists to fix ─────────────────────
|
||||
|
||||
|
||||
class TestPreFixReproduction(unittest.TestCase):
|
||||
"""The exact wiring gap: renewal evidence was invisible to later gates."""
|
||||
|
||||
def test_recovery_only_rebuild_cannot_see_a_renewal_lock(self):
|
||||
# This is the pre-fix behaviour of every enforcement path. It is correct
|
||||
# for the recovery rebuild to ignore a renewal block -- the defect was
|
||||
# that nothing else looked at it.
|
||||
self.assertIsNone(
|
||||
issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
|
||||
)
|
||||
|
||||
def test_renewal_lock_produced_no_exemption_before_the_fix(self):
|
||||
# Feeding the gate what the pre-fix code fed it (recovery rebuild only)
|
||||
# reproduces the reported refusal at the commit phase.
|
||||
token = issue_lock_recovery.recovered_owning_pr_from_lock(renewal_lock())
|
||||
result = gate(PHASE_COMMIT, token=token)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertEqual(result["outcome"], "duplicate_commit_prevented")
|
||||
self.assertFalse(result["owning_pr_recovery_exempted"])
|
||||
self.assertEqual(result["owning_pr_recovery_notes"], [])
|
||||
|
||||
def test_shared_resolver_now_sees_it(self):
|
||||
self.assertIsNotNone(
|
||||
gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
)
|
||||
|
||||
|
||||
# ───────────────────────── rebuild: the granted case ─────────────────────────
|
||||
|
||||
|
||||
class TestRenewalRebuildGranted(unittest.TestCase):
|
||||
def test_sanctioned_renewal_rebuilds_owning_pr_evidence(self):
|
||||
token = issue_lock_renewal.owning_pr_renewal_from_lock(renewal_lock())
|
||||
self.assertEqual(
|
||||
token,
|
||||
{
|
||||
"issue_number": ISSUE,
|
||||
"pr_number": OWNING_PR,
|
||||
"branch_name": BRANCH,
|
||||
"head_sha": HEAD,
|
||||
"recorded_head": HEAD,
|
||||
"accepted_head": HEAD,
|
||||
"head_relation": "equal",
|
||||
},
|
||||
)
|
||||
|
||||
def test_branch_falls_back_to_the_lock_branch(self):
|
||||
lock = renewal_lock(renewal_record(branch_name=""))
|
||||
token = issue_lock_renewal.owning_pr_renewal_from_lock(lock)
|
||||
self.assertEqual(token["branch_name"], BRANCH)
|
||||
|
||||
def test_claimant_may_live_under_work_lease(self):
|
||||
lock = renewal_lock(claimant=False)
|
||||
lock["work_lease"] = {"claimant": {"username": IDENTITY, "profile": PROFILE}}
|
||||
self.assertIsNotNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
|
||||
|
||||
def test_rebuild_does_not_mutate_the_lock(self):
|
||||
lock = renewal_lock()
|
||||
before = copy.deepcopy(lock)
|
||||
issue_lock_renewal.owning_pr_renewal_from_lock(lock)
|
||||
self.assertEqual(lock, before)
|
||||
|
||||
|
||||
# ───────────────────────── rebuild: fails closed ─────────────────────────
|
||||
|
||||
|
||||
class TestRenewalRebuildFailsClosed(unittest.TestCase):
|
||||
def assertNoEvidence(self, lock):
|
||||
self.assertIsNone(issue_lock_renewal.owning_pr_renewal_from_lock(lock))
|
||||
|
||||
def test_no_lock_at_all(self):
|
||||
self.assertNoEvidence(None)
|
||||
self.assertNoEvidence({})
|
||||
self.assertNoEvidence("not-a-mapping")
|
||||
|
||||
def test_lock_without_renewal_block(self):
|
||||
# A fresh claim, or a lock whose renewal block was replaced.
|
||||
self.assertNoEvidence({"issue_number": ISSUE, "branch_name": BRANCH})
|
||||
|
||||
def test_renewal_not_granted(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(renewed=False)))
|
||||
|
||||
def test_renewal_flag_missing(self):
|
||||
record = renewal_record()
|
||||
del record["renewed"]
|
||||
self.assertNoEvidence(renewal_lock(record))
|
||||
|
||||
def test_renewal_block_malformed(self):
|
||||
self.assertNoEvidence(renewal_lock("not-a-mapping"))
|
||||
|
||||
def test_local_head_diverged_from_pr_head(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(head_sha=OTHER_HEAD)))
|
||||
|
||||
def test_remote_head_diverged_from_pr_head(self):
|
||||
# Force-push or unrelated remote movement.
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha=OTHER_HEAD)))
|
||||
|
||||
def test_local_head_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(head_sha="")))
|
||||
|
||||
def test_remote_head_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(remote_head_sha="")))
|
||||
|
||||
def test_pr_head_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(pr_head_sha="")))
|
||||
|
||||
def test_pr_number_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(pr_number=None)))
|
||||
|
||||
def test_pr_number_malformed(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(pr_number="not-a-number")))
|
||||
|
||||
def test_issue_number_missing_from_lock(self):
|
||||
self.assertNoEvidence(renewal_lock(issue_number=None))
|
||||
|
||||
def test_branch_unknown_everywhere(self):
|
||||
lock = renewal_lock(renewal_record(branch_name=""))
|
||||
lock["branch_name"] = ""
|
||||
self.assertNoEvidence(lock)
|
||||
|
||||
def test_identity_mismatch(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(identity="someone-else")))
|
||||
|
||||
def test_profile_mismatch(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(profile="other-profile")))
|
||||
|
||||
def test_identity_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(identity="")))
|
||||
|
||||
def test_profile_missing(self):
|
||||
self.assertNoEvidence(renewal_lock(renewal_record(profile="")))
|
||||
|
||||
def test_claimant_absent(self):
|
||||
self.assertNoEvidence(renewal_lock(claimant=False))
|
||||
|
||||
def test_evidence_from_a_different_session_is_not_reusable(self):
|
||||
# A renewal block left by another workflow session names another
|
||||
# claimant, so the lock it is found on cannot inherit its authority.
|
||||
lock = renewal_lock()
|
||||
lock["claimant"] = {"username": "other-session-user", "profile": PROFILE}
|
||||
self.assertNoEvidence(lock)
|
||||
|
||||
|
||||
# ───────────────────────── the shared resolver ─────────────────────────
|
||||
|
||||
|
||||
class TestSharedResolver(unittest.TestCase):
|
||||
def test_recovery_lock_resolves_to_recovery_evidence(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
|
||||
self.assertEqual(token["pr_number"], OWNING_PR)
|
||||
|
||||
def test_renewal_lock_resolves_to_renewal_evidence(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
self.assertEqual(token["pr_number"], OWNING_PR)
|
||||
|
||||
def test_recovery_takes_precedence_over_renewal(self):
|
||||
# Same precedence gitea_lock_issue applies when granting the waiver, so
|
||||
# the answer cannot differ between the granting and enforcing paths.
|
||||
lock = recovery_lock(pr_number=OTHER_PR, head=OTHER_HEAD)
|
||||
lock["lease_renewal"] = renewal_record()
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(lock)
|
||||
self.assertEqual(token["pr_number"], OTHER_PR)
|
||||
|
||||
def test_no_evidence_resolves_to_none(self):
|
||||
self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock(None))
|
||||
self.assertIsNone(gitea_mcp_server._owning_pr_continuation_from_lock({}))
|
||||
self.assertIsNone(
|
||||
gitea_mcp_server._owning_pr_continuation_from_lock(
|
||||
{"issue_number": ISSUE, "branch_name": BRANCH}
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
# ────────────── every enforcement path uses the same decision ──────────────
|
||||
|
||||
|
||||
class TestEnforcementPathsShareOneDecision(unittest.TestCase):
|
||||
"""AC: commit, push and create-PR gates consume one authoritative token."""
|
||||
|
||||
def setUp(self):
|
||||
self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
|
||||
def test_commit_phase_permits_continuation(self):
|
||||
result = gate(PHASE_COMMIT, token=self.token)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
|
||||
|
||||
def test_create_pr_phase_permits_continuation(self):
|
||||
result = gate(PHASE_CREATE_PR, token=self.token)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
|
||||
def test_push_phase_permits_continuation(self):
|
||||
result = gate(PHASE_PUSH, token=self.token)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
|
||||
def test_lock_phase_permits_continuation(self):
|
||||
result = gate(PHASE_LOCK, token=self.token)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_all_phases_agree(self):
|
||||
outcomes = {
|
||||
phase: gate(phase, token=self.token)["block"]
|
||||
for phase in (PHASE_LOCK, PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR)
|
||||
}
|
||||
self.assertEqual(set(outcomes.values()), {False}, outcomes)
|
||||
|
||||
def test_dead_session_recovery_still_permits_continuation(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(recovery_lock())
|
||||
for phase in (PHASE_COMMIT, PHASE_PUSH, PHASE_CREATE_PR):
|
||||
with self.subTest(phase=phase):
|
||||
result = gate(phase, token=token)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["owning_pr_recovery_exempted"])
|
||||
|
||||
|
||||
# ───────────────── the exemption cannot be widened ─────────────────
|
||||
|
||||
|
||||
class TestExemptionCannotBeWidened(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
|
||||
def test_an_open_pr_alone_grants_nothing(self):
|
||||
result = gate(PHASE_COMMIT, token=None)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertFalse(result["owning_pr_recovery_exempted"])
|
||||
|
||||
def test_a_second_pr_is_refused(self):
|
||||
result = gate(
|
||||
PHASE_CREATE_PR,
|
||||
token=self.token,
|
||||
open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertFalse(result["owning_pr_recovery_exempted"])
|
||||
|
||||
def test_a_different_pr_is_refused(self):
|
||||
result = gate(
|
||||
PHASE_COMMIT, token=self.token, open_prs=[owning_pr(number=OTHER_PR)]
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_a_different_branch_is_refused(self):
|
||||
result = gate(
|
||||
PHASE_COMMIT, token=self.token, open_prs=[owning_pr(ref=OTHER_BRANCH)]
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_locked_branch_mismatch_is_refused(self):
|
||||
result = gate(PHASE_COMMIT, token=self.token, locked_branch=OTHER_BRANCH)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_live_pr_head_divergence_is_refused(self):
|
||||
# Force-push or unrelated remote movement after renewal.
|
||||
result = gate(
|
||||
PHASE_COMMIT, token=self.token, open_prs=[owning_pr(sha=OTHER_HEAD)]
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_evidence_for_another_issue_is_refused(self):
|
||||
foreign = gitea_mcp_server._owning_pr_continuation_from_lock(
|
||||
renewal_lock(issue_number=ISSUE + 1)
|
||||
)
|
||||
result = gate(PHASE_COMMIT, token=foreign)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_sequential_tasks_do_not_inherit_continuation(self):
|
||||
# One daemon serves many tasks. A renewal proved for issue N must not
|
||||
# authorize continuation for the next task's issue.
|
||||
prior_task = gitea_mcp_server._owning_pr_continuation_from_lock(
|
||||
renewal_lock(issue_number=ISSUE + 7)
|
||||
)
|
||||
self.assertIsNotNone(prior_task)
|
||||
self.assertTrue(gate(PHASE_COMMIT, token=prior_task)["block"])
|
||||
|
||||
|
||||
# ───────────────── ordinary duplicate prevention is intact ─────────────────
|
||||
|
||||
|
||||
class TestDuplicatePreventionRetained(unittest.TestCase):
|
||||
def test_competing_branch_still_blocks(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
result = gate(
|
||||
PHASE_COMMIT,
|
||||
token=token,
|
||||
open_prs=[],
|
||||
branch_names=[BRANCH, OTHER_BRANCH],
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_unrelated_work_without_a_lock_still_blocks(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(None)
|
||||
self.assertIsNone(token)
|
||||
self.assertTrue(gate(PHASE_COMMIT, token=token)["block"])
|
||||
|
||||
|
||||
# ───────────────── refusals stay structured and auditable ─────────────────
|
||||
|
||||
|
||||
class TestRefusalShapePreserved(unittest.TestCase):
|
||||
def test_blocked_result_keeps_its_audit_fields(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
result = gate(
|
||||
PHASE_COMMIT, token=token, open_prs=[owning_pr(number=OTHER_PR)]
|
||||
)
|
||||
for field in (
|
||||
"block",
|
||||
"outcome",
|
||||
"reasons",
|
||||
"owning_pr_recovery_exempted",
|
||||
"owning_pr_recovery_notes",
|
||||
):
|
||||
with self.subTest(field=field):
|
||||
self.assertIn(field, result)
|
||||
self.assertTrue(result["reasons"])
|
||||
# A rejected token explains which element of ownership disagreed.
|
||||
self.assertTrue(result["owning_pr_recovery_notes"])
|
||||
|
||||
def test_granted_result_records_why(self):
|
||||
token = gitea_mcp_server._owning_pr_continuation_from_lock(renewal_lock())
|
||||
result = gate(PHASE_COMMIT, token=token)
|
||||
self.assertTrue(result["owning_pr_recovery_notes"])
|
||||
self.assertIn(
|
||||
f"#{OWNING_PR}", " ".join(result["owning_pr_recovery_notes"])
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user