Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2d5d5c9d17 |
@@ -355,6 +355,10 @@ def assess_anti_stomp_preflight(
|
||||
root_head_sha: str | None = None,
|
||||
root_porcelain: str | None = None,
|
||||
remote_master_sha: str | None = None,
|
||||
# #983: the tracking integration ref the SHA above came from, so root-checkout
|
||||
# contamination names the ref actually compared instead of a hardcoded
|
||||
# 'prgs/master'. None preserves the previous generic wording.
|
||||
remote_master_ref: str | None = None,
|
||||
check_root_checkout: bool = True,
|
||||
check_worktree: bool = True,
|
||||
create_issue_bootstrap_assessment: dict[str, Any] | None = None,
|
||||
@@ -553,6 +557,7 @@ def assess_anti_stomp_preflight(
|
||||
remote_master_sha=remote_master_sha,
|
||||
resolved_role=req_role or role,
|
||||
actual_role=role,
|
||||
remote_master_ref=remote_master_ref,
|
||||
)
|
||||
checks["root_checkout"] = {
|
||||
"block": bool(root_assessment.get("block")),
|
||||
|
||||
+243
-14
@@ -37,6 +37,23 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
|
||||
# Candidate git remote names probed when deriving repository identity.
|
||||
_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
|
||||
|
||||
# Fallback integration-branch names, probed only when the remote publishes no
|
||||
# ``refs/remotes/<remote>/HEAD`` symbolic ref (#983). Mirrors the stable base
|
||||
# branches recognised elsewhere in the workflow (``stacked_pr_support``). The
|
||||
# fallback is deliberately *not* ordered-first-wins: when more than one of these
|
||||
# refs exists and git records no default, the integration branch is genuinely
|
||||
# ambiguous and resolution fails closed instead of guessing.
|
||||
INTEGRATION_BRANCH_CANDIDATES: tuple[str, ...] = ("master", "main", "dev")
|
||||
|
||||
# Reason codes for base-ref derivation outcomes (#983), so callers and tests can
|
||||
# assert the refusal cause instead of string-matching prose.
|
||||
BASE_REF_SOURCE_REMOTE_HEAD = "remote_head_symref"
|
||||
BASE_REF_SOURCE_UNIQUE_CANDIDATE = "unique_integration_branch_ref"
|
||||
DENY_NO_IDENTITY_REMOTE = "no_identity_remote"
|
||||
DENY_AMBIGUOUS_REMOTE = "ambiguous_identity_remote"
|
||||
DENY_AMBIGUOUS_BASE_BRANCH = "ambiguous_integration_branch"
|
||||
DENY_NO_BASE_BRANCH = "no_integration_branch_ref"
|
||||
|
||||
# Repository-authority modes (#973 B10). Exactly two values are supported.
|
||||
# ``mode`` selects how repository authority is established, so an unrecognised
|
||||
# value must never be normalised onto one of these: aliasing a trusted mode is
|
||||
@@ -120,17 +137,15 @@ def resolve_repo_toplevel(path: str) -> str | None:
|
||||
return os.path.realpath(top) if top else None
|
||||
|
||||
|
||||
def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
|
||||
"""``owner/repository`` derived from a git remote configured at *path*.
|
||||
def _identity_remote_candidates(path: str, remote: str | None) -> list[str]:
|
||||
"""Ordered remote names to probe for identity at *path*.
|
||||
|
||||
Tries the caller-named remote first, then a small set of known remote names,
|
||||
then whatever remote the repository actually has. Returns None when no remote
|
||||
URL is parseable (identity cannot be proven).
|
||||
Names are used verbatim — never case-folded. Git config subsection names are
|
||||
case-sensitive, so a repository whose remote is ``MDCPS`` is reached only by
|
||||
the exact string ``MDCPS``; the lowercase entry in
|
||||
:data:`_IDENTITY_REMOTE_CANDIDATES` simply does not resolve, and the exact
|
||||
name arrives from ``git remote`` below (#983).
|
||||
"""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return None
|
||||
|
||||
ordered: list[str] = []
|
||||
for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
|
||||
clean = (name or "").strip()
|
||||
@@ -139,7 +154,7 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N
|
||||
|
||||
try:
|
||||
listed = subprocess.run(
|
||||
["git", "-C", text, "remote"],
|
||||
["git", "-C", path, "remote"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
@@ -149,11 +164,20 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N
|
||||
for name in listed:
|
||||
if name and name not in ordered:
|
||||
ordered.append(name)
|
||||
return ordered
|
||||
|
||||
for name in ordered:
|
||||
|
||||
def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[str, str]]:
|
||||
"""``(remote_name, owner/repository)`` for every probe name that resolves.
|
||||
|
||||
Remote names are returned exactly as configured so downstream tracking refs
|
||||
(``refs/remotes/<remote>/<branch>``) address the real ref (#983).
|
||||
"""
|
||||
found: list[tuple[str, str]] = []
|
||||
for name in _identity_remote_candidates(path, remote):
|
||||
try:
|
||||
url = subprocess.run(
|
||||
["git", "-C", text, "remote", "get-url", name],
|
||||
["git", "-C", path, "remote", "get-url", name],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
@@ -162,8 +186,213 @@ def repository_identity_slug(path: str, *, remote: str | None = None) -> str | N
|
||||
continue
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
|
||||
if parsed:
|
||||
return f"{parsed[0]}/{parsed[1]}"
|
||||
return None
|
||||
found.append((name, f"{parsed[0]}/{parsed[1]}"))
|
||||
return found
|
||||
|
||||
|
||||
def resolve_identity_remote(
|
||||
path: str, *, remote: str | None = None
|
||||
) -> tuple[str | None, str | None]:
|
||||
"""``(remote_name, owner/repository)`` for the remote that proves identity.
|
||||
|
||||
The remote *name* is the piece historically thrown away by
|
||||
:func:`repository_identity_slug`, even though resolving the slug already
|
||||
required discovering it. Cross-repository base-ref derivation needs that
|
||||
name to build ``refs/remotes/<remote>/<branch>``, so it is now returned
|
||||
rather than discarded (#983). Returns ``(None, None)`` when no remote URL is
|
||||
parseable (identity cannot be proven).
|
||||
"""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return None, None
|
||||
for name, slug in _configured_remote_identities(text, remote):
|
||||
return name, slug
|
||||
return None, None
|
||||
|
||||
|
||||
def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
|
||||
"""``owner/repository`` derived from a git remote configured at *path*.
|
||||
|
||||
Tries the caller-named remote first, then a small set of known remote names,
|
||||
then whatever remote the repository actually has. Returns None when no remote
|
||||
URL is parseable (identity cannot be proven).
|
||||
"""
|
||||
return resolve_identity_remote(path, remote=remote)[1]
|
||||
|
||||
|
||||
def _base_ref_result(
|
||||
*,
|
||||
proven: bool,
|
||||
reasons: list[str],
|
||||
remote: str | None = None,
|
||||
branch: str | None = None,
|
||||
repository_slug: str | None = None,
|
||||
source: str | None = None,
|
||||
reason_code: str | None = None,
|
||||
) -> dict:
|
||||
"""Build the base-ref derivation payload.
|
||||
|
||||
``tracking_refs`` is the ordered probe tuple downstream guards hand to
|
||||
``git rev-parse``: the ``<remote>/<branch>`` shorthand first, then the fully
|
||||
qualified ``refs/remotes/<remote>/<branch>``. It is empty whenever the
|
||||
derivation is not ``proven``, so an unresolved target can never be probed
|
||||
against some other repository's ref.
|
||||
"""
|
||||
tracking_ref = f"refs/remotes/{remote}/{branch}" if proven and remote and branch else None
|
||||
tracking_refs: tuple[str, ...] = (
|
||||
(f"{remote}/{branch}", tracking_ref) if tracking_ref else ()
|
||||
)
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"remote": remote,
|
||||
"branch": branch,
|
||||
"repository_slug": repository_slug,
|
||||
"tracking_ref": tracking_ref,
|
||||
"tracking_refs": tracking_refs,
|
||||
"source": source,
|
||||
"reason_code": reason_code,
|
||||
"reasons": list(reasons),
|
||||
}
|
||||
|
||||
|
||||
def _ref_exists(path: str, ref: str) -> bool:
|
||||
"""Whether *ref* resolves in the checkout at *path*."""
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", path, "rev-parse", "--verify", "--quiet", ref],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
except Exception:
|
||||
return False
|
||||
return res.returncode == 0 and bool((res.stdout or "").strip())
|
||||
|
||||
|
||||
def resolve_target_base_ref(path: str, *, remote: str | None = None) -> dict:
|
||||
"""Derive the authoritative integration base ref for the checkout at *path*.
|
||||
|
||||
This is the single resolved target shared by cross-repository mutation
|
||||
gating and parity reporting, so the two can never disagree about which
|
||||
commit a checkout is supposed to match (#983).
|
||||
|
||||
Resolution order:
|
||||
|
||||
1. The identity remote — the remote that already proves repository identity
|
||||
via :func:`resolve_identity_remote`, with its **exact configured case**
|
||||
preserved (``MDCPS`` stays ``MDCPS``).
|
||||
2. The integration branch, from ``refs/remotes/<remote>/HEAD`` — git's own
|
||||
record of that remote's default branch, written by ``clone`` and
|
||||
``remote set-head``. This is authoritative repository state, which is why
|
||||
no new configuration field is required.
|
||||
3. Only if the remote publishes no such symbolic ref, exactly one of
|
||||
:data:`INTEGRATION_BRANCH_CANDIDATES` present as a remote-tracking ref.
|
||||
|
||||
Fails closed — ``proven`` False, empty ``tracking_refs``, and a
|
||||
``reason_code`` — when identity is unprovable, when distinct remotes claim
|
||||
different repositories, when several candidate branches exist with no
|
||||
recorded default, or when no candidate exists at all. Nothing here invents a
|
||||
branch, writes a ref, or falls back to another repository's base.
|
||||
"""
|
||||
text = (path or "").strip()
|
||||
if not text:
|
||||
return _base_ref_result(
|
||||
proven=False,
|
||||
reasons=["no repository path supplied for base-ref derivation (fail closed)"],
|
||||
reason_code=DENY_NO_IDENTITY_REMOTE,
|
||||
)
|
||||
|
||||
identities = _configured_remote_identities(text, remote)
|
||||
if not identities:
|
||||
return _base_ref_result(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"no git remote at '{text}' yields a parseable repository identity, so "
|
||||
"the integration base ref cannot be derived (fail closed)"
|
||||
],
|
||||
reason_code=DENY_NO_IDENTITY_REMOTE,
|
||||
)
|
||||
|
||||
# Ambiguity only matters when the caller named no remote: if distinct remotes
|
||||
# describe different repositories there is no single authoritative target,
|
||||
# and picking the first would silently gate against the wrong repository.
|
||||
if not (remote or "").strip():
|
||||
distinct = {slug for _, slug in identities}
|
||||
if len(distinct) > 1:
|
||||
listed = ", ".join(f"{name} -> {slug}" for name, slug in identities)
|
||||
return _base_ref_result(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"ambiguous repository identity at '{text}': remotes resolve to "
|
||||
f"different repositories ({listed}); no single authoritative "
|
||||
"integration base ref can be derived (fail closed)"
|
||||
],
|
||||
reason_code=DENY_AMBIGUOUS_REMOTE,
|
||||
)
|
||||
|
||||
remote_name, slug = identities[0]
|
||||
|
||||
symref = None
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["git", "-C", text, "symbolic-ref", "--quiet", f"refs/remotes/{remote_name}/HEAD"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if res.returncode == 0:
|
||||
symref = (res.stdout or "").strip() or None
|
||||
except Exception:
|
||||
symref = None
|
||||
|
||||
prefix = f"refs/remotes/{remote_name}/"
|
||||
if symref and symref.startswith(prefix):
|
||||
branch = symref[len(prefix):].strip()
|
||||
if branch and branch != "HEAD":
|
||||
return _base_ref_result(
|
||||
proven=True,
|
||||
reasons=[],
|
||||
remote=remote_name,
|
||||
branch=branch,
|
||||
repository_slug=slug,
|
||||
source=BASE_REF_SOURCE_REMOTE_HEAD,
|
||||
)
|
||||
|
||||
present = [
|
||||
candidate
|
||||
for candidate in INTEGRATION_BRANCH_CANDIDATES
|
||||
if _ref_exists(text, f"{prefix}{candidate}")
|
||||
]
|
||||
if len(present) == 1:
|
||||
return _base_ref_result(
|
||||
proven=True,
|
||||
reasons=[],
|
||||
remote=remote_name,
|
||||
branch=present[0],
|
||||
repository_slug=slug,
|
||||
source=BASE_REF_SOURCE_UNIQUE_CANDIDATE,
|
||||
)
|
||||
if len(present) > 1:
|
||||
return _base_ref_result(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"remote '{remote_name}' publishes no '{prefix}HEAD' default and "
|
||||
f"several integration branches exist ({', '.join(present)}); the "
|
||||
"integration base ref is ambiguous (fail closed)"
|
||||
],
|
||||
reason_code=DENY_AMBIGUOUS_BASE_BRANCH,
|
||||
)
|
||||
return _base_ref_result(
|
||||
proven=False,
|
||||
reasons=[
|
||||
f"remote '{remote_name}' publishes no '{prefix}HEAD' default and none of "
|
||||
f"{'/'.join(INTEGRATION_BRANCH_CANDIDATES)} exists as a remote-tracking "
|
||||
f"ref under '{prefix}'; the integration base ref cannot be derived "
|
||||
"(fail closed; no fetch is performed here)"
|
||||
],
|
||||
reason_code=DENY_NO_BASE_BRANCH,
|
||||
)
|
||||
|
||||
|
||||
def assess_canonical_repository_root(
|
||||
|
||||
+27
-4
@@ -1052,9 +1052,16 @@ def _create_issue_bootstrap_assessment(
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
||||
remote_master_sha_error: str | None = None
|
||||
try:
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
# #983: consume the resolved-target state so an *underivable* base ref
|
||||
# reaches the assessor as named missing evidence rather than a bare
|
||||
# None, which the bootstrap would otherwise report only as "live
|
||||
# master tip is unknown" with no cause.
|
||||
_base_state = root_checkout_guard.resolve_remote_master_ref_state(
|
||||
ctx["canonical_repo_root"]
|
||||
)
|
||||
remote_master_sha = _base_state["sha"]
|
||||
if not remote_master_sha and _base_state.get("reasons"):
|
||||
remote_master_sha_error = "; ".join(_base_state["reasons"])
|
||||
except Exception as exc:
|
||||
remote_master_sha = None
|
||||
remote_master_sha_error = (
|
||||
@@ -1082,9 +1089,14 @@ def _create_issue_bootstrap_assessment(
|
||||
# closed instead of proceeding without base-equivalence proof.
|
||||
remote_master_sha_error: str | None = None
|
||||
try:
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(
|
||||
# #983: same resolved-target consumption as the author bootstrap above —
|
||||
# a target whose base ref cannot be derived must say why.
|
||||
_base_state = root_checkout_guard.resolve_remote_master_ref_state(
|
||||
ctx["canonical_repo_root"]
|
||||
)
|
||||
remote_master_sha = _base_state["sha"]
|
||||
if not remote_master_sha and _base_state.get("reasons"):
|
||||
remote_master_sha_error = "; ".join(_base_state["reasons"])
|
||||
except Exception as exc:
|
||||
remote_master_sha = None
|
||||
remote_master_sha_error = f"{type(exc).__name__}: {exc}".strip() or "resolver failed"
|
||||
@@ -1303,7 +1315,12 @@ def _run_anti_stomp_preflight(
|
||||
workspace = ctx["workspace_path"]
|
||||
canonical_root = ctx["canonical_repo_root"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
|
||||
# #983: derive the target base ref for this repository and carry the ref
|
||||
# itself alongside the SHA, so the anti-stomp root-checkout check reports the
|
||||
# ref it actually compared.
|
||||
_root_base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root)
|
||||
remote_master_sha = _root_base_state["sha"]
|
||||
remote_master_ref = _root_base_state.get("ref")
|
||||
|
||||
# Repo/org facts (best-effort; explicit org/repo when provided).
|
||||
resolved_org = org
|
||||
@@ -1432,6 +1449,7 @@ def _run_anti_stomp_preflight(
|
||||
root_head_sha=git_state.get("head_sha"),
|
||||
root_porcelain=git_state.get("porcelain_status") or "",
|
||||
remote_master_sha=remote_master_sha,
|
||||
remote_master_ref=remote_master_ref,
|
||||
startup_head=startup_head,
|
||||
current_code_head=current_code_head,
|
||||
lease_required=lease_required,
|
||||
@@ -1987,7 +2005,11 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
|
||||
canonical_root = ctx["canonical_repo_root"]
|
||||
workspace = ctx["workspace_path"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
|
||||
# #983: consume the resolved target so the contamination message names the
|
||||
# ref that was actually compared (refs/remotes/<remote>/<branch>) instead of
|
||||
# a hardcoded 'prgs/master' the target repository may not have.
|
||||
base_state = root_checkout_guard.resolve_remote_master_ref_state(canonical_root)
|
||||
remote_master_sha = base_state["sha"]
|
||||
assessment = root_checkout_guard.assess_root_checkout_guard(
|
||||
workspace_path=workspace,
|
||||
canonical_repo_root=canonical_root,
|
||||
@@ -1997,6 +2019,7 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
|
||||
remote_master_sha=remote_master_sha,
|
||||
resolved_role=_preflight_resolved_role,
|
||||
actual_role=_actual_profile_role(),
|
||||
remote_master_ref=base_state.get("ref"),
|
||||
)
|
||||
if assessment["block"]:
|
||||
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
|
||||
|
||||
+38
-9
@@ -378,6 +378,11 @@ def format_parity(assessment: dict) -> str:
|
||||
# feeds the mutation gate and never changes startup_head/current_head.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Legacy hardcoded target tracking ref. Retained for callers that still pass an
|
||||
# explicit ref, but no longer the default: assuming a remote named ``origin``
|
||||
# read an unrelated (often orphaned) remote-tracking ref in any checkout whose
|
||||
# remote is named something else, and reported the target stale against a commit
|
||||
# from a remote that may no longer even be configured (#983).
|
||||
DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"
|
||||
|
||||
|
||||
@@ -408,7 +413,7 @@ def assess_target_repository_parity(
|
||||
*,
|
||||
canonical_root: str | None,
|
||||
source: str | None,
|
||||
tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
|
||||
tracking_ref: str | None = None,
|
||||
) -> dict:
|
||||
"""Assess the configured cross-repository target checkout.
|
||||
|
||||
@@ -421,6 +426,11 @@ def assess_target_repository_parity(
|
||||
An unconfigured namespace is ``configured=False`` and never ``stale`` — the
|
||||
single-repository default has no second dimension to be stale about. A
|
||||
configured root that cannot be read is ``determinable=False`` with reasons.
|
||||
|
||||
*tracking_ref* defaults to None, meaning **derive the target from the
|
||||
repository itself** through the same resolver the mutation guard uses, so
|
||||
gating and reporting can never disagree about which ref is authoritative
|
||||
(#983). Passing an explicit ref preserves the previous behaviour verbatim.
|
||||
"""
|
||||
result = {
|
||||
"configured": bool(canonical_root),
|
||||
@@ -429,6 +439,9 @@ def assess_target_repository_parity(
|
||||
"repository_slug": None,
|
||||
"checkout_head": None,
|
||||
"tracking_ref": tracking_ref,
|
||||
"base_remote": None,
|
||||
"base_branch": None,
|
||||
"tracking_ref_source": "explicit_tracking_ref" if tracking_ref else None,
|
||||
"remote_tracking_head": None,
|
||||
"determinable": False,
|
||||
"stale": False,
|
||||
@@ -463,19 +476,35 @@ def assess_target_repository_parity(
|
||||
result["checkout_head"] = head
|
||||
result["determinable"] = True
|
||||
|
||||
remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
|
||||
if remote_url:
|
||||
# Local import keeps this module dependency-light for its startup role.
|
||||
import remote_repo_guard
|
||||
# Local import keeps this module dependency-light for its startup role.
|
||||
import canonical_repository_root as _crr
|
||||
|
||||
parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
|
||||
if parsed:
|
||||
result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
|
||||
if not result["repository_slug"]:
|
||||
# #983: identity comes from whichever remote actually proves it, not from a
|
||||
# remote assumed to be named 'origin'. In a checkout whose only remote is
|
||||
# 'prgs', the old lookup failed outright and reported the identity as
|
||||
# underivable while a leftover refs/remotes/origin/master still resolved.
|
||||
#
|
||||
# Identity is resolved independently of the base ref: a target that has never
|
||||
# been fetched still has a provable repository identity, and reporting it as
|
||||
# unidentifiable would lose real information over an unrelated missing ref.
|
||||
identity_remote, slug = _crr.resolve_identity_remote(canonical_root)
|
||||
result["repository_slug"] = slug
|
||||
if not slug:
|
||||
result["reasons"].append(
|
||||
"target repository identity could not be derived from its git remote"
|
||||
)
|
||||
|
||||
if not tracking_ref:
|
||||
base = _crr.resolve_target_base_ref(canonical_root, remote=identity_remote)
|
||||
if not base.get("proven"):
|
||||
result["reasons"].extend(base.get("reasons") or [])
|
||||
return result
|
||||
tracking_ref = base["tracking_ref"]
|
||||
result["tracking_ref"] = tracking_ref
|
||||
result["tracking_ref_source"] = base.get("source")
|
||||
result["base_remote"] = base.get("remote")
|
||||
result["base_branch"] = base.get("branch")
|
||||
|
||||
tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
|
||||
if not tracking_head:
|
||||
result["reasons"].append(
|
||||
|
||||
+134
-9
@@ -1,9 +1,16 @@
|
||||
"""Root checkout guard (#475).
|
||||
|
||||
The project root checkout is the stable control checkout on master/prgs/master.
|
||||
Author/reviewer/merge flows must fail closed when the control checkout is
|
||||
contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of
|
||||
prgs/master). Isolated ``branches/...`` worktrees remain allowed.
|
||||
The project root checkout is the stable control checkout on its integration
|
||||
branch. Author/reviewer/merge flows must fail closed when the control checkout
|
||||
is contaminated (wrong branch, detached HEAD, dirty, or HEAD behind/ahead of the
|
||||
tracking integration ref). Isolated ``branches/...`` worktrees remain allowed.
|
||||
|
||||
The tracking ref is *derived per repository* (#983) rather than assumed to be
|
||||
``prgs/master``: a namespace bound to another repository — say remote ``MDCPS``
|
||||
on branch ``dev`` — is gated against ``refs/remotes/MDCPS/dev``. Derivation is
|
||||
delegated to :mod:`canonical_repository_root`, the authoritative
|
||||
repository/context resolver, so gating and parity reporting share one resolved
|
||||
target instead of maintaining two disagreeing hardcoded defaults.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -11,6 +18,7 @@ from __future__ import annotations
|
||||
import os
|
||||
import subprocess
|
||||
|
||||
import canonical_repository_root
|
||||
from author_mutation_worktree import is_path_under_branches
|
||||
from reviewer_worktree import parse_dirty_tracked_files
|
||||
|
||||
@@ -20,19 +28,57 @@ REMEDIATION = (
|
||||
)
|
||||
|
||||
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
||||
|
||||
# Legacy PRGS-specific probe order. Retained only for callers that pass an
|
||||
# explicit ``remote_refs`` override; it is no longer the silent default, because
|
||||
# inheriting it in a non-PRGS checkout compared that checkout against a ref it
|
||||
# can never have (#983).
|
||||
REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master")
|
||||
|
||||
|
||||
def _derive_probe_refs(root: str, remote: str | None) -> dict:
|
||||
"""Derive the ordered tracking refs to probe for *root*.
|
||||
|
||||
Returns the derivation payload from :mod:`canonical_repository_root` plus a
|
||||
``refs`` tuple, which is empty when the target is not provable.
|
||||
"""
|
||||
derived = canonical_repository_root.resolve_target_base_ref(root, remote=remote)
|
||||
return {
|
||||
"refs": tuple(derived.get("tracking_refs") or ()),
|
||||
"remote": derived.get("remote"),
|
||||
"branch": derived.get("branch"),
|
||||
"source": derived.get("source"),
|
||||
"proven": bool(derived.get("proven")),
|
||||
"reason_code": derived.get("reason_code"),
|
||||
"reasons": list(derived.get("reasons") or []),
|
||||
}
|
||||
|
||||
|
||||
def resolve_remote_master_sha(
|
||||
canonical_repo_root: str,
|
||||
*,
|
||||
remote_refs: tuple[str, ...] | None = None,
|
||||
remote: str | None = None,
|
||||
) -> str | None:
|
||||
"""Return the commit SHA for the tracking master ref when available."""
|
||||
"""Return the commit SHA for the tracking integration ref when available.
|
||||
|
||||
This remains the single place that turns a ref into a SHA. Returns None when
|
||||
the target cannot be derived or resolved — exactly what this function already
|
||||
returned when ``rev-parse`` failed. Callers that must fail closed on missing
|
||||
evidence (the #749/#757 bootstrap path) surface that None as *missing
|
||||
evidence*, never as "no constraint".
|
||||
"""
|
||||
root = (canonical_repo_root or "").strip()
|
||||
if not root:
|
||||
return None
|
||||
for ref in remote_refs or REMOTE_MASTER_REFS:
|
||||
if remote_refs:
|
||||
probe: tuple[str, ...] = tuple(remote_refs)
|
||||
else:
|
||||
derived = _derive_probe_refs(root, remote)
|
||||
if not derived["proven"]:
|
||||
return None
|
||||
probe = derived["refs"]
|
||||
for ref in probe:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, "rev-parse", "--verify", ref],
|
||||
capture_output=True,
|
||||
@@ -46,6 +92,80 @@ def resolve_remote_master_sha(
|
||||
return None
|
||||
|
||||
|
||||
def resolve_remote_master_ref_state(
|
||||
canonical_repo_root: str,
|
||||
*,
|
||||
remote_refs: tuple[str, ...] | None = None,
|
||||
remote: str | None = None,
|
||||
) -> dict:
|
||||
"""Resolve the tracking integration ref together with its commit SHA.
|
||||
|
||||
Returns ``sha``, the ``ref`` it came from, the derived ``remote`` /
|
||||
``branch``, a machine-checkable ``reason_code``, and ``reasons``. ``sha`` is
|
||||
None whenever the target cannot be resolved — never a fallback to some other
|
||||
repository's commit.
|
||||
|
||||
The SHA itself is obtained through :func:`resolve_remote_master_sha` rather
|
||||
than by re-probing here, so one public function stays authoritative for
|
||||
ref-to-SHA resolution. An explicit *remote_refs* keeps the historical
|
||||
behaviour exactly: those refs are probed in order and no derivation happens.
|
||||
"""
|
||||
root = (canonical_repo_root or "").strip()
|
||||
state: dict = {
|
||||
"sha": None,
|
||||
"ref": None,
|
||||
"remote": None,
|
||||
"branch": None,
|
||||
"source": None,
|
||||
"reason_code": None,
|
||||
"reasons": [],
|
||||
}
|
||||
if not root:
|
||||
state["reasons"].append("no canonical repository root supplied (fail closed)")
|
||||
return state
|
||||
|
||||
if remote_refs:
|
||||
probe: tuple[str, ...] = tuple(remote_refs)
|
||||
state["source"] = "explicit_remote_refs"
|
||||
else:
|
||||
derived = _derive_probe_refs(root, remote)
|
||||
state["remote"] = derived["remote"]
|
||||
state["branch"] = derived["branch"]
|
||||
state["source"] = derived["source"]
|
||||
if not derived["proven"]:
|
||||
state["reason_code"] = derived["reason_code"]
|
||||
state["reasons"] = derived["reasons"]
|
||||
return state
|
||||
probe = derived["refs"]
|
||||
|
||||
sha = resolve_remote_master_sha(root, remote_refs=probe, remote=remote)
|
||||
if not sha:
|
||||
state["reasons"].append(
|
||||
"tracking integration ref "
|
||||
f"{' / '.join(probe) if probe else '(none derived)'} does not resolve in "
|
||||
f"'{root}' (fail closed)"
|
||||
)
|
||||
return state
|
||||
|
||||
state["sha"] = sha
|
||||
# Name the ref that actually carries this commit. When the SHA comes from a
|
||||
# test double no probe will match, so fall back to the first derived ref,
|
||||
# which is the one the guard is conceptually comparing against.
|
||||
for ref in probe:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, "rev-parse", "--verify", ref],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
if res.returncode == 0 and (res.stdout or "").strip() == sha:
|
||||
state["ref"] = ref
|
||||
break
|
||||
else:
|
||||
state["ref"] = probe[0] if probe else None
|
||||
return state
|
||||
|
||||
|
||||
resolve_tracking_master_sha = resolve_remote_master_sha
|
||||
|
||||
|
||||
@@ -59,8 +179,9 @@ def assess_root_checkout_guard(
|
||||
remote_master_sha: str | None,
|
||||
resolved_role: str | None = None,
|
||||
actual_role: str | None = None,
|
||||
remote_master_ref: str | None = None,
|
||||
) -> dict:
|
||||
"""Fail closed when the control checkout is not clean master/prgs/master.
|
||||
"""Fail closed when the control checkout is not clean on its integration ref.
|
||||
|
||||
``resolved_role`` is the preflight-resolved *task* role and ``actual_role``
|
||||
is the *active profile* role (#540). The reconciler exemption honours either
|
||||
@@ -102,9 +223,13 @@ def assess_root_checkout_guard(
|
||||
)
|
||||
|
||||
if remote_master_sha and head_sha and head_sha != remote_master_sha:
|
||||
# #983: name the ref that was actually compared. Reporting a literal
|
||||
# 'prgs/master' in a checkout gated against refs/remotes/MDCPS/dev sends
|
||||
# the operator to inspect a ref that repository does not have.
|
||||
ref_label = (remote_master_ref or "").strip() or "the tracking integration ref"
|
||||
reasons.append(
|
||||
"control checkout HEAD does not match prgs/master "
|
||||
f"(HEAD {head_sha[:12]}, prgs/master {remote_master_sha[:12]})"
|
||||
f"control checkout HEAD does not match {ref_label} "
|
||||
f"(HEAD {head_sha[:12]}, {ref_label} {remote_master_sha[:12]})"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
|
||||
@@ -0,0 +1,482 @@
|
||||
"""Regression tests for Issue #983: derived target base ref for cross-repository checkouts.
|
||||
|
||||
The mutation guard previously assumed ``prgs/master`` and the parity report
|
||||
assumed ``origin/master``. Any repository using neither — for example remote
|
||||
``MDCPS`` on integration branch ``dev`` — could not prove base equivalence, so
|
||||
every gated mutation failed closed with no reachable remedy.
|
||||
|
||||
These tests build hermetic git repositories on disk (no network, no fetch) and
|
||||
assert the derived target end to end: identity remote, integration branch,
|
||||
tracking ref, fail-closed refusals, and agreement between the mutation guard and
|
||||
the parity report.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import inspect
|
||||
import os
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import anti_stomp_preflight
|
||||
import canonical_repository_root as crr
|
||||
import master_parity_gate
|
||||
import root_checkout_guard
|
||||
|
||||
|
||||
def _git(root: str, *args: str) -> str:
|
||||
res = subprocess.run(
|
||||
["git", "-C", root, *args],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
return (res.stdout or "").strip()
|
||||
|
||||
|
||||
def _make_repo(root: str, *, remote: str | None, url: str | None) -> str:
|
||||
"""Initialise a repository with one commit and an optional named remote."""
|
||||
os.makedirs(root, exist_ok=True)
|
||||
_git(root, "init", "--quiet")
|
||||
_git(root, "config", "user.email", "[email protected]")
|
||||
_git(root, "config", "user.name", "Issue983 Test")
|
||||
_git(root, "config", "commit.gpgsign", "false")
|
||||
with open(os.path.join(root, "seed.txt"), "w", encoding="utf-8") as fh:
|
||||
fh.write("seed\n")
|
||||
_git(root, "add", "seed.txt")
|
||||
_git(root, "commit", "--quiet", "-m", "seed")
|
||||
if remote and url:
|
||||
_git(root, "remote", "add", remote, url)
|
||||
return _git(root, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
def _set_remote_branch(root: str, remote: str, branch: str, sha: str) -> None:
|
||||
"""Create refs/remotes/<remote>/<branch> without contacting a network."""
|
||||
_git(root, "update-ref", f"refs/remotes/{remote}/{branch}", sha)
|
||||
|
||||
|
||||
def _set_remote_head(root: str, remote: str, branch: str) -> None:
|
||||
_git(
|
||||
root,
|
||||
"symbolic-ref",
|
||||
f"refs/remotes/{remote}/HEAD",
|
||||
f"refs/remotes/{remote}/{branch}",
|
||||
)
|
||||
|
||||
|
||||
def _advance(root: str, message: str) -> str:
|
||||
with open(os.path.join(root, "seed.txt"), "a", encoding="utf-8") as fh:
|
||||
fh.write(message + "\n")
|
||||
_git(root, "add", "seed.txt")
|
||||
_git(root, "commit", "--quiet", "-m", message)
|
||||
return _git(root, "rev-parse", "HEAD")
|
||||
|
||||
|
||||
class _RepoCase(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.root = os.path.join(self._tmp.name, "repo")
|
||||
|
||||
|
||||
class TestPrgsBehaviourPreserved(_RepoCase):
|
||||
"""AC1: existing PRGS behaviour using prgs/master is unchanged."""
|
||||
|
||||
def test_prgs_master_resolves_unchanged(self):
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="prgs",
|
||||
url="https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "prgs", "master", head)
|
||||
_set_remote_head(self.root, "prgs", "master")
|
||||
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertEqual(got["remote"], "prgs")
|
||||
self.assertEqual(got["branch"], "master")
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master")
|
||||
self.assertEqual(got["repository_slug"], "Scaled-Tech-Consulting/Gitea-Tools")
|
||||
self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head)
|
||||
|
||||
def test_legacy_explicit_remote_refs_path_is_untouched(self):
|
||||
"""An explicit remote_refs override still short-circuits derivation."""
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="prgs",
|
||||
url="https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "prgs", "master", head)
|
||||
|
||||
state = root_checkout_guard.resolve_remote_master_ref_state(
|
||||
self.root, remote_refs=root_checkout_guard.REMOTE_MASTER_REFS
|
||||
)
|
||||
self.assertEqual(state["sha"], head)
|
||||
self.assertEqual(state["source"], "explicit_remote_refs")
|
||||
|
||||
|
||||
class TestCrossRepositoryTarget(_RepoCase):
|
||||
"""AC2/AC3/AC4: MDCPS/dev, no origin remote, exact remote-name case."""
|
||||
|
||||
def _mdcps(self) -> str:
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_head(self.root, "MDCPS", "dev")
|
||||
return head
|
||||
|
||||
def test_mdcps_dev_resolves(self):
|
||||
head = self._mdcps()
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertEqual(got["remote"], "MDCPS")
|
||||
self.assertEqual(got["branch"], "dev")
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
|
||||
self.assertEqual(got["repository_slug"], "MDCPS/WeeklyBriefings-Meta")
|
||||
self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head)
|
||||
|
||||
def test_no_remote_named_origin(self):
|
||||
self._mdcps()
|
||||
self.assertEqual(_git(self.root, "remote"), "MDCPS")
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertNotIn("origin", got["tracking_ref"])
|
||||
|
||||
def test_remote_name_case_is_preserved_exactly(self):
|
||||
self._mdcps()
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertEqual(got["remote"], "MDCPS")
|
||||
self.assertNotEqual(got["remote"], "mdcps")
|
||||
# The tracking ref must address the real ref, which is case-sensitive.
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
|
||||
self.assertTrue(
|
||||
_git(self.root, "rev-parse", "--verify", got["tracking_ref"]),
|
||||
"case-preserved tracking ref must resolve",
|
||||
)
|
||||
|
||||
def test_lowercase_candidate_never_supplies_the_remote_name(self):
|
||||
"""Guards against silently case-folding MDCPS to the candidate 'mdcps'.
|
||||
|
||||
``_IDENTITY_REMOTE_CANDIDATES`` contains a lowercase ``mdcps`` entry and
|
||||
is probed *before* the repository's own remote listing. Git remote names
|
||||
live in case-sensitive config subsections on every platform, so the
|
||||
lowercase probe cannot resolve and the exact-case name must arrive from
|
||||
``git remote``. Asserted through config rather than ref lookup because a
|
||||
case-insensitive filesystem (macOS) resolves loose refs either way, which
|
||||
would make a ref-based assertion test the filesystem instead of the code.
|
||||
"""
|
||||
self._mdcps()
|
||||
res = subprocess.run(
|
||||
["git", "-C", self.root, "remote", "get-url", "mdcps"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
self.assertNotEqual(res.returncode, 0, "git remote names are case-sensitive")
|
||||
|
||||
# Even when the caller *hints* the wrong case, the resolved name is exact.
|
||||
got = crr.resolve_target_base_ref(self.root, remote="mdcps")
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertEqual(got["remote"], "MDCPS")
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
|
||||
|
||||
name, slug = crr.resolve_identity_remote(self.root)
|
||||
self.assertEqual(name, "MDCPS")
|
||||
self.assertEqual(slug, "MDCPS/WeeklyBriefings-Meta")
|
||||
|
||||
|
||||
class TestTargetStaleness(_RepoCase):
|
||||
"""AC5/AC6: local equal to, behind, or divergent from the resolved tip."""
|
||||
|
||||
def _repo_with_tip(self) -> tuple[str, str]:
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_head(self.root, "MDCPS", "dev")
|
||||
return head, self.root
|
||||
|
||||
def test_local_equal_to_resolved_tip_is_not_stale(self):
|
||||
self._repo_with_tip()
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
self.assertTrue(got["determinable"])
|
||||
self.assertFalse(got["stale"])
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev")
|
||||
self.assertEqual(got["base_remote"], "MDCPS")
|
||||
self.assertEqual(got["base_branch"], "dev")
|
||||
|
||||
def test_local_behind_resolved_tip_is_stale(self):
|
||||
head, _ = self._repo_with_tip()
|
||||
advanced = _advance(self.root, "remote moved on")
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", advanced)
|
||||
_git(self.root, "reset", "--hard", "--quiet", head)
|
||||
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
self.assertTrue(got["determinable"])
|
||||
self.assertTrue(got["stale"])
|
||||
self.assertEqual(got["checkout_head"], head)
|
||||
self.assertEqual(got["remote_tracking_head"], advanced)
|
||||
|
||||
def test_local_divergent_from_resolved_tip_is_stale(self):
|
||||
head, _ = self._repo_with_tip()
|
||||
remote_side = _advance(self.root, "remote side")
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", remote_side)
|
||||
_git(self.root, "reset", "--hard", "--quiet", head)
|
||||
local_side = _advance(self.root, "local side")
|
||||
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
self.assertTrue(got["stale"])
|
||||
self.assertEqual(got["checkout_head"], local_side)
|
||||
self.assertNotEqual(local_side, remote_side)
|
||||
|
||||
|
||||
class TestFailClosed(_RepoCase):
|
||||
"""AC7/AC8: missing remote/ref and ambiguous resolution never guess."""
|
||||
|
||||
def test_missing_remote_fails_closed(self):
|
||||
_make_repo(self.root, remote=None, url=None)
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertFalse(got["proven"])
|
||||
self.assertEqual(got["reason_code"], crr.DENY_NO_IDENTITY_REMOTE)
|
||||
self.assertEqual(got["tracking_refs"], ())
|
||||
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
|
||||
|
||||
def test_missing_tracking_ref_fails_closed(self):
|
||||
_make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
# Remote configured, but nothing has ever been fetched.
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertFalse(got["proven"])
|
||||
self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH)
|
||||
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
|
||||
|
||||
def test_ambiguous_integration_branch_fails_closed(self):
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
# Two candidate integration branches and no recorded remote default.
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_branch(self.root, "MDCPS", "main", head)
|
||||
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertFalse(got["proven"])
|
||||
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH)
|
||||
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
|
||||
|
||||
def test_recorded_remote_head_resolves_otherwise_ambiguous_branches(self):
|
||||
"""Ambiguity is refused only when git records no default."""
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_branch(self.root, "MDCPS", "main", head)
|
||||
_set_remote_head(self.root, "MDCPS", "dev")
|
||||
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertEqual(got["branch"], "dev")
|
||||
self.assertEqual(got["source"], crr.BASE_REF_SOURCE_REMOTE_HEAD)
|
||||
|
||||
def test_ambiguous_identity_remote_fails_closed(self):
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_git(
|
||||
self.root,
|
||||
"remote",
|
||||
"add",
|
||||
"prgs",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_branch(self.root, "prgs", "master", head)
|
||||
|
||||
got = crr.resolve_target_base_ref(self.root)
|
||||
self.assertFalse(got["proven"])
|
||||
self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE)
|
||||
self.assertEqual(got["tracking_refs"], ())
|
||||
|
||||
def test_named_remote_disambiguates(self):
|
||||
"""An explicitly named remote is authoritative and not ambiguous."""
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_git(
|
||||
self.root,
|
||||
"remote",
|
||||
"add",
|
||||
"prgs",
|
||||
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_branch(self.root, "prgs", "master", head)
|
||||
|
||||
got = crr.resolve_target_base_ref(self.root, remote="MDCPS")
|
||||
self.assertTrue(got["proven"], got["reasons"])
|
||||
self.assertEqual(got["remote"], "MDCPS")
|
||||
self.assertEqual(got["branch"], "dev")
|
||||
|
||||
def test_orphan_tracking_ref_from_removed_remote_is_ignored(self):
|
||||
"""The live Gitea-Tools symptom: refs/remotes/origin/* outlives its remote."""
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="prgs",
|
||||
url="https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "prgs", "master", head)
|
||||
_set_remote_head(self.root, "prgs", "master")
|
||||
# An abandoned ref left behind by a remote that no longer exists.
|
||||
_set_remote_branch(self.root, "origin", "master", head)
|
||||
_advance(self.root, "orphan must not be consulted")
|
||||
|
||||
got = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master")
|
||||
self.assertEqual(got["repository_slug"], "Scaled-Tech-Consulting/Gitea-Tools")
|
||||
self.assertNotIn(
|
||||
"target repository identity could not be derived from its git remote",
|
||||
got["reasons"],
|
||||
)
|
||||
|
||||
|
||||
class TestGatingAndReportingAgree(_RepoCase):
|
||||
"""AC9: mutation gating and parity reporting resolve the same target."""
|
||||
|
||||
def test_same_resolved_target_for_gate_and_report(self):
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_head(self.root, "MDCPS", "dev")
|
||||
|
||||
gate = root_checkout_guard.resolve_remote_master_ref_state(self.root)
|
||||
report = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
|
||||
self.assertEqual(gate["remote"], report["base_remote"])
|
||||
self.assertEqual(gate["branch"], report["base_branch"])
|
||||
self.assertEqual(gate["sha"], report["remote_tracking_head"])
|
||||
self.assertIn(
|
||||
gate["ref"],
|
||||
(report["tracking_ref"], f"{gate['remote']}/{gate['branch']}"),
|
||||
)
|
||||
|
||||
def test_both_sides_refuse_the_same_unresolvable_target(self):
|
||||
_make_repo(self.root, remote=None, url=None)
|
||||
|
||||
self.assertIsNone(root_checkout_guard.resolve_remote_master_sha(self.root))
|
||||
report = master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
self.assertIsNone(report["remote_tracking_head"])
|
||||
self.assertFalse(report["stale"])
|
||||
self.assertTrue(report["reasons"])
|
||||
|
||||
|
||||
class TestProductionCallers(unittest.TestCase):
|
||||
"""AC10: every affected production caller supplies/consumes the resolved target."""
|
||||
|
||||
def test_guard_reports_the_ref_it_actually_compared(self):
|
||||
assessment = root_checkout_guard.assess_root_checkout_guard(
|
||||
workspace_path="/tmp/nonexistent-workspace-983",
|
||||
canonical_repo_root="/tmp/nonexistent-root-983",
|
||||
current_branch="dev",
|
||||
head_sha="a" * 40,
|
||||
porcelain_status="",
|
||||
remote_master_sha="b" * 40,
|
||||
remote_master_ref="refs/remotes/MDCPS/dev",
|
||||
)
|
||||
self.assertTrue(assessment["block"])
|
||||
joined = " ".join(assessment["reasons"])
|
||||
self.assertIn("refs/remotes/MDCPS/dev", joined)
|
||||
self.assertNotIn("prgs/master", joined)
|
||||
|
||||
def test_guard_message_without_a_ref_stays_generic(self):
|
||||
assessment = root_checkout_guard.assess_root_checkout_guard(
|
||||
workspace_path="/tmp/nonexistent-workspace-983",
|
||||
canonical_repo_root="/tmp/nonexistent-root-983",
|
||||
current_branch="master",
|
||||
head_sha="a" * 40,
|
||||
porcelain_status="",
|
||||
remote_master_sha="b" * 40,
|
||||
)
|
||||
joined = " ".join(assessment["reasons"])
|
||||
self.assertIn("the tracking integration ref", joined)
|
||||
self.assertNotIn("prgs/master", joined)
|
||||
|
||||
def test_anti_stomp_preflight_forwards_the_resolved_ref(self):
|
||||
sig = inspect.signature(anti_stomp_preflight.assess_anti_stomp_preflight)
|
||||
self.assertIn("remote_master_ref", sig.parameters)
|
||||
src = inspect.getsource(anti_stomp_preflight.assess_anti_stomp_preflight)
|
||||
self.assertIn("remote_master_ref=remote_master_ref", src)
|
||||
|
||||
def test_no_production_caller_inherits_the_prgs_default(self):
|
||||
"""Every resolve site must derive, or pass remote_refs explicitly."""
|
||||
import gitea_mcp_server
|
||||
|
||||
src = inspect.getsource(gitea_mcp_server)
|
||||
# The four historical call sites now consume the resolved-target state.
|
||||
self.assertGreaterEqual(src.count("resolve_remote_master_ref_state("), 4)
|
||||
self.assertNotIn("resolve_remote_master_sha(canonical_root)", src)
|
||||
|
||||
def test_resolver_signature_supports_explicit_remote(self):
|
||||
sig = inspect.signature(root_checkout_guard.resolve_remote_master_sha)
|
||||
self.assertIn("remote", sig.parameters)
|
||||
self.assertIn("remote_refs", sig.parameters)
|
||||
|
||||
|
||||
class TestRepositoryStructureUntouched(_RepoCase):
|
||||
"""Derivation is strictly read-only: it never writes refs or branches."""
|
||||
|
||||
def test_resolution_creates_no_refs_or_branches(self):
|
||||
head = _make_repo(
|
||||
self.root,
|
||||
remote="MDCPS",
|
||||
url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git",
|
||||
)
|
||||
_set_remote_branch(self.root, "MDCPS", "dev", head)
|
||||
_set_remote_head(self.root, "MDCPS", "dev")
|
||||
|
||||
fmt = "--format=%(refname) %(objectname)"
|
||||
before = _git(self.root, "for-each-ref", fmt)
|
||||
before_remotes = _git(self.root, "remote")
|
||||
|
||||
crr.resolve_target_base_ref(self.root)
|
||||
root_checkout_guard.resolve_remote_master_sha(self.root)
|
||||
master_parity_gate.assess_target_repository_parity(
|
||||
canonical_root=self.root, source="test"
|
||||
)
|
||||
|
||||
self.assertEqual(_git(self.root, "for-each-ref", fmt), before)
|
||||
self.assertEqual(_git(self.root, "remote"), before_remotes)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -87,13 +87,27 @@ class TestAssessRootCheckoutGuard(unittest.TestCase):
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("tracked local edits", result["reasons"][0])
|
||||
|
||||
def test_head_behind_prgs_master_blocked(self):
|
||||
def test_head_behind_tracking_base_ref_blocked(self):
|
||||
"""#983: the base ref is derived, so the message no longer hardcodes PRGS."""
|
||||
result = self._assess(
|
||||
head_sha=OTHER_SHA,
|
||||
remote_master_sha=MASTER_SHA,
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("does not match prgs/master", result["reasons"][0])
|
||||
self.assertIn(
|
||||
"does not match the tracking integration ref", result["reasons"][0]
|
||||
)
|
||||
|
||||
def test_head_behind_named_base_ref_reports_that_ref(self):
|
||||
"""The resolved ref is named, so a non-PRGS target is reported accurately."""
|
||||
result = self._assess(
|
||||
head_sha=OTHER_SHA,
|
||||
remote_master_sha=MASTER_SHA,
|
||||
remote_master_ref="refs/remotes/MDCPS/dev",
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("does not match refs/remotes/MDCPS/dev", result["reasons"][0])
|
||||
self.assertNotIn("prgs/master", result["reasons"][0])
|
||||
|
||||
def test_merger_requires_clean_control_checkout(self):
|
||||
result = self._assess(
|
||||
|
||||
Reference in New Issue
Block a user