"""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", "test@example.com") 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()