Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 102 Pull Requests 10 Actions Packages Projects Releases Wiki Activity
Files
c70675a9554b031c20cba514876358d566fc9605
Gitea-Tools/tests/test_issue_953_bootstrap_lock_contract.py
T
sysadminandClaude Opus 4.8 b4c9f55890 test(author): execute the native #953 tool functions and compensation paths (#953) 
Review 632 F4. Every prior reference to `gitea_recover_incomplete_bootstrap_lock`
and `gitea_inspect_issue_lock_contract` under `tests/` was a string literal — a
`tool=` argument, an assertion on returned prose, or a docs substring check —
and the suite reconstructed the recovery sequence by hand from
`assess_bootstrap_lock_recovery`, `build_canonical_issue_lock`,
`build_recovery_record`, and `bind_session_lock`. A hand-written sequence
validates the decision layer but cannot see a divergence between itself and
the tool body, which is exactly how F1 and F2 — both call-site defects —
survived 61 passing cases.

The new cases drive the registered functions against a real `git init`
repository, a real durable lock file, and config-backed profiles:

* `NamespaceMutationWallOnRecovery` — the gate is reached with this task and
  `author_role_exclusive=True`; its return value aborts the tool rather than
  being computed and discarded; the author namespace succeeds and produces a
  canonical lock; a reviewer namespace is refused with `namespace_block` even
  when the claimant data would otherwise match, and emits the standard BLOCKED
  audit record; merger is refused; a mismatched claimant profile is refused by
  the exact-owner layer with the namespace wall explicitly clear; a mismatched
  head is refused; every refusal leaves the lock bytes, generation, branch,
  worktree, and an unrelated lock untouched. A subtest matrix asserts the
  role-kind wall admits `author` and refuses reviewer, merger, limited, and
  mixed.
* `InspectionToolExecutes` — the registered read-only tool reports the contract
  and the recovery preview while leaving lock bytes, mtime, HEAD, and porcelain
  status unchanged, and reports an absent lock without creating one.
* `Ac7PostCompensationGuidance` — drives the real bootstrap to its AC7 refusal
  with a forced partial lock and asserts the returned action against the state
  the rollback actually left: complete cleanup directs to a bootstrap retry and
  that retry is then executed and succeeds, leaving exactly one canonical lock
  and one branch; partial cleanup with a surviving lock, and with a surviving
  branch and worktree, each get their own executable action; a rollback that
  never completed is distinguished from both; no recommendation names a deleted
  artifact; unrelated locks are byte-identical afterwards. Two cases cover
  `release_session_lock` directly — that the rollback now really removes the
  lock, and that it refuses a lock owned by another session.
* `NativeEndToEndBootstrapToCreatePr` — bootstrap, inspect, heartbeat,
  legitimate divergence (commit and push), pre-mutation ownership re-check, and
  the unchanged #447 create-PR provenance guard, in one sequence against a real
  origin. `gitea_lock_issue` is patched to fail the test if anything reaches for
  it, so the bootstrap lock is proved to carry the whole cycle unrepaired.
* `DeadProvenanceConstantRemoved` — the removed constant stays removed and the
  sanctioned source set stays unwidened.

`_NativeToolBase` clears `role_session_router` route state per test: the sticky
reviewer-stop marker is process-global and, now that this task is registered in
`AUTHOR_TASKS`, an earlier suite leaving it set would make the first gate refuse
before the namespace gate under test is reached.

Also documents both new tools in `docs/mcp-tool-inventory.md`, so this branch
adds no drift to `test_documented_inventory_equals_registered_tools`; the
failure reason there is now identical to the pinned base's.

Suite: 61 -> 92 cases.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-28 02:08:21 -04:00

1976 lines
80 KiB
Python
Raw Blame History

