jcwalker3andClaude Opus 4.8 (1M context) <[email protected]>
99fda93bcc
Entry point B of #812 is the state where an author's work has already advanced to a local commit: the worktree is registered, clean, on the issue branch, and carries the only copy of the implementation, but the branch has never been published. Two individually correct predicates close a cycle around it: exact-owner lease renewal refuses without an observable remote head, and every publication path is lock-derived under #618, so nothing can create that remote head without first holding the lock renewal would grant.
This adds the missing operation. gitea_publish_unpublished_issue_branch publishes an already-committed local head to its remote branch, so exact-owner renewal has the evidence its model requires. Publication is the whole of its authority: it renews, reclaims, rebinds, and clears nothing.
Why this is not a lock bypass: the operation can only publish a branch whose durable issue-lock record already names the caller as claimant. Ownership is read from the lock file, never asserted by the caller. Guard strictness is unchanged (AC15): a dirty tree, an untracked file the commit does not carry, an unregistered worktree, a non-issue or stable branch, a changed local HEAD, a remote head that is not an ancestor of the commit, a competing open PR for the same issue on another branch, and any declared-hash mismatch each fail closed. The refspec names the commit SHA explicitly and never forces.
Record separation (AC23): the durable issue-lock file and the control-plane workflow lease are distinct records. This reads the former as ownership evidence and writes neither.
Truthful process evidence (AC24): the recorded owner pid's liveness is never consulted or asserted. A regression pins the recorded pid to a live process, proves publication still succeeds, and proves expired-lock reclaim still refuses for that same pid.
Scope is AC20 only. AC21 cannot unblock on its own, so the two are separable and only the smaller one is implemented here.
Tests: 36 new cases against synthetic fixtures only, using a real git repository with a real local bare remote so publication and read-after-write verification are genuinely executed rather than mocked. AC17 is honoured and a regression asserts the protected worktree is never referenced.
Full suite: 11 failed, 4354 passed, 6 skipped, 533 subtests passed. The 11 failures are the documented pre-existing drift baseline at 9eb0f29, unchanged in count and identity.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
651 lines
25 KiB
Python
651 lines
25 KiB
Python
"""Publication of an unpublished local commit (#812 AC20).
|
|
|
|
Entry point B of #812: a registered worktree, clean, on its issue branch,
|
|
holding a local commit that has never been published. Exact-owner lease renewal
|
|
refuses such a claim for want of an observable remote head, and every existing
|
|
publication path is lock-derived, so the two predicates close a cycle around
|
|
work that is otherwise complete.
|
|
|
|
These tests exercise the disposition through its *evidence*, never through any
|
|
particular issue number: every case uses an arbitrary issue number against a
|
|
synthetic repository, and the same assertions hold for any other. Nothing here
|
|
reads, writes, or references the live protected worktree named in #812 AC17 —
|
|
that content is preserved evidence for the duration of this work, so the
|
|
fixtures below build their own repositories from scratch.
|
|
|
|
The remote is a local bare repository, so publication and read-after-write
|
|
verification are genuinely executed rather than mocked.
|
|
"""
|
|
|
|
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 branch_publish # noqa: E402
|
|
import issue_lock_provenance # noqa: E402
|
|
import issue_lock_renewal # noqa: E402
|
|
import issue_lock_store # noqa: E402
|
|
import mcp_server # noqa: E402
|
|
from mutation_profile_fixture import shared_mutation_env # noqa: E402
|
|
|
|
ISSUE = 9812
|
|
BRANCH = f"feat/issue-{ISSUE}-publish-fixture"
|
|
IDENTITY = "example-user"
|
|
PROFILE = "test-author-prgs"
|
|
ORG = "Scaled-Tech-Consulting"
|
|
REPO = "Gitea-Tools"
|
|
GIT_REMOTE = "prgs"
|
|
|
|
|
|
def _ts(hours: int) -> str:
|
|
return (
|
|
(datetime.now(timezone.utc) + timedelta(hours=hours))
|
|
.isoformat()
|
|
.replace("+00:00", "Z")
|
|
)
|
|
|
|
|
|
class _PublishBase(unittest.TestCase):
|
|
"""Real git repo + real bare remote + durable lock naming the caller.
|
|
|
|
The recorded owner pid is deliberately **this live process**. That mirrors
|
|
the production shape #812 documents, where the pid belongs to a long-running
|
|
MCP daemon rather than to a dead author client, and it proves publication
|
|
never depends on a dead process (#812 AC24).
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.lock_dir = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.lock_dir.cleanup)
|
|
self.origin = tempfile.mkdtemp(prefix="issue812-origin-")
|
|
self.repo = tempfile.mkdtemp(prefix="issue812-work-")
|
|
for path in (self.origin, self.repo):
|
|
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()
|
|
|
|
# ── fixture construction ─────────────────────────────────────────────
|
|
def _git(self, *args, cwd=None):
|
|
return subprocess.run(
|
|
["git", "-C", cwd or self.repo, *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.repo, "seed.txt"), "w") as fh:
|
|
fh.write("seed\n")
|
|
self._git("add", "seed.txt")
|
|
self._git("commit", "-q", "-m", "seed")
|
|
self.base_sha = self._git("rev-parse", "HEAD").stdout.strip()
|
|
self._git("push", "-q", GIT_REMOTE, "master")
|
|
|
|
self._git("checkout", "-q", "-b", BRANCH)
|
|
with open(os.path.join(self.repo, "work.txt"), "w") as fh:
|
|
fh.write("unpublished implementation\n")
|
|
self._git("add", "work.txt")
|
|
self._git("commit", "-q", "-m", "unpublished implementation")
|
|
self.head_sha = 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="prgs", org=ORG, repo=REPO, issue_number=ISSUE,
|
|
lock_dir=self.lock_dir.name,
|
|
)
|
|
|
|
def write_lock(self, **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),
|
|
# Expired: entry point B's lease has lapsed, which is precisely why
|
|
# renewal — and therefore a published head — is needed.
|
|
"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)
|
|
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
|
|
# These tests repoint PROJECT_ROOT at a synthetic repository so the
|
|
# registered-worktree proof runs for real. Pin the parity gate to the
|
|
# server's own startup head so that repointing does not read as a stale
|
|
# daemon; the gate itself stays live and enforced.
|
|
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
|
|
|
|
# ── tool driver ──────────────────────────────────────────────────────
|
|
def run_publish(self, *, open_prs=None, expected_head=None, **kwargs):
|
|
"""Drive the public publication tool against the synthetic fixture."""
|
|
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(
|
|
mcp_server, "PROJECT_ROOT", self.repo
|
|
), 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 TestSuccessfulPublication(_PublishBase):
|
|
"""AC20 — the branch becomes observable and is verified after the write."""
|
|
|
|
def test_publishes_clean_unpublished_commit(self):
|
|
self.write_lock()
|
|
self.assertIsNone(self.remote_head(), "fixture must start unpublished")
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
self.assertTrue(result["performed"])
|
|
self.assertTrue(result["published"])
|
|
self.assertTrue(result["verified"], "read-after-write must be proven")
|
|
self.assertEqual(result["remote_head_sha"], self.head_sha)
|
|
self.assertEqual(self.remote_head(), self.head_sha)
|
|
|
|
def test_publication_does_not_rewrite_the_commit(self):
|
|
self.write_lock()
|
|
self.run_publish()
|
|
# The published object is the same commit, not a copy or a rewrite.
|
|
self.assertEqual(self.remote_head(), self.head_sha)
|
|
self.assertEqual(
|
|
self._git("rev-parse", "HEAD").stdout.strip(), self.head_sha
|
|
)
|
|
|
|
def test_exact_next_action_names_the_lock_call(self):
|
|
self.write_lock()
|
|
result = self.run_publish()
|
|
self.assertIn("gitea_lock_issue", result["exact_next_action"])
|
|
|
|
|
|
class TestFailsClosed(_PublishBase):
|
|
"""AC20/AC9 — each refusal reason, exercised independently."""
|
|
|
|
def test_changed_local_head_refuses(self):
|
|
self.write_lock()
|
|
stale = self.base_sha # a real commit, but not the declared head
|
|
result = self.run_publish(expected_head=stale)
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("local commit changed" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
self.assertIsNone(self.remote_head(), "refusal must not publish")
|
|
|
|
def test_abbreviated_sha_refuses(self):
|
|
self.write_lock()
|
|
result = self.run_publish(expected_head=self.head_sha[:8])
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("40-character" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
|
|
def test_dirty_tracked_worktree_refuses(self):
|
|
self.write_lock()
|
|
with open(os.path.join(self.repo, "work.txt"), "a") as fh:
|
|
fh.write("uncommitted edit\n")
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("dirty tracked files" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
self.assertIn("work.txt", result["evidence"]["dirty_tracked_files"])
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_untracked_file_refuses(self):
|
|
self.write_lock()
|
|
with open(os.path.join(self.repo, "stray.txt"), "w") as fh:
|
|
fh.write("not committed\n")
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("untracked files" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
self.assertIn("stray.txt", result["evidence"]["untracked_files"])
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_unexpected_remote_head_refuses(self):
|
|
"""A remote head that is not an ancestor must never be overwritten."""
|
|
self.write_lock()
|
|
# Publish a divergent commit to the branch from a separate line.
|
|
self._git("checkout", "-q", "-b", "divergent", self.base_sha)
|
|
with open(os.path.join(self.repo, "other.txt"), "w") as fh:
|
|
fh.write("someone else's work\n")
|
|
self._git("add", "other.txt")
|
|
self._git("commit", "-q", "-m", "divergent")
|
|
divergent = self._git("rev-parse", "HEAD").stdout.strip()
|
|
self._git("push", "-q", GIT_REMOTE, f"{divergent}:refs/heads/{BRANCH}")
|
|
self._git("checkout", "-q", BRANCH)
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("not an ancestor" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
self.assertEqual(
|
|
self.remote_head(), divergent, "the other head must survive intact"
|
|
)
|
|
|
|
def test_fast_forward_remote_head_is_allowed(self):
|
|
"""An ancestor head is an honest fast-forward, not a conflict."""
|
|
self.write_lock()
|
|
self._git("push", "-q", GIT_REMOTE, f"{self.base_sha}:refs/heads/{BRANCH}")
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
self.assertTrue(result["evidence"]["fast_forward_from_remote"])
|
|
self.assertEqual(self.remote_head(), self.head_sha)
|
|
|
|
def test_content_hash_mismatch_refuses(self):
|
|
self.write_lock()
|
|
wrong = {"work.txt": "0" * 64}
|
|
|
|
result = self.run_publish(expected_file_hashes=wrong)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("declared content hashes" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
self.assertFalse(result["evidence"]["file_hashes_verified"])
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_matching_content_hashes_publish(self):
|
|
self.write_lock()
|
|
digests = branch_publish.hash_worktree_files(self.worktree, ["work.txt"])
|
|
|
|
result = self.run_publish(expected_file_hashes=digests)
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
self.assertTrue(result["evidence"]["file_hashes_verified"])
|
|
|
|
def test_missing_declared_file_refuses(self):
|
|
self.write_lock()
|
|
result = self.run_publish(expected_file_hashes={"absent.txt": "0" * 64})
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("missing or unreadable" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
def test_foreign_claimant_refuses(self):
|
|
"""Ownership comes from the durable record, not from the caller."""
|
|
self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("foreign claim" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_foreign_profile_refuses(self):
|
|
self.write_lock(
|
|
claimant={"username": IDENTITY, "profile": "test-reviewer-prgs"}
|
|
)
|
|
result = self.run_publish()
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("claimant profile" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
|
|
def test_absent_lock_record_refuses(self):
|
|
"""No recorded claim means this cannot be used to bypass the lock."""
|
|
result = self.run_publish() # no write_lock()
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("no durable issue-lock record" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_branch_mismatch_against_lock_refuses(self):
|
|
self.write_lock(branch_name=f"feat/issue-{ISSUE}-different")
|
|
result = self.run_publish()
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("records branch" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
|
|
def test_worktree_mismatch_against_lock_refuses(self):
|
|
self.write_lock(worktree_path="/tmp/some/other/worktree")
|
|
result = self.run_publish()
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("records worktree" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
|
|
def test_competing_open_pr_on_another_branch_refuses(self):
|
|
self.write_lock()
|
|
competing = [{"number": 4242, "head": {"ref": f"fix/issue-{ISSUE}-rival"}}]
|
|
|
|
result = self.run_publish(open_prs=competing)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("already claim issue" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
self.assertIsNone(self.remote_head())
|
|
|
|
def test_open_pr_on_the_same_branch_is_not_competing(self):
|
|
"""This branch's own PR is not a rival claim against itself."""
|
|
self.write_lock()
|
|
own = [{"number": 77, "head": {"ref": BRANCH}}]
|
|
|
|
result = self.run_publish(open_prs=own)
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
|
|
|
|
class TestGuardStrictnessPreserved(_PublishBase):
|
|
"""AC15 — publication is an operation, never a weakening of the guards."""
|
|
|
|
def test_non_issue_branch_refuses(self):
|
|
self._git("checkout", "-q", "-b", "scratch/not-issue-linked")
|
|
self.write_lock(branch_name="scratch/not-issue-linked")
|
|
|
|
result = self.run_publish(branch_name="scratch/not-issue-linked")
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("issue-linked" in r for r in result["reasons"]), result["reasons"]
|
|
)
|
|
|
|
def test_stable_branch_refuses(self):
|
|
self.write_lock(branch_name="master")
|
|
result = self.run_publish(branch_name="master")
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("issue-linked" in r or "stable branch" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
def test_branch_number_must_match_the_issue(self):
|
|
other = "feat/issue-7777-mismatched"
|
|
self._git("checkout", "-q", "-b", other)
|
|
self.write_lock(branch_name=other)
|
|
result = self.run_publish(branch_name=other)
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("does not carry issue number" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
def test_unregistered_worktree_refuses(self):
|
|
"""#713 — an improvised directory is not a registered worktree."""
|
|
path = self.write_lock()
|
|
assessment = branch_publish.assess_unpublished_commit_publication(
|
|
issue_lock_store.read_lock_file(path),
|
|
issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
|
|
expected_head=self.head_sha, remote="prgs", org=ORG, repo=REPO,
|
|
identity=IDENTITY, profile=PROFILE,
|
|
worktree_state={
|
|
"current_branch": BRANCH, "porcelain_status": "",
|
|
"head_sha": self.head_sha,
|
|
},
|
|
worktree_registered=False,
|
|
remote_probe={"probe_ok": True, "remote_branch_exists": False},
|
|
)
|
|
self.assertEqual(assessment["outcome"], branch_publish.REFUSED)
|
|
self.assertTrue(
|
|
any("not listed in git worktree list" in r
|
|
for r in assessment["reasons"]),
|
|
assessment["reasons"],
|
|
)
|
|
|
|
def test_unobservable_remote_refuses(self):
|
|
"""An unknown remote state must not be mistaken for an absent branch."""
|
|
self.write_lock()
|
|
result = self.run_publish(git_remote_name="no-such-remote")
|
|
self.assertFalse(result["success"])
|
|
self.assertTrue(
|
|
any("could not be observed" in r for r in result["reasons"]),
|
|
result["reasons"],
|
|
)
|
|
|
|
|
|
class TestRecordSeparation(_PublishBase):
|
|
"""AC23 — the durable issue lock and the workflow lease are distinct."""
|
|
|
|
def test_publication_leaves_the_issue_lock_byte_identical(self):
|
|
path = self.write_lock()
|
|
with open(path, "rb") as fh:
|
|
before = fh.read()
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
with open(path, "rb") as fh:
|
|
after = fh.read()
|
|
self.assertEqual(before, after, "publication must not mutate the lock record")
|
|
self.assertFalse(result["issue_lock_record_mutated"])
|
|
self.assertFalse(result["workflow_lease_touched"])
|
|
|
|
def test_refusal_also_reports_untouched_records(self):
|
|
result = self.run_publish() # refuses: no lock record
|
|
self.assertFalse(result["issue_lock_record_mutated"])
|
|
self.assertFalse(result["workflow_lease_touched"])
|
|
|
|
def test_lock_generation_is_not_advanced(self):
|
|
path = self.write_lock()
|
|
self.run_publish()
|
|
lock = issue_lock_store.read_lock_file(path)
|
|
self.assertEqual(lock["lock_generation"], 1)
|
|
|
|
|
|
class TestTruthfulProcessEvidence(_PublishBase):
|
|
"""AC24 — a live daemon pid is never represented as a dead process."""
|
|
|
|
def test_live_recorded_pid_does_not_block_publication(self):
|
|
# The recorded pid is this live process, standing in for the live MCP
|
|
# daemon. Reclaim would refuse here; publication legitimately does not.
|
|
path = self.write_lock(session_pid=os.getpid())
|
|
lock = issue_lock_store.read_lock_file(path)
|
|
self.assertEqual(lock["pid"], os.getpid())
|
|
|
|
result = self.run_publish()
|
|
|
|
self.assertTrue(result["success"], result.get("reasons"))
|
|
self.assertEqual(self.remote_head(), self.head_sha)
|
|
|
|
def test_liveness_is_not_consulted_as_evidence(self):
|
|
self.write_lock(session_pid=os.getpid())
|
|
result = self.run_publish()
|
|
self.assertFalse(result["evidence"]["owner_pid_liveness_consulted"])
|
|
|
|
def test_reclaim_still_refuses_for_the_same_live_pid(self):
|
|
"""Publication does not soften the reclaim predicate it routes around."""
|
|
path = self.write_lock(session_pid=os.getpid())
|
|
lock = issue_lock_store.read_lock_file(path)
|
|
reclaim = issue_lock_store.assess_expired_lock_reclaim(lock)
|
|
self.assertFalse(reclaim["reclaim_allowed"])
|
|
|
|
|
|
class TestIdempotentRetry(_PublishBase):
|
|
"""AC20 — retry is safe and read-after-write is proven every time."""
|
|
|
|
def test_second_publication_reports_already_published(self):
|
|
self.write_lock()
|
|
first = self.run_publish()
|
|
self.assertTrue(first["performed"])
|
|
|
|
second = self.run_publish()
|
|
|
|
self.assertTrue(second["success"], second.get("reasons"))
|
|
self.assertFalse(second["performed"], "no second push is needed")
|
|
self.assertTrue(second["published"])
|
|
self.assertTrue(second["verified"])
|
|
self.assertEqual(second["outcome"], branch_publish.ALREADY_PUBLISHED)
|
|
self.assertEqual(self.remote_head(), self.head_sha)
|
|
|
|
|
|
class TestDryRun(_PublishBase):
|
|
"""AC12 — dry run reports the decision and mutates nothing."""
|
|
|
|
def test_dry_run_reports_intent_without_publishing(self):
|
|
self.write_lock()
|
|
|
|
result = self.run_publish(dry_run=True)
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["dry_run"])
|
|
self.assertTrue(result["would_publish"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertIsNone(self.remote_head(), "dry run must not publish")
|
|
|
|
def test_dry_run_and_apply_agree_on_a_refusal(self):
|
|
"""AC11 — the reported decision does not depend on which mode ran."""
|
|
self.write_lock(claimant={"username": "someone-else", "profile": PROFILE})
|
|
|
|
dry = self.run_publish(dry_run=True)
|
|
applied = self.run_publish()
|
|
|
|
self.assertFalse(dry["success"])
|
|
self.assertFalse(applied["success"])
|
|
self.assertEqual(dry["reasons"], applied["reasons"])
|
|
|
|
|
|
class TestRenewalUnblocked(_PublishBase):
|
|
"""AC20/AC21 — renewal is permitted only after verified publication."""
|
|
|
|
def _renewal(self, remote_head):
|
|
return issue_lock_renewal.assess_exact_owner_lease_renewal(
|
|
issue_lock_store.read_lock_file(self.lock_path()),
|
|
issue_number=ISSUE, branch_name=BRANCH, worktree_path=self.worktree,
|
|
remote="prgs", org=ORG, repo=REPO,
|
|
identity=IDENTITY, profile=PROFILE,
|
|
current_branch=BRANCH, porcelain_status="", worktree_exists=True,
|
|
head_sha=self.head_sha, remote_head_sha=remote_head,
|
|
)
|
|
|
|
def test_renewal_refuses_before_publication(self):
|
|
self.write_lock()
|
|
decision = self._renewal(None)
|
|
self.assertFalse(decision["renewal_sanctioned"])
|
|
self.assertTrue(
|
|
any("unpublished branch" in r for r in decision["reasons"]),
|
|
decision["reasons"],
|
|
)
|
|
|
|
def test_renewal_is_sanctioned_after_publication(self):
|
|
self.write_lock()
|
|
result = self.run_publish()
|
|
self.assertTrue(result["verified"], result.get("reasons"))
|
|
|
|
decision = self._renewal(self.remote_head())
|
|
|
|
self.assertTrue(decision["renewal_sanctioned"], decision["reasons"])
|
|
|
|
|
|
class TestProtectedAssetUntouched(unittest.TestCase):
|
|
"""AC17 — no test or fixture may reference the protected worktree."""
|
|
|
|
def test_no_reference_to_the_protected_worktree(self):
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
root = os.path.dirname(here)
|
|
needle = "issue-635-project-registry" + "-api"
|
|
for path in (
|
|
os.path.join(here, "test_issue_812_publish_unpublished_commit.py"),
|
|
os.path.join(root, "branch_publish.py"),
|
|
):
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
body = fh.read()
|
|
self.assertNotIn(needle, body)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
unittest.main()
|