Compare commits

..
1 Commits
Author SHA1 Message Date
jcwalker3andClaude Opus 4.8 95a5eb254f fix(mcp): forward worktree_path into publication preflight (Closes #815)
gitea_publish_unpublished_issue_branch accepted a required worktree_path
but resolved it only after verify_preflight_purity had already run, so the
#618 branches-only guard and every other workspace-resolution layer behind
preflight received None and fell back to the MCP process root. A daemon
rooted at the stable control checkout therefore refused a valid registered
issue worktree the caller had explicitly supplied, before the publication
assessor could use it — the sole verify_preflight_purity call site that
accepted a worktree argument and dropped it.

Resolve the workspace once, before preflight, and forward it. A blank or
absent path forwards None and keeps the ordinary #618 fail-closed fallback,
so guard strictness is unchanged for missing, empty, unregistered, foreign,
or control-checkout worktrees. Public tool contract, ownership, cleanliness,
hash, ancestry, and read-after-write protections from PR #814 are untouched.

Adds tests/test_issue_815_preflight_worktree_forwarding.py: a #735-style
capture proving the argument reaches verify_preflight_purity, a faithful
production reproduction (control-rooted daemon, no session lock, explicit
worktree) that clears #618 on the fixed source and is trapped at #618 on the
unpatched source, an end-to-end control-rooted publication in the real
topology (PROJECT_ROOT is the stable control checkout, the issue worktree is
a distinct registered path, production guards forced on), and negative
coverage keeping every #618 and assessor refusal intact. The prior #812
suite masked the defect by patching PROJECT_ROOT to equal the issue worktree.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-22 17:27:49 -05:00
2 changed files with 640 additions and 2 deletions
+18 -2
View File
@@ -9173,11 +9173,27 @@ def gitea_publish_unpublished_issue_branch(
if blocked:
return blocked
verify_preflight_purity(remote, task=task, org=org, repo=repo)
# #815: resolve the caller's worktree *before* preflight and forward it, so
# every workspace-resolution layer behind verify_preflight_purity — including
# the #618 branches-only guard — judges the registered issue worktree this
# publication actually operates on. Resolving it afterwards let preflight
# fall back to the MCP process root, so a daemon rooted at the stable control
# checkout refused a valid explicit worktree before the assessor ever ran.
# A caller supplying nothing usable forwards None and keeps the ordinary
# fail-closed fallback.
explicit_worktree = (worktree_path or "").strip()
workspace = os.path.realpath(os.path.abspath(explicit_worktree or "."))
verify_preflight_purity(
remote,
worktree_path=workspace if explicit_worktree else None,
task=task,
org=org,
repo=repo,
)
h, o, r = _resolve(remote, host, org, repo)
git_remote = (git_remote_name or remote or "").strip()
workspace = os.path.realpath(os.path.abspath((worktree_path or "").strip() or "."))
existing_lock = issue_lock_store.load_issue_lock(
remote=remote, org=o, repo=r, issue_number=int(issue_number)
@@ -0,0 +1,622 @@
"""Publication preflight must receive the caller's worktree (#815).
``gitea_publish_unpublished_issue_branch`` takes a **required** ``worktree_path``
but resolved it only *after* ``verify_preflight_purity`` had already run. Every
workspace-resolution layer behind that preflight — canonical root, root checkout,
create-issue bootstrap, the #618 branches-only guard, issue scope, and anti-stomp
— therefore received ``None`` and fell back to the MCP process root. A daemon
rooted at the stable control checkout refused a valid registered issue worktree
that the caller had explicitly supplied, before the publication assessor ever ran.
The #812 suite could not see this. Its fixture sets ``self.worktree =
os.path.realpath(self.repo)`` and patches ``PROJECT_ROOT`` to that same path, so
the fallback resolved to the very worktree the argument named. The production
topology — control checkout on a stable branch, issue worktree somewhere else —
was never constructed, and preflight additionally no-ops under pytest unless
production guards are forced on.
These tests build that topology honestly:
* ``PROJECT_ROOT`` is a control checkout sitting on ``master``;
* the registered issue worktree is a genuinely separate path under ``branches/``;
* ``GITEA_TEST_FORCE_PRODUCTION_GUARDS`` is set so the #618 guard really runs;
* no patch makes the issue worktree appear to be ``PROJECT_ROOT``.
Nothing here reads, writes, or references the protected worktree named in #812
AC17 and #815 AC9. Every fixture is built from scratch against a local bare
remote, so publication and read-after-write verification genuinely execute.
"""
from __future__ import annotations
import os
import subprocess
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import issue_lock_provenance # noqa: E402
import issue_lock_store # noqa: E402
import mcp_server # noqa: E402
from mutation_profile_fixture import shared_mutation_env # noqa: E402
ISSUE = 9815
BRANCH = f"feat/issue-{ISSUE}-forwarding-fixture"
WORKTREE_DIRNAME = BRANCH.replace("/", "-")
IDENTITY = "example-user"
PROFILE = "test-author-prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
GIT_REMOTE = "prgs"
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"role": "author",
"allowed_operations": [
"gitea.read", "gitea.issue.create", "gitea.issue.comment",
"gitea.pr.create", "gitea.repo.commit", "gitea.branch.push",
],
"forbidden_operations": [],
"audit_label": "prgs-author",
}
def _ts(hours: int) -> str:
return (
(datetime.now(timezone.utc) + timedelta(hours=hours))
.isoformat()
.replace("+00:00", "Z")
)
class TestPreflightReceivesTheWorktree(unittest.TestCase):
"""AC1 — the supplied path reaches ``verify_preflight_purity`` itself.
Follows the #735 capture pattern: replace preflight with a recorder that
raises, so the argument can be proven forwarded without performing the
mutation. This is the direct unit-level statement of the defect.
"""
def _capture_preflight(self, **kwargs):
captured: dict = {}
def _capture(*a, **kw):
captured.update(kw)
captured["_args"] = a
raise RuntimeError("capture-only")
with patch.object(
mcp_server, "verify_preflight_purity", side_effect=_capture
), patch.object(
mcp_server, "get_profile", return_value=AUTHOR_PROFILE
), patch.object(
mcp_server, "_resolve",
return_value=("gitea.prgs.cc", ORG, REPO),
), patch.object(
mcp_server, "_auth", return_value="token fake",
), patch.object(
mcp_server.role_session_router,
"check_author_mutation_after_reviewer_stop",
return_value=(True, []),
), patch.object(
mcp_server, "_namespace_mutation_block", return_value=None
), patch.object(
mcp_server, "_profile_permission_block", return_value=None
):
try:
mcp_server.gitea_publish_unpublished_issue_branch(**kwargs)
except RuntimeError as exc:
if "capture-only" not in str(exc) and not captured:
raise
self.assertTrue(
captured,
"gitea_publish_unpublished_issue_branch never called "
"verify_preflight_purity",
)
return captured
def _base_kwargs(self, **overrides):
kwargs = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"worktree_path": "/tmp/issue-815-explicit-worktree",
"expected_head": "a" * 40,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"git_remote_name": GIT_REMOTE,
}
kwargs.update(overrides)
return kwargs
def test_explicit_worktree_path_reaches_preflight(self):
captured = self._capture_preflight(**self._base_kwargs())
self.assertEqual(
captured.get("worktree_path"),
os.path.realpath(os.path.abspath("/tmp/issue-815-explicit-worktree")),
"the authoritative worktree_path must be forwarded into preflight",
)
def test_forwarded_path_is_the_one_publication_uses(self):
"""AC4 — preflight and publication must judge the same resolved path."""
raw = "/tmp/issue-815-explicit-worktree/./"
captured = self._capture_preflight(**self._base_kwargs(worktree_path=raw))
expected = os.path.realpath(os.path.abspath(raw.strip()))
self.assertEqual(captured.get("worktree_path"), expected)
def test_blank_worktree_path_forwards_none(self):
"""AC5/AC8 — nothing usable supplied keeps the fail-closed fallback."""
for blank in ("", " "):
with self.subTest(blank=repr(blank)):
captured = self._capture_preflight(
**self._base_kwargs(worktree_path=blank)
)
self.assertIsNone(
captured.get("worktree_path"),
"a blank worktree must not resolve to the process cwd",
)
def test_org_repo_and_task_forwarding_are_not_regressed(self):
"""AC6 — #735's org/repo forwarding and the task name still hold."""
captured = self._capture_preflight(**self._base_kwargs())
self.assertEqual(captured.get("org"), ORG)
self.assertEqual(captured.get("repo"), REPO)
self.assertEqual(captured.get("task"), "publish_unpublished_branch")
class _ProductionTopologyBase(unittest.TestCase):
"""Control checkout on master + a distinct registered issue worktree.
This is the shape the production daemon runs in and the shape the #812
fixture never built. ``PROJECT_ROOT`` is the control checkout; the issue
worktree is a real registered worktree at a different path; production
guards are forced on so the #618 branches-only guard genuinely evaluates.
"""
def setUp(self):
self.lock_dir = tempfile.TemporaryDirectory()
self.addCleanup(self.lock_dir.cleanup)
self.origin = tempfile.mkdtemp(prefix="issue815-origin-")
self.control = tempfile.mkdtemp(prefix="issue815-control-")
for path in (self.origin, self.control):
self.addCleanup(
lambda p=path: subprocess.run(["rm", "-rf", p], check=False)
)
self._init_repos()
self.remotes = patch.dict(
mcp_server.REMOTES,
{"prgs": {"host": "gitea.prgs.cc", "org": ORG, "repo": REPO}},
)
self.remotes.start()
self.addCleanup(patch.stopall)
mcp_server._IDENTITY_CACHE.clear()
def _git(self, *args, cwd=None):
return subprocess.run(
["git", "-C", cwd or self.control, *args],
capture_output=True, text=True, check=True,
)
def _init_repos(self):
subprocess.run(
["git", "init", "-q", "--bare", "-b", "master", self.origin], check=True
)
self._git("init", "-q", "-b", "master")
self._git("config", "user.email", "[email protected]")
self._git("config", "user.name", "Test")
self._git("remote", "add", GIT_REMOTE, self.origin)
with open(os.path.join(self.control, "seed.txt"), "w") as fh:
fh.write("seed\n")
# The real repository gitignores branches/, so a registered worktree
# living there does not dirty the stable control checkout. Mirror that,
# or the #615 dirty-runtime block fires on the worktree we just created.
with open(os.path.join(self.control, ".gitignore"), "w") as fh:
fh.write("branches/\n")
self._git("add", "seed.txt", ".gitignore")
self._git("commit", "-q", "-m", "seed")
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
self._git("push", "-q", GIT_REMOTE, "master")
# The control checkout STAYS on master. This is the whole point: the
# daemon's process root is the stable control checkout, never the
# worktree the publication targets.
self.worktree = os.path.realpath(
os.path.join(self.control, "branches", WORKTREE_DIRNAME)
)
self._git("worktree", "add", "-q", "-b", BRANCH, self.worktree, "master")
with open(os.path.join(self.worktree, "work.txt"), "w") as fh:
fh.write("unpublished implementation\n")
self._git("add", "work.txt", cwd=self.worktree)
self._git("commit", "-q", "-m", "unpublished implementation", cwd=self.worktree)
self.head_sha = self._git("rev-parse", "HEAD", cwd=self.worktree).stdout.strip()
self.control_branch = self._git(
"rev-parse", "--abbrev-ref", "HEAD"
).stdout.strip()
# ── durable lock naming the caller and the issue worktree ────────────
def lock_path(self):
return issue_lock_store.lock_file_path(
remote="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
lock_dir=self.lock_dir.name,
)
def write_lock(self, *, bind_session=True, **overrides):
path = self.lock_path()
claimant = overrides.pop(
"claimant", {"username": IDENTITY, "profile": PROFILE}
)
pid = overrides.pop("session_pid", os.getpid())
lease = {
"operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
"issue_number": ISSUE,
"pr_number": None,
"branch": overrides.get("branch_name", BRANCH),
"worktree_path": overrides.get("worktree_path", self.worktree),
"claimant": claimant,
"created_at": _ts(-2),
"last_heartbeat_at": _ts(-2),
"expires_at": _ts(-1),
}
lease.update(overrides.pop("work_lease", {}))
data = {
"issue_number": ISSUE,
"branch_name": BRANCH,
"remote": "prgs",
"org": ORG,
"repo": REPO,
"worktree_path": self.worktree,
"session_pid": pid,
"pid": pid,
"lock_generation": 1,
"work_lease": lease,
"lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
tool="gitea_lock_issue", claimant=claimant
),
}
data.update(overrides)
data["lock_file_path"] = path
issue_lock_store.save_lock_file(path, data)
# Bind the session pointer so the #683 issue-scope guard resolves an
# owning issue for this author session. In real production the publish
# task does not require a session lock — require_author_lock is keyed on
# the test-only production_guards_forced() flag, which this suite must
# set to make preflight run at all — so this pointer is fixture
# scaffolding to clear a guard production would not apply here, never a
# softening of the worktree-forwarding behaviour under test. The
# preflight-negative cases below leave it unbound precisely so the #618
# guard is reached with no session fallback to rescue a bad worktree.
if bind_session:
pointer = {
"pid": os.getpid(),
"lock_file_path": path,
"issue_number": ISSUE,
"branch_name": data["branch_name"],
"remote": "prgs",
"org": ORG,
"repo": REPO,
}
issue_lock_store.save_lock_file(
issue_lock_store.session_pointer_path(self.lock_dir.name), pointer
)
return path
def _tool_env(self):
env = shared_mutation_env(
PROFILE, include_example_repo=True,
GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
)
env["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
# The defect only exists where preflight actually runs. Under pytest the
# production root/branches/scope guards are skipped unless forced on, so
# force them: this test exists to exercise the #618 guard, not to bypass
# it. Parity is pinned to the server's own startup head so repointing
# PROJECT_ROOT does not read as a stale daemon.
env["GITEA_TEST_FORCE_PRODUCTION_GUARDS"] = "1"
# Production is a promoted stable-control runtime. The pytest process
# itself runs from a branches/ worktree, which the #615 runtime-mode
# gate correctly classifies as dev-test; declaring the sanctioned mode
# models the production daemon rather than defeating the gate. Without
# this, forcing production guards on would trip the *runtime-mode* block
# for a reason unrelated to the #815 worktree-forwarding defect.
env["GITEA_MCP_RUNTIME_MODE"] = "stable-control"
startup_head = mcp_server._STARTUP_PARITY.get("startup_head") or ""
env["GITEA_TEST_CURRENT_HEAD"] = startup_head
env["GITEA_TEST_LIVE_REMOTE_HEAD"] = startup_head
return env
def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
"""Drive the public tool with PROJECT_ROOT pinned to the CONTROL checkout."""
env = self._tool_env()
with patch(
"mcp_server._list_open_pulls", return_value=list(open_prs or [])
), patch(
"mcp_server._auth", return_value="token x"
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": IDENTITY, "profile": PROFILE},
), patch.object(
# NOTE: the control checkout — deliberately NOT self.worktree.
mcp_server, "PROJECT_ROOT", self.control
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
return mcp_server.gitea_publish_unpublished_issue_branch(
issue_number=kwargs.pop("issue_number", ISSUE),
branch_name=kwargs.pop("branch_name", BRANCH),
worktree_path=kwargs.pop("worktree_path", self.worktree),
expected_head=expected_head or self.head_sha,
remote="prgs",
git_remote_name=kwargs.pop("git_remote_name", GIT_REMOTE),
**kwargs,
)
def remote_head(self, branch=BRANCH):
res = subprocess.run(
["git", "-C", self.origin, "rev-parse", "--verify", "--quiet", branch],
capture_output=True, text=True, check=False,
)
return (res.stdout or "").strip() or None
class TestForwardingClearsThe618Guard(_ProductionTopologyBase):
"""AC2 — the faithful production reproduction, and the sharpest fix proof.
The production recovery worker had **no** session issue lock — acquiring one
was the very thing the deadlock prevented — so preflight had nothing but the
explicit ``worktree_path`` argument to resolve the workspace from. This class
reproduces exactly that: no session pointer is bound, so there is no
author-lock fallback to rescue a dropped argument.
With the argument forwarded (fixed source) the #618 branches-only guard
accepts the registered issue worktree and the call advances to the next
guard. With the argument dropped (the buggy source this issue reports)
preflight falls back to ``PROJECT_ROOT`` — the stable control checkout — and
the #618 guard traps the call there. The two outcomes are told apart by the
guard that fired, on its own error text.
This test therefore *fails* against the unpatched source (the call is trapped
at #618 instead of clearing it), which is what makes it a regression rather
than a smoke test.
"""
_CONTROL_CHECKOUT_MARKERS = ("stable control checkout", "#618")
def test_explicit_worktree_clears_618_without_a_session_lock(self):
# No write_lock(): the session is deliberately unbound, as in production.
with self.assertRaises(RuntimeError) as ctx:
self.run_publish()
message = str(ctx.exception)
# The workspace guard is satisfied — the failure is the *later* scope
# guard (no owning issue), never the control-checkout refusal. If the
# argument were dropped, this call would be trapped at #618 instead.
for marker in self._CONTROL_CHECKOUT_MARKERS:
self.assertNotIn(
marker, message,
f"the explicit worktree must clear #618; got a control-checkout "
f"refusal instead: {message}",
)
self.assertIn(
"owning issue", message,
f"expected the downstream scope guard to fire, got: {message}",
)
self.assertIsNone(self.remote_head())
def test_dropped_argument_would_be_trapped_at_618(self):
# Simulate the buggy call shape directly: no session lock, and preflight
# given no worktree, exactly as the unpatched source left it. This pins
# the control-checkout refusal that the fix eliminates, so the pair of
# tests brackets the defect from both sides regardless of which source
# version is loaded.
env = self._tool_env()
with patch(
"mcp_server._list_open_pulls", return_value=[]
), patch(
"mcp_server._auth", return_value="token x"
), patch(
"mcp_server.get_auth_header", return_value="token x"
), patch(
"mcp_server._work_lease_claimant",
return_value={"username": IDENTITY, "profile": PROFILE},
), patch.object(
mcp_server, "PROJECT_ROOT", self.control
), patch.dict(os.environ, env, clear=True):
os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
with self.assertRaises(RuntimeError) as ctx:
# Drive verify_preflight_purity the way the buggy body did:
# no worktree_path forwarded at all.
mcp_server.verify_preflight_purity(
"prgs",
task="publish_unpublished_branch",
org=ORG,
repo=REPO,
)
message = str(ctx.exception)
self.assertTrue(
any(m in message for m in self._CONTROL_CHECKOUT_MARKERS),
f"a dropped worktree must trap at the control checkout: {message}",
)
self.assertIsNone(self.remote_head())
class TestProductionTopologyPublishes(_ProductionTopologyBase):
"""AC2/AC4/AC7 — the explicit registered worktree is what preflight validates."""
def test_fixture_is_genuinely_the_production_topology(self):
"""Guard the guard: if this drifts, the regression stops meaning anything."""
self.assertNotEqual(
os.path.realpath(self.control), self.worktree,
"the issue worktree must not be PROJECT_ROOT",
)
self.assertEqual(
self.control_branch, "master",
"the control checkout must sit on a stable branch",
)
self.assertTrue(
os.path.realpath(self.worktree).startswith(
os.path.realpath(os.path.join(self.control, "branches")) + os.sep
),
"the issue worktree must live under branches/",
)
listed = subprocess.run(
["git", "-C", self.control, "worktree", "list"],
capture_output=True, text=True, check=True,
).stdout
self.assertIn(
self.worktree, listed, "the issue worktree must be genuinely registered"
)
def test_publishes_from_a_control_rooted_daemon(self):
"""The exact production failure: this refused with #618 before the fix."""
self.write_lock()
self.assertIsNone(self.remote_head(), "fixture must start unpublished")
res = self.run_publish()
self.assertTrue(res.get("success"), res)
self.assertTrue(res.get("performed"), res)
self.assertEqual(self.remote_head(), self.head_sha)
def test_dry_run_uses_the_explicit_worktree(self):
"""AC4 — dry-run reaches the same decision without publishing."""
self.write_lock()
res = self.run_publish(dry_run=True)
self.assertTrue(res.get("success"), res)
self.assertFalse(res.get("performed"), res)
self.assertTrue(res.get("would_publish"), res)
self.assertIsNone(self.remote_head(), "dry-run must not publish")
def test_dry_run_and_apply_agree_on_the_same_worktree(self):
"""AC4 — both paths resolve the same workspace, so both succeed."""
self.write_lock()
dry = self.run_publish(dry_run=True)
self.assertTrue(dry.get("would_publish"), dry)
applied = self.run_publish()
self.assertTrue(applied.get("performed"), applied)
self.assertEqual(self.remote_head(), self.head_sha)
def test_read_after_write_verification_still_runs(self):
"""AC6 — PR #814's post-publication verification is unchanged."""
self.write_lock()
res = self.run_publish()
self.assertTrue(res.get("verified"), res)
self.assertEqual(res.get("remote_head_sha"), self.head_sha)
class TestProductionTopologyFailsClosed(_ProductionTopologyBase):
"""AC3/AC5/AC8 — the fix does not weaken any refusal.
A refusal reaches the caller by one of two mechanisms, and this class holds
them apart deliberately. A bad *workspace* is caught by the #618 preflight
guard, which raises before the assessor is built. A bad *content/ownership*
fact passes preflight (the worktree itself is fine) and is then refused by
the publication assessor, which returns ``success: False``. Both are
fail-closed; asserting the wrong mechanism would hide a regression.
"""
# ── #618 preflight refusals: no session lock, so nothing rescues a bad
# workspace and the guard fires exactly as it does in production ──────
def _assert_preflight_raises(self, **kwargs):
with self.assertRaises(RuntimeError) as ctx:
self.run_publish(**kwargs)
self.assertIsNone(
self.remote_head(), "a blocked publication must not reach the remote"
)
return str(ctx.exception)
def test_blank_worktree_path_fails_closed_via_618(self):
"""AC5 — a blank path forwards None, so preflight sees the control root."""
for blank in ("", " "):
with self.subTest(blank=repr(blank)):
message = self._assert_preflight_raises(worktree_path=blank)
self.assertIn("618", message)
def test_control_checkout_as_worktree_fails_closed_via_618(self):
"""AC5 — naming the stable control checkout explicitly is still refused."""
message = self._assert_preflight_raises(worktree_path=self.control)
self.assertIn("618", message)
def test_unregistered_directory_fails_closed(self):
"""AC3 — a plain directory under branches/ is not a registered worktree."""
bogus = os.path.join(self.control, "branches", "not-a-worktree")
os.makedirs(bogus, exist_ok=True)
self._assert_preflight_raises(worktree_path=bogus)
def test_missing_worktree_path_fails_closed(self):
"""AC3 — a path that does not exist is refused, not silently replaced."""
missing = os.path.join(self.control, "branches", "absent-worktree")
self._assert_preflight_raises(worktree_path=missing)
# ── assessor refusals: preflight passes on a valid worktree, then the
# publication assessor refuses on content/ownership evidence ──────────
def _assert_assessor_refuses(self, **kwargs):
res = self.run_publish(**kwargs)
self.assertFalse(res.get("success"), res)
self.assertFalse(res.get("performed"), res)
self.assertIsNone(self.remote_head())
return res
def test_changed_local_head_still_refuses(self):
"""AC6 — the declared expected_head remains authoritative."""
self.write_lock()
self._assert_assessor_refuses(expected_head="b" * 40)
def test_foreign_claimant_still_refuses(self):
"""AC6 — ownership still comes from the durable lock record."""
self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
self._assert_assessor_refuses()
def test_dirty_worktree_still_refuses(self):
"""AC6 — cleanliness enforcement survives the forwarding change."""
self.write_lock()
with open(os.path.join(self.worktree, "work.txt"), "a") as fh:
fh.write("uncommitted drift\n")
self._assert_assessor_refuses()
def test_competing_open_pr_still_refuses(self):
"""AC6 — a rival claim on another branch still blocks."""
self.write_lock()
self._assert_assessor_refuses(
open_prs=[{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
)
def test_issue_lock_record_is_not_mutated_by_a_refusal(self):
"""AC6 — record separation (#812 AC23) is unaffected by this change."""
path = self.write_lock()
with open(path, "rb") as fh:
before = fh.read()
self._assert_assessor_refuses(expected_head="c" * 40)
with open(path, "rb") as fh:
self.assertEqual(before, fh.read())
class TestProtectedFixtureNotReferenced(unittest.TestCase):
"""AC9 — this regression never names the protected #635 fixture.
The forbidden tokens are reconstructed from fragments so this assertion
file does not itself contain them and produce a false positive.
"""
def test_no_reference_to_the_protected_worktree(self):
forbidden = [
"issue-635-" + "project-registry-api",
"b2f6e9a6dc40e9651ef8" + "76f322dd0a68bddebfd8",
]
here = os.path.abspath(__file__)
with open(here, "r", encoding="utf-8") as fh:
text = fh.read()
for token in forbidden:
self.assertNotIn(
token, text,
f"the protected #635 fixture must not be referenced: {token}",
)
if __name__ == "__main__":
unittest.main()