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]>
This commit is contained in:
Claude Opus 4.8
@@ -12,19 +12,29 @@ 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"
|
||||
@@ -964,6 +974,961 @@ class BootstrapWiring(unittest.TestCase):
|
||||
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."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user