Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 97 Pull Requests 7 Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into feat/issue-633-console-authz-audit-model

Browse Source
This commit is contained in:
jcwalker3
2026-07-22 21:57:53 -05:00
parent 479e434f92 53c2c92782
commit 66a89a46bb
15 changed files with 3337 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_issue_812_publish_unpublished_commit.py
+650
View File
@@ -0,0 +1,650 @@
"""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()
tests/test_issue_815_preflight_worktree_forwarding.py
+622
View File
@@ -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()
tests/test_task_capability_role_invariants.py
+3
View File
@@ -139,6 +139,9 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
"gitea_release_merger_pr_lease",
"create_branch",
"push_branch",
# #812 AC20: publishing an unpublished local head is author-only for the
# same reason every other push is — it writes a branch to the remote.
"publish_unpublished_branch",
"create_pr",
"commit_files",
"gitea_commit_files",
tests/test_webui_project_registry.py
+336 -31
View File
@@ -1,4 +1,4 @@
"""Tests for web UI project registry (#427)."""
"""Tests for web UI project registry (#427) and its API evolution (#635)."""
import json
import sys
import tempfile
@@ -11,57 +11,246 @@ from starlette.testclient import TestClient
from webui.app import create_app
from webui.project_registry import (
CURRENT_SCHEMA_VERSION,
REGISTRY_API_VERSION,
SUPPORTED_SCHEMA_VERSIONS,
RegistryError,
default_registry_path,
load_registry,
onboarding_summary,
project_to_dict,
)
from webui.registry_safety import is_forbidden_key
_REPO_ROOT = Path(__file__).resolve().parent.parent
_API_DOC = _REPO_ROOT / "docs" / "webui-project-registry-api.md"
class TestProjectRegistryLoader(unittest.TestCase):
def _valid_project(**overrides):
project = {
"id": "example",
"repo_name": "Example",
"gitea_owner": "Org",
"remote_host": "https://gitea.example.invalid",
"default_branch": "main",
"local_checkout_path": ".",
"profiles": {"author": "a", "reviewer": "r", "reconciler": "c"},
"workflow_paths": {"skill": "skills/x.md"},
}
project.update(overrides)
return project
def _write_registry(payload) -> Path:
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
json.dump(payload, handle)
return Path(handle.name)
class RegistryFileCase(unittest.TestCase):
"""Base class that cleans up temporary registry files."""
def setUp(self):
self._temp_paths: list[Path] = []
def tearDown(self):
for path in self._temp_paths:
path.unlink(missing_ok=True)
def write_registry(self, payload) -> Path:
path = _write_registry(payload)
self._temp_paths.append(path)
return path
class TestProjectRegistryLoader(RegistryFileCase):
def test_default_registry_loads_gitea_tools(self):
registry = load_registry()
self.assertEqual(registry.version, 1)
self.assertEqual(registry.version, CURRENT_SCHEMA_VERSION)
self.assertEqual(registry.schema_version, CURRENT_SCHEMA_VERSION)
self.assertEqual(registry.api_version, REGISTRY_API_VERSION)
self.assertEqual(len(registry.projects), 1)
project = registry.projects[0]
self.assertEqual(project.id, "gitea-tools")
self.assertEqual(project.repo_name, "Gitea-Tools")
self.assertEqual(project.gitea_owner, "Scaled-Tech-Consulting")
self.assertEqual(project.repo_full_name, "Scaled-Tech-Consulting/Gitea-Tools")
self.assertEqual(project.remote_host, "https://gitea.prgs.cc")
self.assertEqual(project.remote_name, "prgs")
self.assertEqual(project.status, "active")
self.assertEqual(project.profiles["author"], "prgs-author")
self.assertEqual(project.profiles["reviewer"], "prgs-reviewer")
self.assertEqual(project.profiles["reconciler"], "prgs-reconciler")
self.assertIn("skill", project.workflow_paths)
self.assertGreaterEqual(len(project.onboarding_checklist), 4)
def test_registry_rejects_credential_keys(self):
payload = {
def test_default_registry_onboarding_summary_is_complete(self):
summary = onboarding_summary(load_registry().projects[0])
self.assertEqual(summary.total, summary.complete)
self.assertEqual(summary.required_outstanding, 0)
self.assertTrue(summary.onboarding_complete)
def test_version_1_registry_still_loads_with_defaults(self):
path = self.write_registry({
"version": 1,
"projects": [
{
"id": "bad",
"repo_name": "Bad",
"gitea_owner": "Org",
"remote_host": "https://gitea.example.invalid",
"default_branch": "main",
"local_checkout_path": ".",
"profiles": {
"author": "a",
"reviewer": "r",
"reconciler": "c",
},
"workflow_paths": {"skill": "skills/x.md"},
"api_token": "secret",
}
_valid_project(
onboarding_checklist=[
{"id": "step", "title": "Step", "description": "Do it"}
]
)
],
}
})
registry = load_registry(path)
self.assertEqual(registry.schema_version, 1)
self.assertIn(1, SUPPORTED_SCHEMA_VERSIONS)
project = registry.projects[0]
self.assertEqual(project.status, "active")
self.assertIsNone(project.remote_name)
self.assertIsNone(project.last_seen_health)
step = project.onboarding_checklist[0]
self.assertEqual(step.state, "pending")
self.assertTrue(step.required)
self.assertFalse(onboarding_summary(project).onboarding_complete)
def test_onboarding_summary_counts_states(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
onboarding_checklist=[
{"id": "a", "title": "A", "description": "d", "state": "complete"},
{"id": "b", "title": "B", "description": "d", "state": "blocked"},
{
"id": "c",
"title": "C",
"description": "d",
"state": "pending",
"required": False,
},
{
"id": "d",
"title": "D",
"description": "d",
"state": "not_applicable",
},
]
)
],
})
summary = onboarding_summary(load_registry(path).projects[0])
self.assertEqual(summary.total, 4)
self.assertEqual(summary.complete, 1)
self.assertEqual(summary.blocked, 1)
self.assertEqual(summary.pending, 1)
self.assertEqual(summary.not_applicable, 1)
# Only the blocked step is both required and outstanding.
self.assertEqual(summary.required_outstanding, 1)
self.assertFalse(summary.onboarding_complete)
def test_last_seen_health_is_parsed_when_present(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
last_seen_health={
"status": "degraded",
"checked_at": "2026-01-01T00:00:00Z",
"detail": "daemon restart pending",
}
)
],
})
health = load_registry(path).projects[0].last_seen_health
self.assertIsNotNone(health)
self.assertEqual(health.status, "degraded")
self.assertEqual(health.checked_at, "2026-01-01T00:00:00Z")
def test_registry_rejects_credential_keys(self):
path = self.write_registry({
"version": 1,
"projects": [_valid_project(id="bad", api_token="redacted-placeholder")],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("credential", ctx.exception.remediation.lower())
self.assertEqual(ctx.exception.field_path, "projects[0].api_token")
def test_unsupported_version_fails_closed_with_remediation(self):
path = self.write_registry({"version": 99, "projects": [_valid_project()]})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("unsupported registry version", ctx.exception.message)
self.assertIn(str(CURRENT_SCHEMA_VERSION), ctx.exception.remediation)
self.assertEqual(ctx.exception.field_path, "version")
def test_missing_required_field_fails_closed(self):
broken = _valid_project()
del broken["default_branch"]
path = self.write_registry({"version": 2, "projects": [broken]})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("default_branch", ctx.exception.message)
self.assertEqual(ctx.exception.field_path, "projects[0]")
def test_unknown_status_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [_valid_project(status="mystery")],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects[0].status")
self.assertIn("active", ctx.exception.remediation)
def test_unknown_onboarding_state_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [
_valid_project(
onboarding_checklist=[
{"id": "a", "title": "A", "description": "d", "state": "almost"}
]
)
],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(
ctx.exception.field_path,
"projects[0].onboarding_checklist[0].state",
)
def test_missing_profile_role_fails_closed(self):
path = self.write_registry({
"version": 2,
"projects": [_valid_project(profiles={"author": "a", "reviewer": "r"})],
})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects[0].profiles.reconciler")
def test_empty_projects_fails_closed(self):
path = self.write_registry({"version": 2, "projects": []})
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertEqual(ctx.exception.field_path, "projects")
def test_invalid_json_fails_closed_with_location(self):
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
json.dump(payload, handle)
handle.write("{not json")
path = Path(handle.name)
try:
with self.assertRaises(ValueError):
load_registry(path)
finally:
path.unlink(missing_ok=True)
self._temp_paths.append(path)
with self.assertRaises(RegistryError) as ctx:
load_registry(path)
self.assertIn("not valid JSON", ctx.exception.message)
self.assertIn("line", ctx.exception.remediation)
def test_missing_file_fails_closed(self):
missing = Path(tempfile.gettempdir()) / "webui-registry-does-not-exist.json"
with self.assertRaises(RegistryError) as ctx:
load_registry(missing)
self.assertIn("could not be read", ctx.exception.message)
def test_default_registry_path_points_at_packaged_data(self):
path = default_registry_path()
@@ -81,31 +270,147 @@ class TestProjectRegistryRoutes(unittest.TestCase):
self.assertIn("prgs-author", response.text)
self.assertNotIn("child issue", response.text.lower())
def test_projects_page_shows_status_and_progress(self):
response = self.client.get("/projects")
self.assertIn("Status", response.text)
self.assertIn("Onboarding", response.text)
self.assertIn("4/4 complete", response.text)
def test_project_detail_renders_checklist(self):
response = self.client.get("/projects/gitea-tools")
self.assertEqual(response.status_code, 200)
self.assertIn("Onboarding checklist", response.text)
self.assertIn("Configure execution profiles", response.text)
self.assertIn("branches/", response.text)
self.assertIn("Complete", response.text)
self.assertIn("required outstanding 0", response.text)
def test_project_detail_404(self):
response = self.client.get("/projects/unknown-repo")
self.assertEqual(response.status_code, 404)
def test_api_projects_json(self):
def test_api_projects_alias_stays_compatible(self):
response = self.client.get("/api/projects")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["version"], 1)
# #427 consumers keep these keys.
self.assertEqual(data["version"], CURRENT_SCHEMA_VERSION)
self.assertIn("source_path", data)
self.assertEqual(len(data["projects"]), 1)
self.assertEqual(data["projects"][0]["id"], "gitea-tools")
self.assertIn("onboarding_checklist", data["projects"][0])
def test_api_v1_projects_payload(self):
response = self.client.get("/api/v1/projects")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
self.assertEqual(data["schema_version"], CURRENT_SCHEMA_VERSION)
self.assertEqual(data["project_count"], 1)
self.assertEqual(data["source"]["kind"], "file")
self.assertTrue(data["source"]["inventory_complete"])
project = data["projects"][0]
self.assertEqual(project["status"], "active")
self.assertEqual(project["remote_name"], "prgs")
self.assertEqual(
project["repo_full_name"], "Scaled-Tech-Consulting/Gitea-Tools"
)
self.assertTrue(project["onboarding_summary"]["onboarding_complete"])
self.assertEqual(project["onboarding_checklist"][0]["state"], "complete")
self.assertIsNone(project["last_seen_health"])
def test_api_v1_project_detail(self):
response = self.client.get("/api/v1/projects/gitea-tools")
self.assertEqual(response.status_code, 200)
data = response.json()
self.assertEqual(data["api_version"], REGISTRY_API_VERSION)
self.assertEqual(data["project"]["id"], "gitea-tools")
self.assertEqual(data["source"]["kind"], "file")
def test_api_v1_project_detail_missing_fails_closed(self):
response = self.client.get("/api/v1/projects/not-registered")
self.assertEqual(response.status_code, 404)
data = response.json()
self.assertEqual(data["error"], "project_not_found")
self.assertEqual(data["project_id"], "not-registered")
self.assertIn("gitea-tools", data["known_project_ids"])
self.assertIn("remediation", data)
def test_api_v1_projects_is_read_only(self):
response = self.client.post("/api/v1/projects", json={})
self.assertEqual(response.status_code, 405)
self.assertEqual(response.json()["error"], "read-only-mvp")
def test_project_to_dict_is_json_safe(self):
registry = load_registry()
encoded = json.dumps(project_to_dict(registry.projects[0]))
dto = project_to_dict(registry.projects[0])
encoded = json.dumps(dto)
self.assertIn("gitea-tools", encoded)
# Prose may mention tokens; no serialized *key* may look like a secret.
for key in dto:
with self.subTest(key=key):
self.assertFalse(is_forbidden_key(key))
class TestInvalidRegistryFailsClosedOverHttp(RegistryFileCase):
def setUp(self):
super().setUp()
self.path = self.write_registry({"version": 42, "projects": []})
self.client = TestClient(create_app())
def _with_bad_registry(self, url: str):
import os
from unittest import mock
with mock.patch.dict(
os.environ, {"WEBUI_PROJECT_REGISTRY": str(self.path)}, clear=False
):
return self.client.get(url)
def test_api_v1_reports_actionable_error(self):
response = self._with_bad_registry("/api/v1/projects")
self.assertEqual(response.status_code, 500)
data = response.json()
self.assertEqual(data["error"], "registry_invalid")
self.assertIn("unsupported registry version", data["detail"])
self.assertTrue(data["remediation"])
self.assertEqual(data["field_path"], "version")
def test_unversioned_alias_reports_actionable_error(self):
response = self._with_bad_registry("/api/projects")
self.assertEqual(response.status_code, 500)
self.assertEqual(response.json()["error"], "registry_invalid")
def test_html_page_reports_actionable_error(self):
response = self._with_bad_registry("/projects")
self.assertEqual(response.status_code, 500)
self.assertIn("Project registry unavailable", response.text)
self.assertIn("Remediation", response.text)
class TestProjectRegistryApiDocs(unittest.TestCase):
def test_api_contract_is_documented(self):
self.assertTrue(_API_DOC.is_file(), f"missing {_API_DOC}")
text = _API_DOC.read_text(encoding="utf-8")
for token in (
"/api/v1/projects",
"/api/v1/projects/{project_id}",
"/api/projects",
"onboarding_summary",
"last_seen_health",
"registry_invalid",
"#635",
):
with self.subTest(token=token):
self.assertIn(token, text)
def test_route_table_lists_versioned_routes(self):
local_dev = (_REPO_ROOT / "docs" / "webui-local-dev.md").read_text(
encoding="utf-8"
)
self.assertIn("/api/v1/projects", local_dev)
self.assertIn("webui-project-registry-api.md", local_dev)
if __name__ == "__main__":
unittest.main()
unittest.main()