diff --git a/canonical_repository_root.py b/canonical_repository_root.py index 3b4dd53..9983c2f 100644 --- a/canonical_repository_root.py +++ b/canonical_repository_root.py @@ -37,18 +37,29 @@ 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//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. +# Fallback integration-branch names, probed only when the checkout declares no +# configured upstream (#983). Mirrors the stable base branches recognised +# elsewhere in the workflow (``stacked_pr_support``, ``root_checkout_guard``). +# The fallback is deliberately *not* ordered-first-wins: when more than one of +# these refs exists and the checkout records no upstream, the integration branch +# is genuinely ambiguous and resolution fails closed instead of guessing. INTEGRATION_BRANCH_CANDIDATES: tuple[str, ...] = ("master", "main", "dev") +# Sources for a *proven* base-ref derivation (#983 B1). +# +# ``refs/remotes//HEAD`` is deliberately absent from this list. It is a +# local symbolic-ref *cache* written once at clone time and refreshed only by an +# explicit ``git remote set-head``; an ordinary fetch never updates it. When the +# upstream default branch changes afterwards the cache keeps naming the old +# branch, so trusting it derives the wrong integration branch for a checkout +# that is sitting exactly on its tip. The cache is still read, but only as a +# corroborating observation reported back to the caller — never as an authority, +# and never as a tie-breaker between otherwise ambiguous candidates. +BASE_REF_SOURCE_CONFIGURED_UPSTREAM = "configured_branch_upstream" +BASE_REF_SOURCE_UNIQUE_CANDIDATE = "unique_integration_branch_ref" + # 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" @@ -190,6 +201,153 @@ def _configured_remote_identities(path: str, remote: str | None) -> list[tuple[s return found +def _configured_branch_upstream(path: str) -> tuple[str | None, str | None]: + """``(remote_name, branch)`` the checked-out branch is configured to track. + + Reads ``branch..remote`` and ``branch..merge`` — the + checkout's own explicitly configured integration target, equivalent to + ``@{upstream}``. Unlike ``refs/remotes//HEAD`` this is not a + clone-time cache: it is written when the branch is set up to track an + upstream and rewritten whenever that tracking changes, so it states what the + checkout actually integrates onto today (#983 B1). + + Returns ``(None, None)`` on a detached HEAD or an untracked branch. Names are + returned verbatim; git remote names are case-sensitive. + """ + text = (path or "").strip() + if not text: + return None, None + + branch = _git_read(text, "symbolic-ref", "--quiet", "--short", "HEAD") + if not branch: + return None, None + + remote = _git_read(text, "config", "--get", f"branch.{branch}.remote") + merge = _git_read(text, "config", "--get", f"branch.{branch}.merge") + if not remote or not merge: + return None, None + + prefix = "refs/heads/" + upstream_branch = merge[len(prefix):].strip() if merge.startswith(prefix) else merge.strip() + if not upstream_branch: + return None, None + return remote, upstream_branch + + +def _cached_remote_head_branch(path: str, remote: str) -> str | None: + """Branch named by the *cached* ``refs/remotes//HEAD`` symref. + + Read for observability only. This value is never authoritative (see + :data:`BASE_REF_SOURCE_CONFIGURED_UPSTREAM`); it is surfaced so an operator + can see that the local cache disagrees with the configured upstream, and so + a refusal can name the misleading signal explicitly. + """ + prefix = f"refs/remotes/{remote}/" + symref = _git_read(path, "symbolic-ref", "--quiet", f"{prefix}HEAD") + if not symref or not symref.startswith(prefix): + return None + branch = symref[len(prefix):].strip() + return branch or None + + +def assess_identity_remote( + path: str, *, explicit_remote: str | None = None +) -> dict: + """Resolve the identity remote, failing closed when the target is ambiguous. + + This is the ambiguity-aware counterpart to :func:`resolve_identity_remote`, + and the reason gating and reporting can no longer disagree (#983 B2). + + *explicit_remote* is a **caller-supplied disambiguation** and nothing else. + It must come from an operator or an explicitly sanctioned repository + context; a value this module inferred while probing must never be handed + back in through it, because doing so re-labels an internal first-wins guess + as deliberate caller intent and silently suppresses the ambiguity gate. + + Decision table: + + * No remote yields a parseable identity -> :data:`DENY_NO_IDENTITY_REMOTE`. + * *explicit_remote* names one of the resolving remotes -> that remote is + authoritative, ``explicit`` True. + * Otherwise, when every resolving remote claims the **same** repository the + target is unambiguous and is accepted, ``explicit`` False. The remote + named by the checkout's configured upstream is preferred among equals so + the choice is deterministic rather than probe-order dependent. + * Otherwise distinct remotes claim different repositories and there is no + sanctioned disambiguation -> :data:`DENY_AMBIGUOUS_REMOTE`. + + Returns a dict with ``remote``, ``slug``, ``identities``, ``ambiguous``, + ``explicit``, ``reason_code`` and ``reasons``. ``remote``/``slug`` are None + on any refusal, so a caller cannot report a repository the gate refuses. + """ + text = (path or "").strip() + result: dict = { + "remote": None, + "slug": None, + "identities": [], + "ambiguous": False, + "explicit": False, + "reason_code": None, + "reasons": [], + } + if not text: + result["reason_code"] = DENY_NO_IDENTITY_REMOTE + result["reasons"].append( + "no repository path supplied for identity-remote resolution (fail closed)" + ) + return result + + named = (explicit_remote or "").strip() or None + identities = _configured_remote_identities(text, named) + result["identities"] = list(identities) + if not identities: + result["reason_code"] = DENY_NO_IDENTITY_REMOTE + result["reasons"].append( + f"no git remote at '{text}' yields a parseable repository identity " + "(fail closed)" + ) + return result + + if named: + for name, slug in identities: + if name == named: + result["remote"] = name + result["slug"] = slug + result["explicit"] = True + return result + + distinct = {slug for _, slug in identities} + if len(distinct) > 1: + listed = ", ".join(f"{name} -> {slug}" for name, slug in identities) + result["ambiguous"] = True + result["reason_code"] = DENY_AMBIGUOUS_REMOTE + detail = ( + f"explicitly named remote '{named}' does not resolve a repository identity " + "there, so it cannot disambiguate; " + if named + else "" + ) + result["reasons"].append( + f"ambiguous repository identity at '{text}': {detail}remotes resolve to " + f"different repositories ({listed}); no single authoritative target can " + "be established (fail closed)" + ) + return result + + # One repository, possibly reachable through several remote names (a mirror). + # Prefer the remote the checkout is actually configured to track so the + # choice is deterministic instead of probe-order dependent. + upstream_remote, _ = _configured_branch_upstream(text) + chosen = identities[0] + if upstream_remote: + for entry in identities: + if entry[0] == upstream_remote: + chosen = entry + break + result["remote"], result["slug"] = chosen + return result + + def resolve_identity_remote( path: str, *, remote: str | None = None ) -> tuple[str | None, str | None]: @@ -201,6 +359,14 @@ def resolve_identity_remote( name to build ``refs/remotes//``, so it is now returned rather than discarded (#983). Returns ``(None, None)`` when no remote URL is parseable (identity cannot be proven). + + First-wins by design: this is the identity lookup behind + :func:`repository_identity_slug` and the #706/#973 canonical-root + validation, which compare an observed slug against an independently trusted + expected slug and therefore do not need an ambiguity verdict. Callers that + *derive* a target rather than validate one — the mutation guard and the + parity report — must use :func:`assess_identity_remote`, which fails closed + on ambiguity (#983 B2). """ text = (path or "").strip() if not text: @@ -229,6 +395,10 @@ def _base_ref_result( repository_slug: str | None = None, source: str | None = None, reason_code: str | None = None, + identity_explicit: bool = False, + configured_upstream_remote: str | None = None, + configured_upstream_branch: str | None = None, + cached_remote_head_branch: str | None = None, ) -> dict: """Build the base-ref derivation payload. @@ -237,6 +407,12 @@ def _base_ref_result( qualified ``refs/remotes//``. It is empty whenever the derivation is not ``proven``, so an unresolved target can never be probed against some other repository's ref. + + ``cached_remote_head_branch`` reports what the local + ``refs/remotes//HEAD`` cache claims, and + ``cached_remote_head_conflicts`` whether that claim disagrees with the branch + actually derived. Both are observability only: the cache never decides the + outcome (#983 B1). """ tracking_ref = f"refs/remotes/{remote}/{branch}" if proven and remote and branch else None tracking_refs: tuple[str, ...] = ( @@ -252,10 +428,33 @@ def _base_ref_result( "tracking_refs": tracking_refs, "source": source, "reason_code": reason_code, + "identity_explicit": identity_explicit, + "configured_upstream_remote": configured_upstream_remote, + "configured_upstream_branch": configured_upstream_branch, + "cached_remote_head_branch": cached_remote_head_branch, + "cached_remote_head_conflicts": bool( + cached_remote_head_branch and branch and cached_remote_head_branch != branch + ), "reasons": list(reasons), } +def _git_read(path: str, *args: str) -> str | None: + """Run a read-only git command in *path*; ``None`` on any failure.""" + try: + res = subprocess.run( + ["git", "-C", path, *args], + capture_output=True, + text=True, + check=False, + ) + except Exception: + return None + if res.returncode != 0: + return None + return (res.stdout or "").strip() or None + + def _ref_exists(path: str, ref: str) -> bool: """Whether *ref* resolves in the checkout at *path*.""" try: @@ -270,30 +469,45 @@ def _ref_exists(path: str, ref: str) -> bool: return res.returncode == 0 and bool((res.stdout or "").strip()) -def resolve_target_base_ref(path: str, *, remote: str | None = None) -> dict: +def resolve_target_base_ref(path: str, *, explicit_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). + *explicit_remote* is a caller-supplied disambiguation only; see + :func:`assess_identity_remote` for why an internally inferred remote must + never be passed back in here (#983 B2). + 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//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. + 1. **Identity remote** — via :func:`assess_identity_remote`, with its exact + configured case preserved (``MDCPS`` stays ``MDCPS``). Ambiguous identity + fails closed here rather than resolving to whichever remote probed first. + 2. **The checkout's configured upstream** — ``branch..remote`` plus + ``branch..merge``, accepted when it names the identity remote + and its remote-tracking ref actually exists. This is the checkout's own + declaration of what it integrates onto, and unlike the remote-HEAD cache + it is rewritten whenever that tracking changes. + 3. **Fallback** — only when the checkout declares no usable upstream, + exactly one of :data:`INTEGRATION_BRANCH_CANDIDATES` present as a + remote-tracking ref. + + ``refs/remotes//HEAD`` is **not** a step. It is read for reporting + (``cached_remote_head_branch`` / ``cached_remote_head_conflicts``) and never + decides the branch, because it is a clone-time cache that an ordinary fetch + does not refresh: a checkout whose upstream default moved on still has the + old branch cached, and trusting it gates that checkout against a ref it does + not integrate onto (#983 B1). 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. + configured upstream, or when no candidate exists at all. Nothing here + invents a branch, writes a ref, runs a fetch, or falls back to another + repository's base. Every ``proven`` result names a tracking ref that + resolves in this checkout. """ text = (path or "").strip() if not text: @@ -303,62 +517,45 @@ def resolve_target_base_ref(path: str, *, remote: str | None = None) -> dict: reason_code=DENY_NO_IDENTITY_REMOTE, ) - identities = _configured_remote_identities(text, remote) - if not identities: + identity = assess_identity_remote(text, explicit_remote=explicit_remote) + remote_name, slug = identity["remote"], identity["slug"] + if not remote_name: 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, + reasons=list(identity["reasons"]), + reason_code=identity["reason_code"], + identity_explicit=bool(identity["explicit"]), ) - # 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, - ) + cached = _cached_remote_head_branch(text, remote_name) + upstream_remote, upstream_branch = _configured_branch_upstream(text) + common = { + "repository_slug": slug, + "identity_explicit": bool(identity["explicit"]), + "configured_upstream_remote": upstream_remote, + "configured_upstream_branch": upstream_branch, + "cached_remote_head_branch": cached, + } + # 2. The checkout's configured upstream, when it belongs to the identity + # remote and its tracking ref is actually present. Requiring the ref to + # exist keeps 'proven' honest: a proven target is always resolvable. + if ( + upstream_remote == remote_name + and upstream_branch + and _ref_exists(text, f"{prefix}{upstream_branch}") + ): + return _base_ref_result( + proven=True, + reasons=[], + remote=remote_name, + branch=upstream_branch, + source=BASE_REF_SOURCE_CONFIGURED_UPSTREAM, + **common, + ) + + # 3. Exactly one known integration branch present as a remote-tracking ref. present = [ candidate for candidate in INTEGRATION_BRANCH_CANDIDATES @@ -370,28 +567,44 @@ def resolve_target_base_ref(path: str, *, remote: str | None = None) -> dict: reasons=[], remote=remote_name, branch=present[0], - repository_slug=slug, source=BASE_REF_SOURCE_UNIQUE_CANDIDATE, + **common, ) + + # The cached remote HEAD is named in the refusal so the operator can see the + # signal that looks authoritative but is not, and is told the read-only fix. + cache_note = ( + f" the cached '{prefix}HEAD' names '{cached}', but that cache is written at " + "clone time and is not refreshed by fetch, so it cannot break the tie;" + if cached + else "" + ) + remedy = ( + f" Configure the checkout's upstream (git branch --set-upstream-to={remote_name}/" + ") so the integration target is declared rather than guessed." + ) 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)" + f"the checkout at '{text}' declares no upstream on remote " + f"'{remote_name}' and several integration branches exist " + f"({', '.join(present)});{cache_note} the integration base ref is " + f"ambiguous (fail closed).{remedy}" ], reason_code=DENY_AMBIGUOUS_BASE_BRANCH, + **common, ) 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)" + f"the checkout at '{text}' declares no upstream on remote '{remote_name}' " + f"and none of {'/'.join(INTEGRATION_BRANCH_CANDIDATES)} exists as a " + f"remote-tracking ref under '{prefix}';{cache_note} the integration base " + f"ref cannot be derived (fail closed; no fetch is performed here).{remedy}" ], reason_code=DENY_NO_BASE_BRANCH, + **common, ) diff --git a/master_parity_gate.py b/master_parity_gate.py index 91a60e8..a55faf3 100644 --- a/master_parity_gate.py +++ b/master_parity_gate.py @@ -445,6 +445,7 @@ def assess_target_repository_parity( "remote_tracking_head": None, "determinable": False, "stale": False, + "reason_code": None, "reasons": [], } if not canonical_root: @@ -484,19 +485,38 @@ def assess_target_repository_parity( # '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" + # #983 B2: this must be the *ambiguity-aware* resolver. The first-wins + # `resolve_identity_remote` picks whichever remote probes first, so on a + # target where distinct remotes claim different repositories the report + # confidently named one of them while the mutation gate refused the same + # target — gating and reporting evaluating different repositories, which is + # precisely the divergence this issue exists to end. + identity = _crr.assess_identity_remote(canonical_root) + result["repository_slug"] = identity["slug"] + if not identity["slug"]: + result["reason_code"] = identity["reason_code"] + result["reasons"].extend( + identity["reasons"] + or ["target repository identity could not be derived from its git remote"] ) + if identity["ambiguous"]: + # An ambiguous target has no single authoritative base, so reporting one + # would be a guess. Fail closed here exactly as the gate does. + return result + # Identity is otherwise 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. if not tracking_ref: - base = _crr.resolve_target_base_ref(canonical_root, remote=identity_remote) + # No remote argument. The identity remote resolved just above was + # *inferred here*, and feeding it back in would tell the resolver a + # caller had explicitly disambiguated the repository, suppressing its + # ambiguity gate (#983 B2). Only an operator-supplied remote may do that, + # and this call site has none. + base = _crr.resolve_target_base_ref(canonical_root) if not base.get("proven"): + result["reason_code"] = base.get("reason_code") result["reasons"].extend(base.get("reasons") or []) return result tracking_ref = base["tracking_ref"] diff --git a/root_checkout_guard.py b/root_checkout_guard.py index 588252e..bac50e9 100644 --- a/root_checkout_guard.py +++ b/root_checkout_guard.py @@ -36,13 +36,15 @@ BASE_BRANCHES = frozenset({"master", "main", "dev"}) REMOTE_MASTER_REFS = ("prgs/master", "refs/remotes/prgs/master") -def _derive_probe_refs(root: str, remote: str | None) -> dict: +def _derive_probe_refs(root: str, explicit_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) + derived = canonical_repository_root.resolve_target_base_ref( + root, explicit_remote=explicit_remote + ) return { "refs": tuple(derived.get("tracking_refs") or ()), "remote": derived.get("remote"), @@ -51,6 +53,8 @@ def _derive_probe_refs(root: str, remote: str | None) -> dict: "proven": bool(derived.get("proven")), "reason_code": derived.get("reason_code"), "reasons": list(derived.get("reasons") or []), + "cached_remote_head_branch": derived.get("cached_remote_head_branch"), + "cached_remote_head_conflicts": bool(derived.get("cached_remote_head_conflicts")), } @@ -58,7 +62,7 @@ def resolve_remote_master_sha( canonical_repo_root: str, *, remote_refs: tuple[str, ...] | None = None, - remote: str | None = None, + explicit_remote: str | None = None, ) -> str | None: """Return the commit SHA for the tracking integration ref when available. @@ -67,6 +71,10 @@ def resolve_remote_master_sha( 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". + + *explicit_remote* is a caller-supplied disambiguation and is named that way + deliberately: an internally inferred remote handed back in would suppress the + resolver's ambiguity gate (#983 B2). No production caller supplies it. """ root = (canonical_repo_root or "").strip() if not root: @@ -74,7 +82,7 @@ def resolve_remote_master_sha( if remote_refs: probe: tuple[str, ...] = tuple(remote_refs) else: - derived = _derive_probe_refs(root, remote) + derived = _derive_probe_refs(root, explicit_remote) if not derived["proven"]: return None probe = derived["refs"] @@ -96,7 +104,7 @@ def resolve_remote_master_ref_state( canonical_repo_root: str, *, remote_refs: tuple[str, ...] | None = None, - remote: str | None = None, + explicit_remote: str | None = None, ) -> dict: """Resolve the tracking integration ref together with its commit SHA. @@ -119,6 +127,8 @@ def resolve_remote_master_ref_state( "source": None, "reason_code": None, "reasons": [], + "cached_remote_head_branch": None, + "cached_remote_head_conflicts": False, } if not root: state["reasons"].append("no canonical repository root supplied (fail closed)") @@ -128,17 +138,19 @@ def resolve_remote_master_ref_state( probe: tuple[str, ...] = tuple(remote_refs) state["source"] = "explicit_remote_refs" else: - derived = _derive_probe_refs(root, remote) + derived = _derive_probe_refs(root, explicit_remote) state["remote"] = derived["remote"] state["branch"] = derived["branch"] state["source"] = derived["source"] + state["cached_remote_head_branch"] = derived["cached_remote_head_branch"] + state["cached_remote_head_conflicts"] = derived["cached_remote_head_conflicts"] 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) + sha = resolve_remote_master_sha(root, remote_refs=probe, explicit_remote=explicit_remote) if not sha: state["reasons"].append( "tracking integration ref " diff --git a/tests/test_issue_983_cross_repo_base_ref.py b/tests/test_issue_983_cross_repo_base_ref.py index 5e97c61..27c3665 100644 --- a/tests/test_issue_983_cross_repo_base_ref.py +++ b/tests/test_issue_983_cross_repo_base_ref.py @@ -5,6 +5,22 @@ 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. +Two further defects were found by review at head ``2d5d5c9d`` and are covered +here: + +* **B1** — the first fix derived the integration branch from + ``refs/remotes//HEAD``. That symref is a *local cache* written at clone + time and never refreshed by fetch, so on the real Weekly Briefings target it + still named ``main`` while the checkout tracked and sat exactly on ``dev``. The + authoritative signal is the checkout's own configured upstream. The fixtures + below therefore reproduce the **disagreement**: the cache says ``main``, the + configured upstream says ``dev``, and ``dev`` must win. +* **B2** — the parity report passed its internally inferred identity remote back + into the resolver, which reads a caller-supplied remote as "the caller already + disambiguated" and skips its ambiguity gate. Gating and reporting could then + evaluate different repositories. An inferred remote is now never laundered into + explicit caller intent, and ambiguity fails closed on both sides. + 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 @@ -24,6 +40,11 @@ import canonical_repository_root as crr import master_parity_gate import root_checkout_guard +MDCPS_URL = "https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git" +MDCPS_SLUG = "MDCPS/WeeklyBriefings-Meta" +PRGS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" +PRGS_SLUG = "Scaled-Tech-Consulting/Gitea-Tools" + def _git(root: str, *args: str) -> str: res = subprocess.run( @@ -57,6 +78,11 @@ def _set_remote_branch(root: str, remote: str, branch: str, sha: str) -> None: def _set_remote_head(root: str, remote: str, branch: str) -> None: + """Write the *cached* refs/remotes//HEAD symref. + + This is the signal B1 proved untrustworthy. Fixtures use it to reproduce a + stale cache, never to manufacture the answer under test. + """ _git( root, "symbolic-ref", @@ -65,6 +91,22 @@ def _set_remote_head(root: str, remote: str, branch: str) -> None: ) +def _set_upstream(root: str, remote: str, branch: str) -> None: + """Configure the current branch's upstream, exactly as git tracking does. + + Writes ``branch..remote`` / ``branch..merge`` directly + rather than via ``--set-upstream-to`` so no ref is required to pre-exist and + no network is touched. + """ + current = _git(root, "symbolic-ref", "--short", "HEAD") + _git(root, "config", f"branch.{current}.remote", remote) + _git(root, "config", f"branch.{current}.merge", f"refs/heads/{branch}") + + +def _checkout_new_branch(root: str, branch: str) -> None: + _git(root, "checkout", "--quiet", "-b", 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") @@ -79,34 +121,61 @@ class _RepoCase(unittest.TestCase): self.addCleanup(self._tmp.cleanup) self.root = os.path.join(self._tmp.name, "repo") + def _weekly_briefings_shape(self) -> str: + """The real Weekly Briefings target, including its stale cache. + + Remote ``MDCPS``; checked out on ``dev``; upstream configured to + ``MDCPS/dev``; both ``dev`` and ``main`` present as tracking refs; and + ``refs/remotes/MDCPS/HEAD`` still cached at ``main`` from clone time. + """ + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _checkout_new_branch(self.root, "dev") + _set_remote_branch(self.root, "MDCPS", "dev", head) + stale = _advance(self.root, "main diverged long ago") + _set_remote_branch(self.root, "MDCPS", "main", stale) + _git(self.root, "reset", "--hard", "--quiet", head) + _set_remote_head(self.root, "MDCPS", "main") # stale clone-time cache + _set_upstream(self.root, "MDCPS", "dev") # authoritative + return head + class TestPrgsBehaviourPreserved(_RepoCase): - """AC1: existing PRGS behaviour using prgs/master is unchanged.""" + """Required coverage 1: PRGS prgs/master compatibility.""" - 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", - ) + def _prgs(self) -> str: + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) _set_remote_branch(self.root, "prgs", "master", head) _set_remote_head(self.root, "prgs", "master") + _set_upstream(self.root, "prgs", "master") + return head + + def test_prgs_master_resolves_unchanged(self): + head = self._prgs() 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(got["repository_slug"], PRGS_SLUG) + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM) + # Cache and upstream agree here, which is the ordinary PRGS state. + self.assertFalse(got["cached_remote_head_conflicts"]) self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) + def test_prgs_resolves_without_a_configured_upstream(self): + """A PRGS checkout with no tracking config still resolves master.""" + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) + _set_remote_branch(self.root, "prgs", "master", head) + + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["tracking_ref"], "refs/remotes/prgs/master") + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE) + 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", - ) + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) _set_remote_branch(self.root, "prgs", "master", head) state = root_checkout_guard.resolve_remote_master_ref_state( @@ -116,38 +185,124 @@ class TestPrgsBehaviourPreserved(_RepoCase): self.assertEqual(state["source"], "explicit_remote_refs") -class TestCrossRepositoryTarget(_RepoCase): - """AC2/AC3/AC4: MDCPS/dev, no origin remote, exact remote-name case.""" +class TestStaleCachedRemoteHead(_RepoCase): + """Required coverage 3: configured upstream MDCPS/dev vs stale cache -> main. - def _mdcps(self) -> str: - head = _make_repo( - self.root, - remote="MDCPS", - url="https://gitea.example.net/MDCPS/WeeklyBriefings-Meta.git", + This is B1. The fixture deliberately does **not** point + ``refs/remotes/MDCPS/HEAD`` at ``dev``; it reproduces the disagreement that + was live on ``/Users/jasonwalker/Development/weekly-briefings``. + """ + + def test_configured_upstream_beats_stale_cached_remote_head(self): + head = self._weekly_briefings_shape() + + # Precondition: the fixture really is in the defective state. + self.assertEqual( + _git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"), + "refs/remotes/MDCPS/main", ) + self.assertEqual(_git(self.root, "config", "--get", "branch.dev.merge"), "refs/heads/dev") + self.assertNotEqual( + _git(self.root, "rev-parse", "refs/remotes/MDCPS/main"), + _git(self.root, "rev-parse", "refs/remotes/MDCPS/dev"), + ) + + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["branch"], "dev") + self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertEqual(got["source"], crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM) + # The stale cache is reported, never obeyed. + self.assertEqual(got["cached_remote_head_branch"], "main") + self.assertTrue(got["cached_remote_head_conflicts"]) + self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) + + def test_checkout_on_its_integration_tip_is_not_blocked(self): + """The live symptom: a checkout exactly on its tip was reported stale.""" + head = self._weekly_briefings_shape() + + state = root_checkout_guard.resolve_remote_master_ref_state(self.root) + self.assertEqual(state["sha"], head) + self.assertTrue(state["cached_remote_head_conflicts"]) + + assessment = root_checkout_guard.assess_root_checkout_guard( + workspace_path=self.root, + canonical_repo_root=self.root, + current_branch="dev", + head_sha=head, + porcelain_status="", + remote_master_sha=state["sha"], + remote_master_ref=state["ref"], + ) + self.assertTrue(assessment["proven"], assessment["reasons"]) + + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertFalse(report["stale"]) + self.assertEqual(report["base_branch"], "dev") + self.assertEqual(report["reasons"], []) + + def test_cached_remote_head_alone_never_proves_a_target(self): + """With no configured upstream, the cache cannot supply the branch.""" + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + # Only a non-candidate branch exists, and only the cache names it. + _set_remote_branch(self.root, "MDCPS", "trunk", head) + _set_remote_head(self.root, "MDCPS", "trunk") + + got = crr.resolve_target_base_ref(self.root) + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_NO_BASE_BRANCH) + self.assertEqual(got["tracking_refs"], ()) + self.assertEqual(got["cached_remote_head_branch"], "trunk") + # The refusal names the misleading signal so an operator is not sent + # chasing a ref that looks authoritative. + self.assertIn("trunk", " ".join(got["reasons"])) + + def test_cached_remote_head_never_breaks_a_tie(self): + """Two candidates, no upstream: the cache must not decide.""" + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "MDCPS", "main", head) _set_remote_head(self.root, "MDCPS", "dev") - return 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_no_proven_source_is_the_remote_head_cache(self): + """Structural guard: the cache is not in the set of proving sources.""" + sources = { + crr.BASE_REF_SOURCE_CONFIGURED_UPSTREAM, + crr.BASE_REF_SOURCE_UNIQUE_CANDIDATE, + } + self.assertNotIn("remote_head_symref", sources) + self.assertFalse(hasattr(crr, "BASE_REF_SOURCE_REMOTE_HEAD")) + + +class TestCrossRepositoryTarget(_RepoCase): + """Required coverage 2/4/5: MDCPS/dev, no origin remote, exact case.""" def test_mdcps_dev_resolves(self): - head = self._mdcps() + head = self._weekly_briefings_shape() 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(got["repository_slug"], MDCPS_SLUG) self.assertEqual(root_checkout_guard.resolve_remote_master_sha(self.root), head) def test_no_remote_named_origin(self): - self._mdcps() + self._weekly_briefings_shape() 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() + self._weekly_briefings_shape() got = crr.resolve_target_base_ref(self.root) self.assertEqual(got["remote"], "MDCPS") self.assertNotEqual(got["remote"], "mdcps") @@ -169,7 +324,7 @@ class TestCrossRepositoryTarget(_RepoCase): 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() + self._weekly_briefings_shape() res = subprocess.run( ["git", "-C", self.root, "remote", "get-url", "mdcps"], capture_output=True, @@ -178,32 +333,24 @@ class TestCrossRepositoryTarget(_RepoCase): ) 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") + # A caller naming the wrong case cannot disambiguate, but the target is + # unambiguous anyway, so the exact-case name is still resolved. + got = crr.resolve_target_base_ref(self.root, explicit_remote="mdcps") self.assertTrue(got["proven"], got["reasons"]) self.assertEqual(got["remote"], "MDCPS") self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") + self.assertFalse(got["identity_explicit"], "a non-matching name is not explicit intent") name, slug = crr.resolve_identity_remote(self.root) self.assertEqual(name, "MDCPS") - self.assertEqual(slug, "MDCPS/WeeklyBriefings-Meta") + self.assertEqual(slug, MDCPS_SLUG) 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 + """Required coverage 6/7: matching tip, and behind or divergent checkout.""" def test_local_equal_to_resolved_tip_is_not_stale(self): - self._repo_with_tip() + self._weekly_briefings_shape() got = master_parity_gate.assess_target_repository_parity( canonical_root=self.root, source="test" ) @@ -212,9 +359,10 @@ class TestTargetStaleness(_RepoCase): self.assertEqual(got["tracking_ref"], "refs/remotes/MDCPS/dev") self.assertEqual(got["base_remote"], "MDCPS") self.assertEqual(got["base_branch"], "dev") + self.assertIsNone(got["reason_code"]) def test_local_behind_resolved_tip_is_stale(self): - head, _ = self._repo_with_tip() + head = self._weekly_briefings_shape() advanced = _advance(self.root, "remote moved on") _set_remote_branch(self.root, "MDCPS", "dev", advanced) _git(self.root, "reset", "--hard", "--quiet", head) @@ -228,7 +376,7 @@ class TestTargetStaleness(_RepoCase): self.assertEqual(got["remote_tracking_head"], advanced) def test_local_divergent_from_resolved_tip_is_stale(self): - head, _ = self._repo_with_tip() + head = self._weekly_briefings_shape() remote_side = _advance(self.root, "remote side") _set_remote_branch(self.root, "MDCPS", "dev", remote_side) _git(self.root, "reset", "--hard", "--quiet", head) @@ -243,7 +391,7 @@ class TestTargetStaleness(_RepoCase): class TestFailClosed(_RepoCase): - """AC7/AC8: missing remote/ref and ambiguous resolution never guess.""" + """Required coverage 8/9: missing remote or ref, and ambiguous targets.""" def test_missing_remote_fails_closed(self): _make_repo(self.root, remote=None, url=None) @@ -254,24 +402,33 @@ class TestFailClosed(_RepoCase): 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", - ) + _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) # 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_configured_upstream_without_a_tracking_ref_fails_closed(self): + """A proven target always names a ref that resolves.""" + _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _set_upstream(self.root, "MDCPS", "dev") # declared, never 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(got["tracking_ref"]) + + def test_every_proven_target_resolves(self): + """No 'proven' result may name an unresolvable tracking ref.""" + head = self._weekly_briefings_shape() + got = crr.resolve_target_base_ref(self.root) + self.assertTrue(got["proven"]) + self.assertEqual(_git(self.root, "rev-parse", got["tracking_ref"]), head) + 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. + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + # Two candidate integration branches and no configured upstream. _set_remote_branch(self.root, "MDCPS", "dev", head) _set_remote_branch(self.root, "MDCPS", "main", head) @@ -280,35 +437,9 @@ class TestFailClosed(_RepoCase): 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", - ) + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) _set_remote_branch(self.root, "MDCPS", "dev", head) _set_remote_branch(self.root, "prgs", "master", head) @@ -317,37 +448,11 @@ class TestFailClosed(_RepoCase): 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", - ) + head = _make_repo(self.root, remote="prgs", url=PRGS_URL) _set_remote_branch(self.root, "prgs", "master", head) - _set_remote_head(self.root, "prgs", "master") + _set_upstream(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") @@ -356,24 +461,82 @@ class TestFailClosed(_RepoCase): 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.assertEqual(got["repository_slug"], PRGS_SLUG) self.assertNotIn( "target repository identity could not be derived from its git remote", got["reasons"], ) +class TestExplicitVersusInferredRemote(_RepoCase): + """Required coverage 11: explicit disambiguation stays distinct from inference. + + This is B2. A remote the module inferred while probing must never re-enter + the resolver as though an operator had named it. + """ + + def _two_remotes(self) -> str: + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "prgs", "master", head) + return head + + def test_explicit_remote_disambiguates(self): + self._two_remotes() + got = crr.resolve_target_base_ref(self.root, explicit_remote="MDCPS") + self.assertTrue(got["proven"], got["reasons"]) + self.assertEqual(got["remote"], "MDCPS") + self.assertEqual(got["branch"], "dev") + self.assertTrue(got["identity_explicit"]) + + def test_inferred_remote_does_not_disambiguate(self): + """Feeding the inferred remote back in must not unlock the target.""" + self._two_remotes() + inferred, _ = crr.resolve_identity_remote(self.root) + self.assertIsNotNone(inferred, "the first-wins probe still returns a name") + + # The report infers internally and must still refuse. + report = master_parity_gate.assess_target_repository_parity( + canonical_root=self.root, source="test" + ) + self.assertEqual(report["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + self.assertIsNone(report["repository_slug"]) + self.assertIsNone(report["tracking_ref"]) + self.assertFalse(report["stale"]) + self.assertTrue(report["reasons"]) + + def test_report_never_passes_a_remote_into_the_resolver(self): + """Structural guard against the exact B2 regression.""" + src = inspect.getsource(master_parity_gate.assess_target_repository_parity) + self.assertIn("resolve_target_base_ref(canonical_root)", src) + self.assertNotIn("resolve_target_base_ref(canonical_root, remote=", src) + self.assertNotIn("explicit_remote=identity", src) + # Reporting must use the ambiguity-aware identity resolver. + self.assertIn("assess_identity_remote(canonical_root)", src) + + def test_explicit_parameter_is_named_for_its_meaning(self): + for fn in ( + crr.resolve_target_base_ref, + root_checkout_guard.resolve_remote_master_sha, + root_checkout_guard.resolve_remote_master_ref_state, + ): + params = inspect.signature(fn).parameters + self.assertIn("explicit_remote", params, fn.__name__) + self.assertNotIn("remote", params, fn.__name__) + + def test_unmatched_explicit_remote_cannot_unlock_an_ambiguous_target(self): + self._two_remotes() + got = crr.resolve_target_base_ref(self.root, explicit_remote="nonexistent") + self.assertFalse(got["proven"]) + self.assertEqual(got["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + + class TestGatingAndReportingAgree(_RepoCase): - """AC9: mutation gating and parity reporting resolve the same target.""" + """Required coverage 10: guard and parity report make identical decisions.""" 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") + self._weekly_briefings_shape() gate = root_checkout_guard.resolve_remote_master_ref_state(self.root) report = master_parity_gate.assess_target_repository_parity( @@ -391,17 +554,52 @@ class TestGatingAndReportingAgree(_RepoCase): 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)) + 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.assertIsNone(gate["sha"]) self.assertIsNone(report["remote_tracking_head"]) self.assertFalse(report["stale"]) self.assertTrue(report["reasons"]) + self.assertEqual(gate["reason_code"], report["reason_code"]) + + def test_both_sides_refuse_the_same_ambiguous_target(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _git(self.root, "remote", "add", "prgs", PRGS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "prgs", "master", head) + + 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.assertIsNone(gate["sha"]) + self.assertEqual(gate["reason_code"], crr.DENY_AMBIGUOUS_REMOTE) + # The report must not name a repository the gate refuses to act on. + self.assertEqual(report["reason_code"], gate["reason_code"]) + self.assertIsNone(report["repository_slug"]) + self.assertIsNone(report["base_remote"]) + self.assertIsNone(report["base_branch"]) + self.assertFalse(report["stale"]) + + def test_both_sides_refuse_the_same_ambiguous_branch(self): + head = _make_repo(self.root, remote="MDCPS", url=MDCPS_URL) + _set_remote_branch(self.root, "MDCPS", "dev", head) + _set_remote_branch(self.root, "MDCPS", "main", head) + + 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["reason_code"], crr.DENY_AMBIGUOUS_BASE_BRANCH) + self.assertEqual(report["reason_code"], gate["reason_code"]) + self.assertIsNone(report["tracking_ref"]) class TestProductionCallers(unittest.TestCase): - """AC10: every affected production caller supplies/consumes the resolved target.""" + """Required coverage 13: every production caller consumes the same target.""" def test_guard_reports_the_ref_it_actually_compared(self): assessment = root_checkout_guard.assess_root_checkout_guard( @@ -446,36 +644,48 @@ class TestProductionCallers(unittest.TestCase): 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) + def test_no_production_caller_supplies_an_explicit_remote(self): + """Nothing in production may suppress the ambiguity gate (#983 B2).""" + import gitea_mcp_server + + for module in (gitea_mcp_server, anti_stomp_preflight, master_parity_gate): + src = inspect.getsource(module) + self.assertNotIn("explicit_remote=", src, module.__name__) class TestRepositoryStructureUntouched(_RepoCase): - """Derivation is strictly read-only: it never writes refs or branches.""" + """Required coverage 12: resolution mutates no ref, remote, config, or checkout.""" 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") + self._weekly_briefings_shape() fmt = "--format=%(refname) %(objectname)" - before = _git(self.root, "for-each-ref", fmt) + before_refs = _git(self.root, "for-each-ref", fmt) before_remotes = _git(self.root, "remote") + before_config = _git(self.root, "config", "--local", "--list") + before_head = _git(self.root, "rev-parse", "HEAD") + before_branch = _git(self.root, "symbolic-ref", "--short", "HEAD") + before_status = _git(self.root, "status", "--porcelain", "--untracked-files=all") + before_symref = _git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD") crr.resolve_target_base_ref(self.root) + crr.assess_identity_remote(self.root) root_checkout_guard.resolve_remote_master_sha(self.root) + root_checkout_guard.resolve_remote_master_ref_state(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, "for-each-ref", fmt), before_refs) self.assertEqual(_git(self.root, "remote"), before_remotes) + self.assertEqual(_git(self.root, "config", "--local", "--list"), before_config) + self.assertEqual(_git(self.root, "rev-parse", "HEAD"), before_head) + self.assertEqual(_git(self.root, "symbolic-ref", "--short", "HEAD"), before_branch) + self.assertEqual( + _git(self.root, "status", "--porcelain", "--untracked-files=all"), before_status + ) + # The stale cache is specifically NOT repaired: that would be a mutation. + self.assertEqual(_git(self.root, "symbolic-ref", "refs/remotes/MDCPS/HEAD"), before_symref) if __name__ == "__main__":