"""One canonical bootstrap/lock contract and its recovery path (#953).
Covers the defect in which ``gitea_bootstrap_author_issue_worktree`` reported a
lock as created, wrote a shape no downstream reader accepts, and then directed
the author to implement — after which heartbeat, re-lock, exact-owner renewal,
and the #447 create-PR guard all refuse simultaneously and no sanctioned
recovery path remains eligible.
Every fixture here is synthetic and isolated: locks are written into temporary
directories and the git repositories are created per-test with ``git init``.
The #949-shaped regression reproduces that lock *shape*; it never touches the
real issue #949 branch, worktree, lock, issue, or head.
"""
import contextlib
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest import mock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent))
from mutation_profile_fixture import shared_mutation_env # noqa: E402
import author_issue_bootstrap # noqa: E402
import author_lock_contract # noqa: E402
import bootstrap_lock_recovery # noqa: E402
import gitea_audit # noqa: E402
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import mcp_server # noqa: E402
import role_namespace_gate # noqa: E402
import role_session_router # noqa: E402
ISSUE = 9530
BRANCH = f"fix/issue-{ISSUE}-canonical-contract"
IDENTITY = "example-author-user"
PROFILE = "example-author-profile"
FOREIGN_IDENTITY = "example-other-user"
FOREIGN_PROFILE = "example-other-profile"
REMOTE = "prgs"
ORG = "ExampleOrg"
REPO = "ExampleRepo"
HEAD = "a" * 40
OTHER_HEAD = "b" * 40
def _ts(delta_minutes: int = 0) -> str:
return (
(datetime.now(timezone.utc) + timedelta(minutes=delta_minutes))
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z")
)
def bootstrap_shaped_lock(**overrides):
"""The exact malformed shape #949 was left in by the old bootstrap.
Claimant at the lock top level, ``lease_id`` null, and no ``work_lease``,
``lock_provenance``, or ``expires_at``.
"""
lock = {
"remote": REMOTE,
"org": ORG,
"repo": REPO,
"issue_number": ISSUE,
"branch": BRANCH,
"branch_name": BRANCH,
"worktree_path": "/scratch/wt-9530",
"owner_session": "author_issue_work-deadbeefdeadbeef",
"claimant": {"username": IDENTITY, "profile": PROFILE},
"assignment_id": None,
"lease_id": None,
"expected_base_sha": OTHER_HEAD,
"lock_generation": 1,
}
lock.update(overrides)
return lock
def canonical_lock(worktree="/scratch/wt-9530", **overrides):
lock = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=worktree,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_lock_issue",
)
lock["lock_generation"] = 2
lock["session_pid"] = os.getpid()
lock.update(overrides)
return lock
class CanonicalContractShape(unittest.TestCase):
"""AC1, AC6, AC13: one contract, emitted with every required field."""
def test_bootstrap_builder_emits_the_full_canonical_schema(self):
lock = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_bootstrap_author_issue_worktree",
source=author_lock_contract.SOURCE_BOOTSTRAP,
)
for field in author_lock_contract.REQUIRED_LOCK_FIELDS:
self.assertIn(field, lock, f"canonical lock missing {field}")
for field in author_lock_contract.REQUIRED_WORK_LEASE_FIELDS:
self.assertIn(field, lock["work_lease"], f"work_lease missing {field}")
self.assertEqual(
lock["work_lease"]["claimant"],
{"username": IDENTITY, "profile": PROFILE},
)
self.assertEqual(
lock["lock_provenance"]["written_by_tool"],
"gitea_bootstrap_author_issue_worktree",
)
def test_bootstrap_and_lock_issue_produce_the_same_contract(self):
"""AC13: the two writers must not disagree about what a lock is."""
from_bootstrap = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_bootstrap_author_issue_worktree",
source=author_lock_contract.SOURCE_BOOTSTRAP,
)
from_lock_issue = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_lock_issue",
)
self.assertEqual(sorted(from_bootstrap.keys()), sorted(from_lock_issue.keys()))
self.assertEqual(
sorted(from_bootstrap["work_lease"].keys()),
sorted(from_lock_issue["work_lease"].keys()),
)
for lock in (from_bootstrap, from_lock_issue):
self.assertTrue(
author_lock_contract.assess_lock_contract(lock)["canonical"]
)
def test_no_successful_build_returns_a_null_ownership_token(self):
"""AC6: the fencing token every later check keys on is never null."""
lock = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_bootstrap_author_issue_worktree",
)
self.assertTrue(lock["work_lease"]["task_session_id"])
self.assertIsNotNone(lock["work_lease"]["expires_at"])
def test_provenance_cannot_be_supplied_by_a_caller(self):
"""Safety: provenance is minted server-side, never accepted."""
import inspect as _inspect
params = _inspect.signature(
author_lock_contract.build_canonical_issue_lock
).parameters
self.assertNotIn("lock_provenance", params)
self.assertNotIn("provenance", params)
class MalformedAndPartialLocks(unittest.TestCase):
"""AC7, AC12, AC19: malformed, partial, and missing-expiration locks."""
def test_bootstrap_shaped_lock_is_reported_as_not_canonical(self):
assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock())
self.assertFalse(assessment["canonical"])
self.assertIn("work_lease", assessment["missing_fields"])
self.assertIn("lock_provenance", assessment["missing_fields"])
def test_missing_fields_are_reported_structurally_and_by_name(self):
"""AC7: the refusal names what is missing, not just that it failed."""
assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock())
message = author_lock_contract.format_contract_refusal(assessment)
self.assertIn("work_lease", message)
self.assertIn("lock_provenance", message)
self.assertIsInstance(assessment["missing_fields"], list)
def test_missing_expiration_is_explicit_not_never_expiring(self):
"""AC12: the bug — absent expiry read as 'not yet expired'."""
lock = bootstrap_shaped_lock()
# The pre-existing reader still reports "not expired" for this lock...
self.assertFalse(issue_lock_store.is_lease_expired(lock))
# ...so the contract states the real fact explicitly instead.
state = author_lock_contract.expiration_state(lock)
self.assertEqual(state["state"], author_lock_contract.EXPIRATION_MISSING)
self.assertIsNone(state["expired"])
assessment = author_lock_contract.assess_lock_contract(lock)
self.assertTrue(
any("neither expirable nor renewable" in r for r in assessment["reasons"])
)
def test_missing_expiration_lock_is_recoverable_rather_than_stranded(self):
"""AC12: it must not be non-expiring *and* ineligible for every path."""
assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery(
bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
observed_head=HEAD,
declared_head=HEAD,
worktree_exists=True,
worktree_registered=True,
current_branch=BRANCH,
)
self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"])
def test_partial_lock_missing_only_provenance_is_not_canonical(self):
lock = canonical_lock()
lock.pop("lock_provenance")
assessment = author_lock_contract.assess_lock_contract(lock)
self.assertFalse(assessment["canonical"])
self.assertFalse(assessment["create_pr_eligible"])
def test_unparseable_expiration_is_named_rather_than_silently_ignored(self):
lock = canonical_lock()
lock["work_lease"]["expires_at"] = "not-a-timestamp"
state = author_lock_contract.expiration_state(lock)
self.assertEqual(state["state"], author_lock_contract.EXPIRATION_UNPARSEABLE)
def test_expired_lock_is_reported_as_expired(self):
lock = canonical_lock()
lock["work_lease"]["expires_at"] = _ts(-60)
state = author_lock_contract.expiration_state(lock)
self.assertEqual(state["state"], author_lock_contract.EXPIRATION_RECORDED)
self.assertTrue(state["expired"])
def test_absent_lock_reports_absent_contract(self):
assessment = author_lock_contract.assess_lock_contract(None)
self.assertEqual(assessment["contract"], author_lock_contract.CONTRACT_ABSENT)
self.assertIn(
"gitea_lock_issue", author_lock_contract.recommended_action(assessment)
)
class ClaimantCompatibility(unittest.TestCase):
"""AC2, AC13, AC14: legacy and canonical claimant placement."""
def test_claimant_is_read_from_the_legacy_top_level_placement(self):
recorded = author_lock_contract.lock_claimant(bootstrap_shaped_lock())
self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE})
def test_claimant_is_read_from_the_canonical_work_lease_placement(self):
recorded = author_lock_contract.lock_claimant(canonical_lock())
self.assertEqual(recorded, {"username": IDENTITY, "profile": PROFILE})
def test_work_lease_placement_wins_over_a_stale_top_level_copy(self):
"""An upgraded lock must not be re-read from its stale legacy copy."""
lock = canonical_lock()
lock["claimant"] = {"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE}
self.assertEqual(
author_lock_contract.lock_claimant(lock),
{"username": IDENTITY, "profile": PROFILE},
)
def test_ownership_check_accepts_the_legacy_placement(self):
"""AC2: the exact refusal that made a fresh bootstrap lock un-heartbeatable."""
refusals = issue_lock_store._ownership_refusals(
bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
identity=IDENTITY,
profile=PROFILE,
)
self.assertNotIn(
"lock does not record both a claimant username and profile", refusals
)
self.assertEqual(refusals, [])
def test_ownership_check_still_refuses_a_mismatched_claimant(self):
"""Tolerating the placement must not tolerate the wrong owner."""
refusals = issue_lock_store._ownership_refusals(
bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
identity=FOREIGN_IDENTITY,
profile=PROFILE,
)
self.assertTrue(any("does not match active identity" in r for r in refusals))
def test_ownership_check_still_refuses_a_lock_with_no_claimant_at_all(self):
lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")
lock.pop("claimant")
refusals = issue_lock_store._ownership_refusals(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
identity=IDENTITY,
profile=PROFILE,
)
self.assertIn(
"lock does not record both a claimant username and profile", refusals
)
class CreatePrProvenanceGuardPreserved(unittest.TestCase):
"""AC4 and the safety requirement that #447 is not weakened."""
def test_canonical_bootstrap_lock_passes_the_447_guard(self):
lock = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_bootstrap_author_issue_worktree",
source=author_lock_contract.SOURCE_BOOTSTRAP,
)
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock)
self.assertTrue(verdict["proven"], verdict["reasons"])
def test_the_old_bootstrap_shape_is_still_rejected_by_the_447_guard(self):
"""The guard must keep failing closed on a lock with no provenance."""
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(
bootstrap_shaped_lock()
)
self.assertFalse(verdict["proven"])
self.assertTrue(verdict["block"])
def test_guard_still_rejects_an_unsanctioned_provenance_source(self):
lock = canonical_lock()
lock["lock_provenance"]["source"] = "hand_written_by_caller"
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock)
self.assertFalse(verdict["proven"])
def test_guard_still_rejects_provenance_without_work_lease(self):
lock = canonical_lock()
lock.pop("work_lease")
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(lock)
self.assertFalse(verdict["proven"])
def test_sanctioned_source_set_was_not_widened(self):
"""Bootstrap satisfies the guard; it does not get its own exemption."""
self.assertEqual(
author_lock_contract.SOURCE_BOOTSTRAP,
issue_lock_provenance.SOURCE_LOCK_ISSUE,
)
class RecoveryOwnershipVerification(unittest.TestCase):
"""AC10, AC11: what recovery proves before it changes lock state."""
def _assess(self, lock=None, **overrides):
kwargs = dict(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
observed_head=HEAD,
declared_head=HEAD,
worktree_exists=True,
worktree_registered=True,
current_branch=BRANCH,
)
kwargs.update(overrides)
target = (
lock
if lock is not None
else bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")
)
return bootstrap_lock_recovery.assess_bootstrap_lock_recovery(target, **kwargs)
def test_exact_owner_recovery_is_sanctioned(self):
self.assertTrue(self._assess()["recovery_sanctioned"])
def test_mismatched_repository_is_refused(self):
result = self._assess(repo="OtherRepo")
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(
result["refusal_code"], bootstrap_lock_recovery.REFUSAL_BINDING_MISMATCH
)
def test_mismatched_org_is_refused(self):
self.assertFalse(self._assess(org="OtherOrg")["recovery_sanctioned"])
def test_mismatched_remote_is_refused(self):
self.assertFalse(self._assess(remote="dadeschools")["recovery_sanctioned"])
def test_mismatched_issue_is_refused(self):
self.assertFalse(self._assess(issue_number=ISSUE + 1)["recovery_sanctioned"])
def test_mismatched_branch_is_refused(self):
self.assertFalse(
self._assess(branch_name="fix/issue-9530-other")["recovery_sanctioned"]
)
def test_mismatched_worktree_is_refused(self):
self.assertFalse(
self._assess(worktree_path="/scratch/elsewhere")["recovery_sanctioned"]
)
def test_missing_worktree_is_refused(self):
result = self._assess(worktree_exists=False)
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(
result["refusal_code"], bootstrap_lock_recovery.REFUSAL_WORKTREE_INVALID
)
def test_unregistered_worktree_is_refused(self):
self.assertFalse(self._assess(worktree_registered=False)["recovery_sanctioned"])
def test_worktree_on_a_different_branch_is_refused(self):
self.assertFalse(self._assess(current_branch="master")["recovery_sanctioned"])
def test_head_mismatch_is_refused(self):
result = self._assess(declared_head=OTHER_HEAD)
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(
result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEAD_MISMATCH
)
def test_unresolvable_identity_is_refused(self):
self.assertFalse(self._assess(identity="")["recovery_sanctioned"])
def test_unresolvable_profile_is_refused(self):
self.assertFalse(self._assess(profile="")["recovery_sanctioned"])
def test_matching_username_alone_does_not_prove_ownership(self):
"""Safety: a matching username with the wrong profile is still foreign."""
self.assertFalse(self._assess(profile=FOREIGN_PROFILE)["recovery_sanctioned"])
def test_healthy_foreign_owned_lock_cannot_be_recovered(self):
"""AC11: the foreign-takeover refusal."""
foreign = canonical_lock(worktree="/scratch/wt-9530")
foreign["work_lease"]["claimant"] = {
"username": FOREIGN_IDENTITY,
"profile": FOREIGN_PROFILE,
}
foreign["session_pid"] = os.getpid() # alive → healthy
result = self._assess(lock=foreign)
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(
result["refusal_code"], bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN
)
def test_foreign_owned_incomplete_lock_is_also_refused(self):
"""A foreign lock is refused whether or not it is healthy."""
foreign = bootstrap_shaped_lock(
worktree_path="/scratch/wt-9530",
claimant={"username": FOREIGN_IDENTITY, "profile": FOREIGN_PROFILE},
)
result = self._assess(lock=foreign)
self.assertFalse(result["recovery_sanctioned"])
self.assertIn(
result["refusal_code"],
{
bootstrap_lock_recovery.REFUSAL_FOREIGN_CLAIMANT,
bootstrap_lock_recovery.REFUSAL_HEALTHY_FOREIGN,
},
)
def test_healthy_same_owner_canonical_lock_is_left_alone(self):
"""Nothing to recover: rewriting would invalidate a live heartbeat token."""
result = self._assess(lock=canonical_lock(worktree="/scratch/wt-9530"))
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(
result["refusal_code"], bootstrap_lock_recovery.REFUSAL_ALREADY_CANONICAL
)
def test_absent_lock_is_refused(self):
result = self._assess(lock={})
self.assertFalse(result["recovery_sanctioned"])
self.assertEqual(result["refusal_code"], bootstrap_lock_recovery.REFUSAL_NO_LOCK)
def test_recovery_never_requires_base_equivalence(self):
"""AC9: the branch carries commits; that must not be disqualifying."""
import inspect as _inspect
params = _inspect.signature(
bootstrap_lock_recovery.assess_bootstrap_lock_recovery
).parameters
self.assertNotIn("base_equivalent", params)
self.assertNotIn("expected_base_sha", params)
def test_recovery_accepts_no_caller_supplied_authorization(self):
"""Safety: no caller-manufactured provenance or authorization."""
import inspect as _inspect
params = _inspect.signature(
bootstrap_lock_recovery.assess_bootstrap_lock_recovery
).parameters
for forbidden in (
"recovery_sanctioned",
"lock_provenance",
"provenance",
"operator_override",
"authorized",
):
self.assertNotIn(forbidden, params)
class RecoveryAuditTrail(unittest.TestCase):
"""AC10: auditable ownership and generation transition."""
def _assessment(self, **overrides):
lock = bootstrap_shaped_lock(worktree_path="/scratch/wt-9530", **overrides)
return bootstrap_lock_recovery.assess_bootstrap_lock_recovery(
lock,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
observed_head=HEAD,
declared_head=HEAD,
worktree_exists=True,
worktree_registered=True,
current_branch=BRANCH,
)
def test_recovery_record_preserves_both_sides_of_the_transition(self):
record = bootstrap_lock_recovery.build_recovery_record(
self._assessment(), recovered_at=_ts(), new_task_session_id="task-new"
)
self.assertEqual(record["recovery_kind"], "incomplete_bootstrap_lock")
self.assertEqual(record["prior_generation"], 1)
self.assertEqual(
record["prior_owner_session"], "author_issue_work-deadbeefdeadbeef"
)
self.assertEqual(record["replacement_task_session_id"], "task-new")
self.assertEqual(record["preserved_head"], HEAD)
self.assertFalse(record["branch_reset"])
self.assertFalse(record["base_equivalence_required"])
self.assertIn("work_lease", record["prior_missing_fields"])
def test_expected_generation_is_reported_for_compare_and_swap(self):
self.assertEqual(
self._assessment(lock_generation=7)["expected_generation"], 7
)
class RecoveryIsolationAndPersistence(unittest.TestCase):
"""AC8, AC16: recovery upgrades only its target and inspection mutates nothing."""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.lock_dir = self.tmp.name
def _write(self, lock, **kwargs):
return issue_lock_store.bind_session_lock(
lock, lock_dir=self.lock_dir, **kwargs
)
def test_recovery_upgrades_the_target_lock_to_canonical(self):
path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"))
before = issue_lock_store.read_lock_file(path)
self.assertFalse(author_lock_contract.assess_lock_contract(before)["canonical"])
self._write(
author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_recover_incomplete_bootstrap_lock",
)
)
after = issue_lock_store.read_lock_file(path)
self.assertTrue(author_lock_contract.assess_lock_contract(after)["canonical"])
self.assertGreater(after["lock_generation"], before["lock_generation"])
def test_recovery_does_not_touch_an_unrelated_lock(self):
"""A failure or success must affect only the exact target lock."""
other_issue = ISSUE + 77
other_path = self._write(
bootstrap_shaped_lock(
issue_number=other_issue,
branch_name=f"fix/issue-{other_issue}-unrelated",
worktree_path="/scratch/wt-other",
)
)
other_before = issue_lock_store.read_lock_file(other_path)
target_path = self._write(
bootstrap_shaped_lock(worktree_path="/scratch/wt-9530")
)
self._write(
author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_recover_incomplete_bootstrap_lock",
)
)
self.assertNotEqual(target_path, other_path)
self.assertEqual(issue_lock_store.read_lock_file(other_path), other_before)
def test_inspection_performs_no_mutation(self):
"""AC16: assessing a lock must not rewrite it."""
path = self._write(bootstrap_shaped_lock(worktree_path="/scratch/wt-9530"))
before = issue_lock_store.read_lock_file(path)
mtime_before = os.path.getmtime(path)
author_lock_contract.assess_lock_contract(before)
bootstrap_lock_recovery.assess_bootstrap_lock_recovery(
before,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path="/scratch/wt-9530",
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
observed_head=HEAD,
declared_head=HEAD,
worktree_exists=True,
worktree_registered=True,
current_branch=BRANCH,
)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
self.assertEqual(os.path.getmtime(path), mtime_before)
class HeartbeatOnFreshAndRecoveredLocks(unittest.TestCase):
"""AC2, AC3: heartbeat and renewal against real durable locks."""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.lock_dir = self.tmp.name
self.worktree = os.path.join(self.tmp.name, "wt")
os.makedirs(self.worktree, exist_ok=True)
def _canonical(self, tool="gitea_bootstrap_author_issue_worktree"):
return author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool=tool,
)
def _heartbeat(self, token):
return issue_lock_store.heartbeat_session_lock(
remote=REMOTE,
org=ORG,
repo=REPO,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.worktree,
identity=IDENTITY,
profile=PROFILE,
task_session_id=token,
lock_dir=self.lock_dir,
)
def test_a_freshly_built_canonical_lock_can_be_heartbeated_immediately(self):
"""AC2: the property a bootstrap lock never had."""
lock = self._canonical()
issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir)
outcome = self._heartbeat(lock["work_lease"]["task_session_id"])
self.assertTrue(outcome["success"], outcome.get("reasons"))
self.assertTrue(outcome["performed"])
def test_heartbeat_does_not_change_ownership_or_create_a_second_lock(self):
lock = self._canonical()
path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir)
self._heartbeat(lock["work_lease"]["task_session_id"])
after = issue_lock_store.read_lock_file(path)
self.assertEqual(
author_lock_contract.lock_claimant(after),
{"username": IDENTITY, "profile": PROFILE},
)
# Count issue locks only: bind_session_lock also writes a
# session-<pid>.json pointer, which is pre-existing behaviour and not a
# second claim on the issue.
locks = [
f
for f in os.listdir(self.lock_dir)
if f.endswith(".json") and not f.startswith("session-")
]
self.assertEqual(len(locks), 1, f"expected exactly one issue lock, got {locks}")
def test_heartbeat_refuses_a_foreign_task_session_token(self):
lock = self._canonical(tool="gitea_lock_issue")
issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir)
outcome = self._heartbeat("author_issue_work-someoneelse")
self.assertFalse(outcome["success"])
class BootstrapToCreatePrRegression(unittest.TestCase):
"""AC17, AC18: the exact #949 sequence, against isolated fixtures only.
This reproduces the *shape* of the #949 failure — bootstrap, implement,
commit, push, create PR — in a throwaway git repository. It never reads or
writes the real issue #949 branch, worktree, lock, issue, or head.
"""
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.lock_dir = os.path.join(self.tmp.name, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.origin = os.path.join(self.tmp.name, "origin.git")
self.repo = os.path.join(self.tmp.name, "repo")
subprocess.run(
["git", "init", "--bare", self.origin], check=True, capture_output=True
)
subprocess.run(["git", "init", self.repo], check=True, capture_output=True)
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Example Author")
with open(os.path.join(self.repo, "README.md"), "w") as handle:
handle.write("base\n")
self._git("add", "README.md")
self._git("commit", "-m", "base commit")
self._git("branch", "-M", "master")
self._git("remote", "add", "origin", self.origin)
self._git("push", "-u", "origin", "master")
def _git(self, *args):
return subprocess.run(
["git", "-C", self.repo, *args], check=True, capture_output=True, text=True
)
def _heartbeat(self, token):
return issue_lock_store.heartbeat_session_lock(
remote=REMOTE,
org=ORG,
repo=REPO,
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.repo,
identity=IDENTITY,
profile=PROFILE,
task_session_id=token,
lock_dir=self.lock_dir,
)
def _implement_and_commit(self):
with open(os.path.join(self.repo, "feature.py"), "w") as handle:
handle.write("VALUE = 1\n")
self._git("add", "feature.py")
self._git("commit", "-m", "feat: implement the issue")
def test_bootstrap_implement_commit_push_create_pr_completes(self):
# 1. Bootstrap: branch, worktree, and a canonical lock.
self._git("checkout", "-b", BRANCH)
lock = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.repo,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_bootstrap_author_issue_worktree",
source=author_lock_contract.SOURCE_BOOTSTRAP,
)
path = issue_lock_store.bind_session_lock(lock, lock_dir=self.lock_dir)
token = lock["work_lease"]["task_session_id"]
# The lock is usable the moment bootstrap returns.
self.assertTrue(
author_lock_contract.assess_lock_contract(
issue_lock_store.read_lock_file(path)
)["canonical"]
)
# 2. Renewal is available *before* the branch diverges (AC3).
before = self._heartbeat(token)
self.assertTrue(before["success"], before.get("reasons"))
# 3. Implement and commit — the branch now carries real work.
self._implement_and_commit()
head = self._git("rev-parse", "HEAD").stdout.strip()
# 4. Push.
self._git("push", "-u", "origin", BRANCH)
remote_head = subprocess.run(
["git", "-C", self.origin, "rev-parse", f"refs/heads/{BRANCH}"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
self.assertEqual(head, remote_head)
# 5. Renewal still available *after* commits (AC3) — no base-equivalence.
after = self._heartbeat(token)
self.assertTrue(after["success"], after.get("reasons"))
# 6. create_pr's #447 provenance guard accepts the lock (AC4).
final = issue_lock_store.read_lock_file(path)
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final)
self.assertTrue(verdict["proven"], verdict["reasons"])
# AC18: completed with no manual lock edit, no branch rewind, no
# fallback transport. The base commit is still an ancestor of head.
merge_base = self._git("merge-base", "master", BRANCH).stdout.strip()
master_head = self._git("rev-parse", "master").stdout.strip()
self.assertEqual(merge_base, master_head)
def test_the_old_bootstrap_shape_reproduces_the_949_dead_end(self):
"""The regression must actually fail without the fix."""
self._git("checkout", "-b", BRANCH)
self._implement_and_commit()
legacy = bootstrap_shaped_lock(worktree_path=self.repo)
issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir)
# create_pr refuses — the #447 guard, unchanged.
self.assertFalse(
issue_lock_provenance.assess_lock_file_for_create_pr(legacy)["proven"]
)
# And it is never classified as expired, so renewal never engages.
self.assertFalse(issue_lock_store.is_lease_expired(legacy))
self.assertEqual(
author_lock_contract.expiration_state(legacy)["state"],
author_lock_contract.EXPIRATION_MISSING,
)
def test_recovery_of_a_committed_branch_preserves_the_commits(self):
"""AC9: recovery must not rewind a branch that carries pushed work."""
self._git("checkout", "-b", BRANCH)
self._implement_and_commit()
self._git("push", "-u", "origin", BRANCH)
head_before = self._git("rev-parse", "HEAD").stdout.strip()
legacy = bootstrap_shaped_lock(worktree_path=self.repo)
path = issue_lock_store.bind_session_lock(legacy, lock_dir=self.lock_dir)
assessment = bootstrap_lock_recovery.assess_bootstrap_lock_recovery(
issue_lock_store.read_lock_file(path),
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.repo,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
observed_head=head_before,
declared_head=head_before,
worktree_exists=True,
worktree_registered=True,
current_branch=BRANCH,
)
self.assertTrue(assessment["recovery_sanctioned"], assessment["reasons"])
recovered = author_lock_contract.build_canonical_issue_lock(
issue_number=ISSUE,
branch_name=BRANCH,
worktree_path=self.repo,
remote=REMOTE,
org=ORG,
repo=REPO,
identity=IDENTITY,
profile=PROFILE,
tool="gitea_recover_incomplete_bootstrap_lock",
)
recovered["bootstrap_lock_recovery"] = (
bootstrap_lock_recovery.build_recovery_record(
assessment,
recovered_at=_ts(),
new_task_session_id=recovered["work_lease"]["task_session_id"],
)
)
issue_lock_store.bind_session_lock(
recovered,
lock_dir=self.lock_dir,
expected_generation=assessment["expected_generation"],
recovery_sanctioned=True,
)
# The branch head is untouched and the lock is now canonical.
self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before)
final = issue_lock_store.read_lock_file(path)
self.assertTrue(author_lock_contract.assess_lock_contract(final)["canonical"])
self.assertTrue(
issue_lock_provenance.assess_lock_file_for_create_pr(final)["proven"]
)
self.assertEqual(
final["bootstrap_lock_recovery"]["preserved_head"], head_before
)
self.assertFalse(final["bootstrap_lock_recovery"]["branch_reset"])
class BootstrapWiring(unittest.TestCase):
"""AC1, AC5, AC7, AC15: what the bootstrap tool itself now does."""
def _source(self):
import author_issue_bootstrap
with open(author_issue_bootstrap.__file__) as handle:
return handle.read()
def test_bootstrap_builds_through_the_canonical_contract(self):
source = self._source()
self.assertIn("author_lock_contract.build_canonical_issue_lock", source)
self.assertIn("author_lock_contract.assess_lock_contract", source)
def test_bootstrap_fails_closed_on_an_incomplete_written_lock(self):
"""AC7: partial lock creation stops before implementation begins."""
source = self._source()
self.assertIn("incomplete_issue_lock_contract", source)
self.assertIn('"implementation_allowed": False', source)
def test_next_action_for_a_canonical_lock_directs_to_implementation(self):
"""AC5: executable under the state actually returned."""
assessment = author_lock_contract.assess_lock_contract(canonical_lock())
self.assertIn("work_issue", author_lock_contract.recommended_action(assessment))
def test_next_action_for_an_incomplete_lock_forbids_implementation(self):
"""AC5, AC15: never direct an author into the unrecoverable state."""
assessment = author_lock_contract.assess_lock_contract(bootstrap_shaped_lock())
action = author_lock_contract.recommended_action(assessment)
self.assertIn("Do not begin implementation", action)
self.assertIn("gitea_recover_incomplete_bootstrap_lock", action)
class _NativeToolBase(unittest.TestCase):
"""A real git worktree, a real durable lock, and the real registered tools.
Review 632 F4: every previous reference to the two new tools under
``tests/`` was a string literal, and the suite reconstructed the recovery
sequence by hand. A hand-written sequence cannot see a divergence between
itself and the tool body — which is exactly how F1 and F2, both call-site
defects, survived 61 passing cases. These call the registered functions.
"""
TOOL_ISSUE = 9531
TOOL_BRANCH = "fix/issue-9531-native-tool-path"
TOOL_ORG = "Example-Org"
TOOL_REPO = "Example-Repo"
AUTHOR_PROFILE = "test-author-prgs"
REVIEWER_PROFILE = "test-reviewer-prgs"
MERGER_PROFILE = "test-merger-prgs"
TOOL_IDENTITY = "example-native-user"
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.repo = tempfile.mkdtemp(prefix="issue953-native-")
self.addCleanup(
lambda: subprocess.run(["rm", "-rf", self.repo], check=False)
)
self._init_repo()
self.remotes = mock.patch.dict(
mcp_server.REMOTES,
{
"prgs": {
"host": "gitea.prgs.cc",
"org": self.TOOL_ORG,
"repo": self.TOOL_REPO,
}
},
)
self.remotes.start()
self.addCleanup(mock.patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
# The sticky reviewer-stop route is process-global and outlives whatever
# test set it. These cases assert on the *namespace* gate, so the gate
# ahead of it must start clean or it refuses first for another reason.
role_session_router.clear_route_state()
self.addCleanup(role_session_router.clear_route_state)
def _git(self, *args):
return subprocess.run(
["git", "-C", self.repo, *args],
capture_output=True,
text=True,
check=True,
)
def _init_repo(self):
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
with open(os.path.join(self.repo, "seed.txt"), "w") as handle:
handle.write("seed\n")
self._git("add", "seed.txt")
self._git("commit", "-q", "-m", "seed")
self._git("checkout", "-q", "-b", self.TOOL_BRANCH)
# The state recovery exists for: the branch already carries pushed work.
with open(os.path.join(self.repo, "impl.txt"), "w") as handle:
handle.write("implementation\n")
self._git("add", "impl.txt")
self._git("commit", "-q", "-m", "implementation")
self.head = self._git("rev-parse", "HEAD").stdout.strip()
self.worktree = os.path.realpath(self.repo)
def _lock_path(self):
return issue_lock_store.lock_file_path(
remote=REMOTE,
org=self.TOOL_ORG,
repo=self.TOOL_REPO,
issue_number=self.TOOL_ISSUE,
lock_dir=self.lock_dir.name,
)
def write_incomplete_lock(self):
"""The exact malformed shape the old bootstrap left behind."""
lock = bootstrap_shaped_lock(
issue_number=self.TOOL_ISSUE,
branch=self.TOOL_BRANCH,
branch_name=self.TOOL_BRANCH,
worktree_path=self.worktree,
org=self.TOOL_ORG,
repo=self.TOOL_REPO,
claimant={
"username": self.TOOL_IDENTITY,
"profile": self.AUTHOR_PROFILE,
},
)
path = self._lock_path()
lock["lock_file_path"] = path
issue_lock_store.save_lock_file(path, lock)
return path
def _env(self, profile_name):
env = shared_mutation_env(
profile_name,
include_example_repo=True,
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return env
def call_recovery(
self,
*,
profile_name=None,
identity=None,
claimant_profile=None,
expected_head=None,
audit_sink=None,
namespace_patch=None,
**kwargs,
):
"""Invoke the registered recovery tool itself, not its assessors."""
profile_name = profile_name or self.AUTHOR_PROFILE
env = self._env(profile_name)
patches = [
mock.patch(
"mcp_server._work_lease_claimant",
return_value={
"username": identity or self.TOOL_IDENTITY,
"profile": claimant_profile or profile_name,
},
),
mock.patch("mcp_server.get_auth_header", return_value="token x"),
mock.patch(
"mcp_server._canonical_local_git_root", return_value=self.worktree
),
mock.patch.dict(os.environ, env, clear=True),
]
if audit_sink is not None:
patches.append(
mock.patch(
"mcp_server._audit",
side_effect=lambda *a, **kw: audit_sink.append((a, kw)),
)
)
if namespace_patch is not None:
patches.append(namespace_patch)
with contextlib.ExitStack() as stack:
for patch in patches:
stack.enter_context(patch)
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_recover_incomplete_bootstrap_lock(
issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE),
branch_name=kwargs.pop("branch_name", self.TOOL_BRANCH),
worktree_path=kwargs.pop("worktree_path", self.worktree),
expected_head=expected_head or self.head,
remote="prgs",
**kwargs,
)
def call_inspection(self, *, profile_name=None, **kwargs):
"""Invoke the registered read-only inspection tool itself."""
env = self._env(profile_name or self.AUTHOR_PROFILE)
with mock.patch(
"mcp_server._work_lease_claimant",
return_value={
"username": self.TOOL_IDENTITY,
"profile": profile_name or self.AUTHOR_PROFILE,
},
), mock.patch(
"mcp_server.get_auth_header", return_value="token x"
), mock.patch(
"mcp_server._canonical_local_git_root", return_value=self.worktree
), mock.patch.dict(
os.environ, env, clear=True
):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_inspect_issue_lock_contract(
issue_number=kwargs.pop("issue_number", self.TOOL_ISSUE),
remote="prgs",
**kwargs,
)
class NamespaceMutationWallOnRecovery(_NativeToolBase):
"""Review 632 F1: the namespace/session wall on the new author mutation.
``gitea_recover_incomplete_bootstrap_lock`` writes the same durable lock as
``gitea_recover_dirty_orphaned_issue_worktree`` and must carry the same
third gate. Exact-owner claimant comparison inside
``assess_bootstrap_lock_recovery`` is a later layer, not a substitute: it
refuses without a namespace evaluation and without a BLOCKED audit record.
"""
def test_the_recovery_tool_calls_the_namespace_mutation_gate(self):
"""The gate must be reached, with this task, before any write."""
self.write_incomplete_lock()
seen = []
def _spy(task, **kwargs):
seen.append((task, kwargs))
return None
self.call_recovery(
namespace_patch=mock.patch(
"mcp_server._namespace_mutation_block", side_effect=_spy
)
)
self.assertEqual(len(seen), 1, seen)
task, kwargs = seen[0]
self.assertEqual(task, "recover_incomplete_bootstrap_lock")
self.assertTrue(kwargs.get("author_role_exclusive"))
def test_the_gate_return_value_is_consumed_and_returned(self):
"""A gate refusal must abort the tool, not be computed and discarded."""
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
refusal = {"success": False, "performed": False, "namespace_block": True}
result = self.call_recovery(
namespace_patch=mock.patch(
"mcp_server._namespace_mutation_block", return_value=refusal
)
)
self.assertIs(result, refusal)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
def test_correct_author_namespace_succeeds(self):
path = self.write_incomplete_lock()
result = self.call_recovery()
self.assertTrue(result.get("success"), result)
self.assertTrue(result.get("performed"), result)
written = issue_lock_store.read_lock_file(path)
self.assertTrue(
author_lock_contract.assess_lock_contract(written)["canonical"], written
)
def test_reviewer_namespace_is_rejected_with_matching_claimant_data(self):
"""Identity that would satisfy the owner check must not be a way in."""
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
result = self.call_recovery(
profile_name=self.REVIEWER_PROFILE,
claimant_profile=self.AUTHOR_PROFILE,
)
self.assertFalse(result.get("success"), result)
self.assertFalse(result.get("performed"), result)
self.assertTrue(result.get("namespace_block"), result)
self.assertEqual(result.get("mcp_namespace"), "gitea-reviewer")
# Not the later exact-owner refusal — the namespace layer stopped it.
self.assertNotEqual(result.get("refusal_code"), "foreign_claimant")
self.assertEqual(issue_lock_store.read_lock_file(path), before)
def test_reviewer_rejection_emits_the_standard_blocked_audit(self):
self.write_incomplete_lock()
audit = []
self.call_recovery(profile_name=self.REVIEWER_PROFILE, audit_sink=audit)
blocked = [
(args, kwargs)
for args, kwargs in audit
if kwargs.get("result") == gitea_audit.BLOCKED
]
self.assertTrue(blocked, audit)
args, kwargs = blocked[0]
self.assertEqual(args[0], "recover_incomplete_bootstrap_lock")
self.assertEqual(
kwargs.get("mutation_task"), "recover_incomplete_bootstrap_lock"
)
def test_merger_profile_is_rejected(self):
"""gitea.issue.comment is held by every role; the wall cannot rely on it."""
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
result = self.call_recovery(profile_name=self.MERGER_PROFILE)
self.assertFalse(result.get("success"), result)
self.assertFalse(result.get("performed"), result)
# Specifically the namespace/role wall, not some later refusal.
self.assertTrue(result.get("namespace_block"), result)
self.assertTrue(
any(
"recover_incomplete_bootstrap_lock' blocked" in reason
for reason in result.get("reasons") or []
),
result,
)
self.assertIsNone(result.get("refusal_code"), result)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
def test_wrong_profile_for_the_claimant_is_rejected(self):
"""A matching username under a different profile is still foreign."""
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
result = self.call_recovery(claimant_profile="test-author-dadeschools")
self.assertFalse(result.get("success"), result)
self.assertFalse(result.get("performed"), result)
# The namespace wall passes here (the session *is* author-bound); this
# must be the exact-owner layer refusing the mismatched profile.
self.assertIsNone(result.get("namespace_block"), result)
self.assertIn(
result.get("refusal_code"), ("foreign_claimant", "healthy_foreign_lock"), result
)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
def test_mismatched_head_is_rejected_without_mutation(self):
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
result = self.call_recovery(expected_head=OTHER_HEAD)
self.assertFalse(result.get("success"), result)
self.assertFalse(result.get("mutation_performed"), result)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
def test_rejection_leaves_branch_worktree_and_unrelated_locks_untouched(self):
unrelated = issue_lock_store.lock_file_path(
remote=REMOTE,
org=self.TOOL_ORG,
repo=self.TOOL_REPO,
issue_number=self.TOOL_ISSUE + 41,
lock_dir=self.lock_dir.name,
)
issue_lock_store.save_lock_file(
unrelated, bootstrap_shaped_lock(issue_number=self.TOOL_ISSUE + 41)
)
unrelated_before = issue_lock_store.read_lock_file(unrelated)
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
head_before = self._git("rev-parse", "HEAD").stdout.strip()
self.call_recovery(profile_name=self.REVIEWER_PROFILE)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
self.assertEqual(issue_lock_store.read_lock_file(unrelated), unrelated_before)
self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before)
self.assertTrue(os.path.isdir(self.worktree))
self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "")
def test_the_gate_routes_this_task_as_author_required(self):
"""Without a router entry the namespace check silently allows everything."""
self.assertEqual(
role_session_router.required_role_for_task(
"recover_incomplete_bootstrap_lock"
),
"author",
)
ok, reasons = role_namespace_gate.check_author_mutation_namespace(
"recover_incomplete_bootstrap_lock",
{
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.read", "gitea.pr.approve"],
"forbidden_operations": [],
},
)
self.assertFalse(ok, reasons)
def test_role_kind_wall_admits_author_and_refuses_every_other_role(self):
cases = {
"author": (["gitea.pr.create", "gitea.branch.push"], True),
"reviewer": (["gitea.pr.approve"], False),
"merger": (["gitea.pr.merge"], False),
"limited": (["gitea.issue.comment", "gitea.read"], False),
"mixed": (["gitea.pr.approve", "gitea.pr.create"], False),
}
for label, (ops, expected) in cases.items():
with self.subTest(role=label):
ok, _ = role_namespace_gate.check_author_role_kind(
"recover_incomplete_bootstrap_lock",
{
"profile_name": f"prgs-{label}",
"allowed_operations": ops,
"forbidden_operations": [],
},
)
self.assertEqual(ok, expected)
class InspectionToolExecutes(_NativeToolBase):
"""AC16 proved against the registered tool, not only its assessors."""
def test_the_registered_inspection_tool_reports_the_contract(self):
self.write_incomplete_lock()
result = self.call_inspection()
self.assertTrue(result.get("success"), result)
self.assertTrue(result.get("read_only"))
self.assertTrue(result.get("lock_present"))
self.assertFalse(result["lock_contract"]["canonical"])
def test_the_registered_inspection_tool_mutates_nothing(self):
path = self.write_incomplete_lock()
before = issue_lock_store.read_lock_file(path)
mtime_before = os.path.getmtime(path)
head_before = self._git("rev-parse", "HEAD").stdout.strip()
result = self.call_inspection(
branch_name=self.TOOL_BRANCH, worktree_path=self.worktree
)
self.assertFalse(result.get("mutation_performed"))
self.assertFalse(result.get("performed"))
self.assertIn("recovery_preview", result)
self.assertEqual(issue_lock_store.read_lock_file(path), before)
self.assertEqual(os.path.getmtime(path), mtime_before)
self.assertEqual(self._git("rev-parse", "HEAD").stdout.strip(), head_before)
self.assertEqual(self._git("status", "--porcelain").stdout.strip(), "")
def test_inspection_reports_an_absent_lock_without_creating_one(self):
result = self.call_inspection()
self.assertTrue(result.get("success"), result)
self.assertFalse(result.get("lock_present"))
self.assertFalse(os.path.exists(self._lock_path()))
class Ac7PostCompensationGuidance(unittest.TestCase):
"""Review 632 F2: the returned action must fit the post-rollback state.
The AC7 refusal runs ``run_compensating_recovery`` first, which releases the
lock and removes the branch and worktree. Recommending incomplete-lock
recovery for those exact artifacts hands the author ``no_durable_lock`` and
then ``worktree_invalid`` — the unexecutable-guidance failure class #953
exists to remove, reintroduced on the new fail-closed path.
"""
ISSUE_NUMBER = 9532
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.repo = os.path.join(self.tmp.name, "repo")
os.makedirs(self.repo)
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
with open(os.path.join(self.repo, "README.md"), "w") as handle:
handle.write("# seed\n")
self._git("add", "README.md")
self._git("commit", "-q", "-m", "seed")
self.master_sha = self._git("rev-parse", "HEAD").stdout.strip()
os.makedirs(os.path.join(self.repo, "branches"), exist_ok=True)
self.lock_dir = os.path.join(self.tmp.name, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.journal_dir = os.path.join(self.tmp.name, "journals")
os.makedirs(self.journal_dir, exist_ok=True)
self._journal_env = mock.patch.dict(
os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir}
)
self._journal_env.start()
self.addCleanup(self._journal_env.stop)
def _git(self, *args):
return subprocess.run(
["git", "-C", self.repo, *args],
capture_output=True,
text=True,
check=True,
)
def _branch_names(self):
out = subprocess.run(
["git", "-C", self.repo, "branch", "--format=%(refname:short)"],
capture_output=True,
text=True,
check=False,
)
return [line.strip() for line in out.stdout.splitlines() if line.strip()]
def _lock_files(self):
"""Durable issue locks only — not phase journals or session pointers."""
found = []
for path in issue_lock_store.iter_lock_files(self.lock_dir):
record = issue_lock_store.read_lock_file(path) or {}
if "lock_generation" in record:
found.append(os.path.basename(path))
return sorted(found)
def _run_bootstrap(self, *, key, partial=False, extra_patches=()):
"""Drive the real bootstrap; optionally force a partial written lock."""
with contextlib.ExitStack() as stack:
if partial:
real_build = author_lock_contract.build_canonical_issue_lock
def _partial(**kwargs):
record = real_build(**kwargs)
# The #949 shape: claimant hoisted to the top level, no
# work_lease, no provenance, no expiry.
return {
"remote": record["remote"],
"org": record["org"],
"repo": record["repo"],
"issue_number": record["issue_number"],
"branch": record["branch"],
"branch_name": record["branch_name"],
"worktree_path": record["worktree_path"],
"claimant": dict(record["work_lease"]["claimant"]),
"lease_id": None,
"owner_session": record.get("owner_session"),
}
stack.enter_context(
mock.patch.object(
author_issue_bootstrap.author_lock_contract,
"build_canonical_issue_lock",
side_effect=_partial,
)
)
for patch in extra_patches:
stack.enter_context(patch)
return author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=self.ISSUE_NUMBER,
canonical_repo_root=self.repo,
expected_base_sha=self.master_sha,
idempotency_key=key,
remote="prgs",
lock_dir=self.lock_dir,
owner_session=f"session-{key}",
active_identity=IDENTITY,
active_profile=PROFILE,
)
def test_ac7_failure_then_complete_compensation_directs_to_retry_bootstrap(self):
result = self._run_bootstrap(key="ac7-complete", partial=True)
self.assertFalse(result.get("success"), result)
self.assertEqual(result.get("reason_code"), "incomplete_issue_lock_contract")
self.assertFalse(result.get("implementation_allowed"))
state = result["post_compensation_state"]
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE)
self.assertEqual(state["surviving_artifacts"], [])
action = result["exact_next_action"]
self.assertIn("gitea_bootstrap_author_issue_worktree", action)
self.assertIn("Do not call gitea_recover_incomplete_bootstrap_lock", action)
def test_the_returned_retry_action_is_executable(self):
"""Follow the advice literally and it must succeed."""
first = self._run_bootstrap(key="ac7-retry-1", partial=True)
self.assertIn(
"gitea_bootstrap_author_issue_worktree", first["exact_next_action"]
)
second = self._run_bootstrap(key="ac7-retry-2")
self.assertTrue(second.get("success"), second)
self.assertTrue(second.get("implementation_allowed"))
self.assertTrue(second["lock_contract"]["canonical"], second["lock_contract"])
self.assertTrue(second.get("task_session_id"))
def test_retry_after_complete_compensation_leaves_exactly_one_lock(self):
self._run_bootstrap(key="ac7-single-1", partial=True)
self.assertEqual(self._lock_files(), [])
second = self._run_bootstrap(key="ac7-single-2")
self.assertTrue(second.get("success"), second)
locks = self._lock_files()
self.assertEqual(len(locks), 1, locks)
written = issue_lock_store.read_lock_file(second["lock_state"])
self.assertTrue(author_lock_contract.assess_lock_contract(written)["canonical"])
branches = [b for b in self._branch_names() if b != "master"]
self.assertEqual(len(branches), 1, branches)
def test_no_recommendation_names_an_artifact_the_rollback_deleted(self):
result = self._run_bootstrap(key="ac7-no-ghosts", partial=True)
action = result["exact_next_action"]
state = result["post_compensation_state"]
self.assertFalse(state["branch_present"])
self.assertFalse(state["worktree_present"])
self.assertFalse(state["lock_present"])
self.assertNotIn(result["worktree_path"], action)
self.assertNotIn(f"branch '{result['branch_name']}'", action)
self.assertFalse(os.path.isdir(result["worktree_path"]))
self.assertNotIn(result["branch_name"], self._branch_names())
def test_partial_compensation_with_a_surviving_lock_is_not_reported_complete(self):
def _boom(**kwargs):
raise RuntimeError("lock release failed")
outcome = self._run_bootstrap(
key="ac7-lock-survives",
partial=True,
extra_patches=[
mock.patch.object(
author_issue_bootstrap.issue_lock_store,
"release_session_lock",
side_effect=_boom,
)
],
)
state = outcome["post_compensation_state"]
self.assertTrue(state["lock_present"], state)
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL)
self.assertIn("lock", state["surviving_artifacts"])
self.assertTrue(state["failed_rollback_steps"], state)
action = outcome["exact_next_action"]
self.assertIn("failed step", action)
self.assertIn("gitea_inspect_issue_lock_contract", action)
# The advice must not send the author at artifacts the rollback removed.
self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action)
def test_partial_compensation_with_surviving_branch_and_worktree(self):
"""A worktree dirty at rollback time is preserved, and so is its branch."""
real_assess = author_lock_contract.assess_lock_contract
def _dirty_then_report(lock):
verdict = real_assess(lock)
path = (lock or {}).get("worktree_path")
if path and os.path.isdir(path):
with open(os.path.join(path, "uncommitted.txt"), "w") as handle:
handle.write("author bytes\n")
return verdict
outcome = self._run_bootstrap(
key="ac7-wt-survives",
partial=True,
extra_patches=[
mock.patch.object(
author_issue_bootstrap.author_lock_contract,
"assess_lock_contract",
side_effect=_dirty_then_report,
)
],
)
state = outcome["post_compensation_state"]
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL)
self.assertTrue(state["worktree_present"], state)
self.assertTrue(state["branch_present"], state)
self.assertTrue(os.path.isdir(outcome["worktree_path"]))
self.assertIn(outcome["branch_name"], self._branch_names())
self.assertIn("gitea_lock_issue", outcome["exact_next_action"])
self.assertIn(outcome["branch_name"], outcome["exact_next_action"])
def test_compensation_failure_is_distinguished_from_partial_cleanup(self):
outcome = self._run_bootstrap(
key="ac7-comp-failed",
partial=True,
extra_patches=[
mock.patch.object(
author_issue_bootstrap,
"run_compensating_recovery",
return_value={
"executed": False,
"rolled_back": [],
"reason": "boom",
},
)
],
)
state = outcome["post_compensation_state"]
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_FAILED)
action = outcome["exact_next_action"]
self.assertIn("did not complete", action)
self.assertIn("gitea_inspect_issue_lock_contract", action)
self.assertNotIn("re-run gitea_bootstrap_author_issue_worktree", action)
def test_a_failed_rollback_step_is_recorded_even_when_nothing_survives(self):
"""A step that errored is still reported, and cleanup is still complete."""
state = author_lock_contract.assess_post_compensation_state(
{
"executed": True,
"rolled_back": ["lease_release_failed:lease-1:RuntimeError"],
},
lock_present=False,
worktree_present=False,
branch_present=False,
)
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_COMPLETE)
self.assertEqual(
state["failed_rollback_steps"],
["lease_release_failed:lease-1:RuntimeError"],
)
def test_the_compensation_lock_release_actually_removes_the_lock(self):
"""The rollback's lock half was dead code before #953 review 632 F2."""
self.assertTrue(hasattr(issue_lock_store, "release_session_lock"))
result = self._run_bootstrap(key="ac7-release-real", partial=True)
self.assertIn(
f"lock:issue-{self.ISSUE_NUMBER}",
result["compensating_recovery"]["rolled_back"],
)
self.assertEqual(self._lock_files(), [])
def test_release_refuses_a_lock_owned_by_a_different_session(self):
created = self._run_bootstrap(key="ac7-foreign-release")
self.assertTrue(created.get("success"), created)
with self.assertRaises(FileNotFoundError):
issue_lock_store.release_session_lock(
issue_number=self.ISSUE_NUMBER,
session="session-somebody-else",
lock_dir=self.lock_dir,
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertEqual(len(self._lock_files()), 1, self._lock_files())
def test_ac7_refusal_never_reports_implementation_ready(self):
result = self._run_bootstrap(key="ac7-never-ready", partial=True)
self.assertFalse(result.get("success"))
self.assertFalse(result.get("implementation_allowed"))
self.assertNotIn(
"proceed with author implementation", result["exact_next_action"]
)
def test_ac7_refusal_touches_no_unrelated_lock(self):
unrelated_path = issue_lock_store.lock_file_path(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=self.ISSUE_NUMBER + 63,
lock_dir=self.lock_dir,
)
issue_lock_store.save_lock_file(
unrelated_path, bootstrap_shaped_lock(issue_number=self.ISSUE_NUMBER + 63)
)
before = issue_lock_store.read_lock_file(unrelated_path)
mtime_before = os.path.getmtime(unrelated_path)
self._run_bootstrap(key="ac7-isolation", partial=True)
self.assertEqual(issue_lock_store.read_lock_file(unrelated_path), before)
self.assertEqual(os.path.getmtime(unrelated_path), mtime_before)
def test_surviving_lock_and_worktree_direct_to_target_specific_recovery(self):
"""The one state in which incomplete-lock recovery *is* executable."""
state = author_lock_contract.assess_post_compensation_state(
{"executed": True, "rolled_back": []},
lock_present=True,
worktree_present=True,
branch_present=True,
)
action = author_lock_contract.post_compensation_action(
state,
issue_number=self.ISSUE_NUMBER,
branch_name=BRANCH,
worktree_path="/scratch/wt",
missing_fields=["work_lease"],
)
self.assertEqual(state["cleanup_state"], author_lock_contract.CLEANUP_PARTIAL)
self.assertIn("gitea_recover_incomplete_bootstrap_lock", action)
self.assertIn(BRANCH, action)
self.assertIn("/scratch/wt", action)
class NativeEndToEndBootstrapToCreatePr(unittest.TestCase):
"""AC17/AC18 driven through the real tools, with no gitea_lock_issue repair.
bootstrap → inspect → heartbeat/renew → legitimate divergence → downstream
validation → create_pr provenance. The lock the real bootstrap writes must
carry the whole sequence on its own; repairing it with the older
``gitea_lock_issue`` path would prove nothing about the new contract, so
that path is patched to fail the test if anything reaches for it.
"""
ISSUE_NUMBER = 9533
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.origin = os.path.join(self.tmp.name, "origin.git")
self.repo = os.path.join(self.tmp.name, "repo")
subprocess.run(
["git", "init", "-q", "--bare", self.origin], check=True, capture_output=True
)
subprocess.run(
["git", "init", "-q", "-b", "master", self.repo],
check=True,
capture_output=True,
)
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Example Author")
with open(os.path.join(self.repo, "README.md"), "w") as handle:
handle.write("base\n")
self._git("add", "README.md")
self._git("commit", "-q", "-m", "base commit")
self._git("remote", "add", "origin", self.origin)
self._git("push", "-q", "-u", "origin", "master")
self.master_sha = self._git("rev-parse", "HEAD").stdout.strip()
self.lock_dir = os.path.join(self.tmp.name, "locks")
os.makedirs(self.lock_dir, exist_ok=True)
self.journal_dir = os.path.join(self.tmp.name, "journals")
os.makedirs(self.journal_dir, exist_ok=True)
self._journal_env = mock.patch.dict(
os.environ, {"GITEA_BOOTSTRAP_JOURNAL_DIR": self.journal_dir}
)
self._journal_env.start()
self.addCleanup(self._journal_env.stop)
self.remotes = mock.patch.dict(
mcp_server.REMOTES,
{
"prgs": {
"host": "gitea.prgs.cc",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
}
},
)
self.remotes.start()
self.addCleanup(mock.patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
def _git(self, *args, cwd=None):
return subprocess.run(
["git", "-C", cwd or self.repo, *args],
check=True,
capture_output=True,
text=True,
)
def _inspect(self, worktree, branch):
env = shared_mutation_env(
"test-author-prgs", GITEA_ISSUE_LOCK_DIR=self.lock_dir
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
with mock.patch(
"mcp_server._work_lease_claimant",
return_value={"username": IDENTITY, "profile": PROFILE},
), mock.patch(
"mcp_server.get_auth_header", return_value="token x"
), mock.patch(
"mcp_server._canonical_local_git_root", return_value=self.repo
), mock.patch.dict(
os.environ, env, clear=True
):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir
return mcp_server.gitea_inspect_issue_lock_contract(
issue_number=self.ISSUE_NUMBER,
branch_name=branch,
worktree_path=worktree,
remote="prgs",
)
def test_bootstrap_inspect_heartbeat_diverge_and_create_pr_provenance(self):
def _forbidden(*args, **kwargs):
raise AssertionError(
"gitea_lock_issue must not be needed to repair a bootstrap lock"
)
with mock.patch.object(mcp_server, "gitea_lock_issue", side_effect=_forbidden):
# 1. Bootstrap through the real function.
result = author_issue_bootstrap.bootstrap_author_issue_worktree(
issue_number=self.ISSUE_NUMBER,
canonical_repo_root=self.repo,
expected_base_sha=self.master_sha,
idempotency_key="e2e-953",
remote="prgs",
lock_dir=self.lock_dir,
owner_session="session-e2e-953",
active_identity=IDENTITY,
active_profile=PROFILE,
)
self.assertTrue(result.get("success"), result)
self.assertTrue(result.get("implementation_allowed"))
branch = result["branch_name"]
worktree = result["worktree_path"]
token = result["task_session_id"]
self.assertTrue(token)
# 2. Inspect through the registered read-only tool.
inspected = self._inspect(worktree, branch)
self.assertTrue(inspected.get("success"), inspected)
self.assertTrue(inspected["lock_contract"]["canonical"], inspected)
self.assertTrue(inspected["lock_contract"]["heartbeatable"])
self.assertTrue(inspected["lock_contract"]["create_pr_eligible"])
self.assertFalse(inspected.get("mutation_performed"))
# 3. Heartbeat/renew while still base-equivalent.
before = issue_lock_store.heartbeat_session_lock(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=self.ISSUE_NUMBER,
branch_name=branch,
worktree_path=worktree,
identity=IDENTITY,
profile=PROFILE,
task_session_id=token,
lock_dir=self.lock_dir,
)
self.assertTrue(before["success"], before.get("reasons"))
# 4. Legitimate divergence: implement, commit, push.
with open(os.path.join(worktree, "feature.py"), "w") as handle:
handle.write("VALUE = 1\n")
self._git("add", "feature.py", cwd=worktree)
self._git("commit", "-q", "-m", "feat: implement", cwd=worktree)
self._git("push", "-q", "-u", "origin", branch, cwd=worktree)
head = self._git("rev-parse", "HEAD", cwd=worktree).stdout.strip()
remote_head = subprocess.run(
["git", "-C", self.origin, "rev-parse", f"refs/heads/{branch}"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
self.assertEqual(head, remote_head)
# 5. Downstream validation after divergence.
after = issue_lock_store.heartbeat_session_lock(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
issue_number=self.ISSUE_NUMBER,
branch_name=branch,
worktree_path=worktree,
identity=IDENTITY,
profile=PROFILE,
task_session_id=token,
lock_dir=self.lock_dir,
)
self.assertTrue(after["success"], after.get("reasons"))
# The pre-mutation ownership re-check (#438) accepts the diverged
# branch without any repair step.
diverged_lock = issue_lock_store.read_lock_file(result["lock_state"])
proof = issue_lock_store.verify_lock_for_mutation(
diverged_lock,
issue_number=self.ISSUE_NUMBER,
branch_name=branch,
worktree_path=worktree,
)
self.assertTrue(proof["proven"], proof["reasons"])
# 6. The unchanged #447 create_pr provenance guard accepts it.
final = issue_lock_store.read_lock_file(result["lock_state"])
verdict = issue_lock_provenance.assess_lock_file_for_create_pr(final)
self.assertTrue(verdict["proven"], verdict["reasons"])
# The branch was never rewound to satisfy any gate.
merge_base = self._git("merge-base", "master", branch, cwd=worktree).stdout.strip()
self.assertEqual(merge_base, self.master_sha)
class DeadProvenanceConstantRemoved(unittest.TestCase):
"""Review 632 F3: no second lock source may appear to exist."""
def test_no_recovery_source_constant_is_exported(self):
self.assertFalse(
hasattr(author_lock_contract, "SOURCE_BOOTSTRAP_LOCK_RECOVERY")
)
def test_bootstrap_source_is_still_the_sanctioned_lock_issue_source(self):
self.assertEqual(
author_lock_contract.SOURCE_BOOTSTRAP,
issue_lock_provenance.SOURCE_LOCK_ISSUE,
)
def test_the_sanctioned_source_set_is_still_not_widened(self):
self.assertNotIn(
"gitea_recover_incomplete_bootstrap_lock",
issue_lock_provenance.SANCTIONED_LOCK_SOURCES,
)
class CapabilityRegistration(unittest.TestCase):
"""The new operations are registered and role-gated."""
def test_recovery_is_registered_as_an_author_operation(self):
import task_capability_map
self.assertEqual(
task_capability_map.required_permission(
"recover_incomplete_bootstrap_lock"
),
"gitea.issue.comment",
)
self.assertEqual(
task_capability_map.required_role("recover_incomplete_bootstrap_lock"),
"author",
)
def test_inspection_requires_only_read_permission(self):
"""AC16: a read-only surface must not carry a mutating permission."""
import task_capability_map
self.assertEqual(
task_capability_map.required_permission("inspect_issue_lock_contract"),
"gitea.read",
)
class WorkflowDocumentation(unittest.TestCase):
"""AC20: the canonical ordering and recovery path are documented."""
def test_author_workflow_documents_ordering_and_recovery(self):
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
doc = os.path.join(root, "docs", "author-issue-lock-contract.md")
self.assertTrue(os.path.exists(doc), f"missing {doc}")
with open(doc) as handle:
text = handle.read()
self.assertIn("gitea_recover_incomplete_bootstrap_lock", text)
self.assertIn("gitea_lock_issue", text)
self.assertIn("gitea_bootstrap_author_issue_worktree", text)
self.assertIn("before writing any implementation", text.lower())
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink