timothy/ersatztv
1
0
Fork 0
Code Issues 51 Pull Requests 5 Actions Packages Projects Releases 9 Wiki Activity
Files
cb11116d4f18dbc73ccaf8bf74f61ffc0f8646b6
ersatztv/scripts/tests/test_ci_toolchain_image_resolves.py
T
timothyandClaude Fable 5.1 a4eea6460b fix(885): a legal challenge in another case read as a missing realm 
Header field names (RFC 9110 §5.1) and auth-param names (RFC 7235 §2.1) are both
case-insensitive, so `WWW-AUTHENTICATE: Bearer REALM="…"` is the same challenge this
registry sends in mixed case today. The preflight matched the header name in a fixed
case for all but four letters and the directive name in lowercase only, so that
spelling fell into the "named no realm" arm: the job fails — the safe direction —
but names a cause that is not the real one and points an operator at a token
endpoint that is healthy.

The header line is now selected by an `awk` comparison on the lowercased field name,
which leaves the value's case alone (a realm URL is case-sensitive), and the
directive name is matched through a character class generated from the key. The new
test drives the whole anonymous read end to end against an all-caps challenge rather
than testing the parser, so the token leg and the authenticated re-read both have to
survive the spelling.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015QqCpYFsKgnAnx6jVwrKiV
2026-09-05 15:15:43 +02:00

550 lines
26 KiB
Python
Raw Blame History

"""Tests for `scripts/ci-toolchain-image-resolves.sh` (ersatztv#772, ersatztv#885).
The script answers one question — does the tag `docker-build.yml` pins still exist? — and the whole
value is in *which answers it refuses to round off*. A registry read has three outcomes, not two:
present, gone, and could-not-tell. Collapsing the third into either of the others is how a preflight
becomes decoration, so each is driven here through the real entry point with a stubbed `curl`.
`test_MUTATION_a_deleted_tag_is_reported_as_a_failure` is the load-bearing one and is declared in
`scripts/tests/mutation_manifest.py`. Note what it can and cannot turn on: since an unverifiable
answer fails the job too, disarming the `404` arm still exits non-zero, so the EXIT CODE separates
nothing. What the disarm destroys is the DIAGNOSTIC — the outage is reported as "could not verify",
which sends an operator to the registry's health instead of to the rebuild that fixes it.
THE READ IS ANONYMOUS since ersatztv#885: this job runs on the `pull_request` route, where the YAML
is head-supplied, so it may hold no stored secret. The stub therefore reproduces the registry's
TWO-LEG shape rather than a credential check — an unauthenticated read is answered `401` carrying a
Bearer CHALLENGE, and only a request bearing the token that challenge leads to is served. That is
what keeps the file from certifying a script that has quietly stopped doing the token exchange:
without it every request would be answered `401` and every assertion below would go red.
"""
from __future__ import annotations
import os
import subprocess
import time
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "scripts" / "ci-toolchain-image-resolves.sh"
# Stands in for the two curl shapes the script issues: the registry GET
# (`-s -w '\n%{http_code}' -D <headers> [-H Authorization] -H Accept <url>`) and the token GET
# (`-s <realm>?scope=…`). It prints a scripted body, a newline and the HTTP code, dumps response
# headers where `-D` asks for them, and logs every call with the Authorization header it carried.
#
# It ENFORCES the protocol rather than ignoring it: an unauthenticated registry read is 401 + a
# challenge, exactly as the live registry answers (measured 2026-09-04), so a script that stopped
# performing the token leg would fail every test in this file rather than sailing through on a stub
# that answers 200 regardless.
CURL_SHIM = r"""#!/usr/bin/env python3
import os, pathlib, sys
state = pathlib.Path(os.environ["STUB_DIR"])
args = sys.argv[1:]
url = [a for a in args if a.startswith("http")][-1]
def flag(name):
return args[args.index(name) + 1] if name in args and args.index(name) + 1 < len(args) else None
def header(name):
prefix = name.lower() + ":"
for i, a in enumerate(args):
if a == "-H" and i + 1 < len(args) and args[i + 1].lower().startswith(prefix):
return args[i + 1].split(":", 1)[1].strip()
return ""
authorization = header("Authorization")
dump = flag("-D")
codes = dict(pair.split("=", 1) for pair in (state / "codes").read_text().split() if pair)
with (state / "calls").open("a") as fh:
fh.write("%s auth=%s\n" % (url, authorization))
# --- the token endpoint --------------------------------------------------------------------
if "/token" in url:
behaviour = codes.get("TOKEN", "issue")
if behaviour == "blip":
# Unreachable on the FIRST ask and healthy afterwards -- a token-endpoint outage shorter
# than the retry budget, which is the shape a retry exists to absorb.
marker = state / "token-blip"
behaviour = "issue" if marker.exists() else "unreachable"
marker.write_text("1")
if behaviour == "unreachable":
sys.exit(7)
if behaviour == "server-error":
# It ANSWERS, but with nothing a client can use -- the other half of "told us nothing".
print("<html>502 Bad Gateway</html>\n502", end="")
sys.exit(0)
if behaviour == "empty":
print('{"expires_in": 300}\n200', end="")
else:
print('{"token": "anon-token"}\n200', end="")
sys.exit(0)
# --- the registry --------------------------------------------------------------------------
tag = url.rsplit("/", 1)[-1]
code = codes.get(tag, codes.get("*", "200"))
if code == "TRANSPORT":
sys.exit(7)
if authorization != "Bearer anon-token" and codes.get("CHALLENGE") == "none":
# A registry that REFUSES an unauthenticated read outright instead of challenging: no
# Www-Authenticate header at all, so a client has nowhere to ask for a token. `REFUSAL` picks
# the code, because a 403 and a challenge-less 401 take DIFFERENT paths through the script --
# `probe` enters the token leg on 401 only.
refusal = codes.get("REFUSAL", "403")
if dump:
pathlib.Path(dump).write_text("HTTP/1.1 %s Refused\r\n" % refusal)
print("{}\n%s" % refusal, end="")
sys.exit(0)
if authorization != "Bearer anon-token":
# 401 for a live tag and a deleted one alike -- the challenge is the ONLY thing that tells a
# client where a token can be had.
if dump:
realm = codes.get("REALM", "http://registry.test/v2/token")
# A legal challenge in a case no part of the script spells: field names and auth-param
# names are both case-insensitive, so this is the same header, not a malformed one.
line = 'Www-Authenticate: Bearer realm="%s",service="container_registry",scope="*"\r\n'
if codes.get("CHALLENGE") == "upper":
line = 'WWW-AUTHENTICATE: Bearer REALM="%s",SERVICE="container_registry",SCOPE="*"\r\n'
pathlib.Path(dump).write_text("HTTP/1.1 401 Unauthorized\r\n" + line % realm)
print("{}\n401", end="")
sys.exit(0)
if dump:
pathlib.Path(dump).write_text("HTTP/1.1 %s\r\n" % code)
if code == "401-AFTER-TOKEN":
print("{}\n401", end="")
sys.exit(0)
body = '{"schemaVersion": 2, "mediaType": "application/vnd.oci.image.manifest.v1+json"}'
if code == "200-NOT-A-MANIFEST":
code, body = "200", "<html><title>Sign in</title></html>"
print("%s\n%s" % (body, code), end="")
"""
WORKFLOW_TEMPLATE = """jobs:
test:
container:
image: 192.168.1.95:3000/timothy/ersatztv-ci:{pin}
"""
@pytest.fixture
def preflight(tmp_path):
bindir = tmp_path / "bin"
bindir.mkdir()
shim = bindir / "curl"
shim.write_text(CURL_SHIM)
shim.chmod(0o755)
state = tmp_path / "state"
state.mkdir()
(state / "codes").write_text("*=200")
workflow = tmp_path / "docker-build.yml"
workflow.write_text(WORKFLOW_TEMPLATE.format(pin="32747a0"))
env = dict(os.environ)
env["PATH"] = f"{bindir}{os.pathsep}{env['PATH']}"
env["STUB_DIR"] = str(state)
env["ETV_CI_WORKFLOW"] = str(workflow)
env.pop("ETV_REGISTRY_AUTH", None)
# The retry PAUSE is what makes failing on an unknown affordable in CI and unaffordable in a
# test suite; the retry COUNT is behaviour, so it is kept and only the wait is removed.
env["ETV_CI_ATTEMPTS"] = "2"
env["ETV_CI_RETRY_SECONDS"] = "0"
class Handle:
def __init__(self):
self.env = env
self.state = state
self.workflow = workflow
self.script = SCRIPT
def set_codes(self, mapping: dict[str, str]):
(state / "codes").write_text(" ".join(f"{k}={v}" for k, v in mapping.items()))
def set_workflow_text(self, text: str):
workflow.write_text(text)
def calls(self):
log = state / "calls"
return log.read_text().splitlines() if log.exists() else []
def token_calls(self):
return [c for c in self.calls() if "/token" in c]
def manifest_calls(self):
return [c for c in self.calls() if "/manifests/" in c]
def authenticated_manifest_calls(self):
"""The manifest reads that actually carried the bearer.
The ATTEMPT count is this, not `manifest_calls`: the run opens with exactly one
unauthenticated read, which is the challenge that starts the token leg, and every read
after the token is acquired carries it. Counting the raw calls would report one more
attempt than the retry loop made.
"""
return [c for c in self.manifest_calls() if c.endswith("auth=Bearer anon-token")]
def run(self, script: Path | None = None):
return subprocess.run(
["bash", str(script or SCRIPT)],
env=env,
capture_output=True,
text=True,
cwd=REPO_ROOT,
)
return Handle()
def test_a_pin_that_resolves_passes(preflight):
preflight.set_codes({"*": "200"})
result = preflight.run()
assert result.returncode == 0, result.stderr
assert "32747a0 resolves" in result.stdout
assert preflight.authenticated_manifest_calls(), "the registry was never read with a token"
def test_MUTATION_a_deleted_tag_is_reported_as_a_failure(preflight):
"""The outage of 2026-08-11..13, in one assertion.
Declared in `mutation_manifest.py`: replacing the `404` arm sends a deleted tag down the
could-not-verify path, which fails the job with the wrong story — a preflight that runs, reddens,
and still misses the only thing it was built to name.
"""
preflight.set_codes({"32747a0": "404"})
result = preflight.run()
assert result.returncode != 0, (
"a deleted tag did not fail the preflight — the 404 arm is not load-bearing:\n"
f"stdout={result.stdout}\nstderr={result.stderr}"
)
assert "IS GONE" in result.stderr, (
"a deleted tag was not reported as GONE — the 404 arm is not load-bearing. Since an "
"unverifiable answer now fails too, exiting non-zero no longer distinguishes 'the image is "
"deleted' from 'the check could not run', and only this message does:\n"
f"stderr={result.stderr}"
)
assert "32747a0" in result.stderr, "the message must name the tag the operator has to restore"
assert "server-management#842" in result.stderr, "and where the durable fix lives"
@pytest.mark.parametrize("code", ["TRANSPORT", "503"])
def test_an_unknown_answer_FAILS_and_is_not_reported_as_gone(preflight, code):
"""Warning and exiting 0 here is how a preflight becomes a no-op.
A missing `curl`, a moved registry or a DNS change all land in this arm, and each would have
been green forever. It fails — but with its own wording, because "could not verify" and "IS
GONE" send an operator to entirely different places.
"""
preflight.set_codes({"32747a0": code})
result = preflight.run()
assert result.returncode != 0, "an unestablished check must not report success"
assert "could NOT VERIFY" in result.stderr
assert "IS GONE" not in result.stderr, "could-not-tell must never be reported as gone"
def test_an_unknown_is_RETRIED_before_it_fails(preflight):
"""Retries are what make failing on unknown affordable rather than flaky."""
preflight.env["ETV_CI_ATTEMPTS"] = "3"
preflight.set_codes({"32747a0": "503"})
assert preflight.run().returncode != 0
assert len(preflight.authenticated_manifest_calls()) == 3, f"expected 3 attempts, got {preflight.calls()}"
def test_an_ANSWER_is_not_retried(preflight):
"""404 and 200 are answers; retrying them would only slow the job down."""
preflight.set_codes({"32747a0": "404"})
assert preflight.run().returncode != 0
assert len(preflight.authenticated_manifest_calls()) == 1, f"a 404 must not be retried, got {preflight.calls()}"
def test_HTTP_200_with_a_body_that_is_not_a_manifest_is_not_a_pass(preflight):
"""A proxy or a login page answers 200 too; the status line alone establishes nothing."""
preflight.set_codes({"32747a0": "200-NOT-A-MANIFEST"})
result = preflight.run()
assert result.returncode != 0
assert "not a manifest" in result.stderr
assert "IS GONE" not in result.stderr
@pytest.mark.parametrize("code", ["401-AFTER-TOKEN", "403"])
def test_a_401_or_403_AFTER_the_token_leg_REFUSES_rather_than_passing(preflight, code):
"""The failure mode that would otherwise make this job green forever.
Reached only once the token leg has run, so it means one specific thing: this registry will not
serve an anonymous pull of this package — normally because the repo or the package has been made
PRIVATE. Treating that as "could not tell, carry on" would turn it into a permanent silent pass,
and it is not a preflight-only problem: since ersatztv#885 every `container:` job pulls the same
image with no credential, so they fail at image pull too.
"""
preflight.set_codes({"32747a0": code})
result = preflight.run()
assert result.returncode != 0
assert "refused an ANONYMOUS read" in result.stderr
assert "even after a Bearer token was obtained" in result.stderr, (
"this arm is the one where a token really was obtained, so it is the only one allowed to "
f"say so. stderr={result.stderr}"
)
assert "PUBLIC" in result.stderr, "the message must name the cause an operator can act on"
assert preflight.authenticated_manifest_calls(), (
"a message claiming the read survived a Bearer token must be reached with a read that "
"CARRIED one — `token_calls` only shows the token was asked for"
)
@pytest.mark.parametrize("refusal", ["403", "401"])
def test_a_refusal_with_NO_CHALLENGE_never_claims_a_token_was_obtained(preflight, refusal):
"""The message must not name a mechanism the run did not perform.
`probe` enters the token leg on a `401` only, so a registry answering `403` on the first read —
or a `401` carrying no `Www-Authenticate` — leaves the script with no token having asked for
nothing. Measured 2026-09-05 on the predecessor of this commit, the `403` shape reported "even
after a Bearer token was obtained", which sends an operator to package visibility on evidence
that does not exist (`dont-narrate-mechanisms-you-didnt-measure`). The two codes are BOTH driven
because they take different paths: the challenge-less `401` still enters and abandons the token
leg, the `403` never enters it.
The shim answering 401 + a challenge to every unauthenticated read is why the pre-existing
`403` case could not reach this — it could only ever be observed AFTER the token leg — so the
shim grew a challenge-less behaviour rather than the assertion being written against the old one.
"""
preflight.set_codes({"*": "200", "CHALLENGE": "none", "REFUSAL": refusal})
result = preflight.run()
assert result.returncode != 0, "an unreadable registry is not a pass"
assert "even after a Bearer token was obtained" not in result.stderr, (
f"no token was obtained on this path. stderr={result.stderr}"
)
assert preflight.authenticated_manifest_calls() == [], "no read can have carried a token here"
if refusal == "403":
assert preflight.token_calls() == [], "a 403 first read must not even ask for a token"
assert "NO TOKEN WAS EVER REQUESTED" in result.stderr, result.stderr
assert "PUBLIC" in result.stderr, "the message must still name a cause an operator can act on"
else:
# A challenge-less 401 DOES enter the token leg (and abandons it for want of a realm), so it
# is the token-endpoint diagnosis rather than the never-asked one.
assert preflight.token_calls() == [], "there was no realm to request a token from"
assert "could NOT OBTAIN an anonymous pull token" in result.stderr, result.stderr
assert "no Www-Authenticate challenge" in result.stderr, (
f"the token-leg message must admit the challenge was missing. stderr={result.stderr}"
)
def test_the_TOKEN_LEG_actually_runs_and_the_bearer_REACHES_the_registry(preflight):
"""Anti-vacuity for every test above, and the shape of the whole anonymous read in one place.
The stub answers 401 to an unauthenticated read exactly as the live registry does, so a script
that stopped exchanging the challenge for a token would redden this entire file rather than
passing on a stub that serves anyone. The call sequence is pinned rather than counted: challenge,
token, re-read WITH the bearer.
"""
preflight.set_codes({"*": "200"})
assert preflight.run().returncode == 0
assert preflight.calls() == [
"http://192.168.1.95:3000/v2/timothy/ersatztv-ci/manifests/32747a0 auth=",
"http://registry.test/v2/token?scope=repository:timothy/ersatztv-ci:pull&service=container_registry auth=",
"http://192.168.1.95:3000/v2/timothy/ersatztv-ci/manifests/32747a0 auth=Bearer anon-token",
]
def test_the_REALM_is_read_from_the_CHALLENGE_rather_than_assumed(preflight):
"""A hardcoded token endpoint would work here and break the day the registry moves it.
The realm is whatever the `Www-Authenticate` header names, so this points the challenge somewhere
else entirely and requires the token request to follow it.
"""
preflight.set_codes({"*": "200", "REALM": "http://elsewhere.test/auth/v2/token"})
assert preflight.run().returncode == 0
assert preflight.token_calls() == [
"http://elsewhere.test/auth/v2/token?scope=repository:timothy/ersatztv-ci:pull&service=container_registry auth="
]
def test_the_CHALLENGE_is_read_in_ANY_case(preflight):
"""`WWW-AUTHENTICATE: Bearer REALM="…"` is the same challenge, and must take the same path.
Field names (RFC 9110 §5.1) and auth-param names (RFC 7235 §2.1) are both case-insensitive, so
a registry may answer in this spelling without being broken. The predecessor read the header
name in a fixed case for all but four letters and the directive name in lowercase only, which
sent a legal challenge down the "named no realm" arm: a `fail` naming a cause that is not the
real one, pointing an operator at a healthy token endpoint. Driven end to end rather than on the
parser, so the whole anonymous read has to survive the spelling.
"""
preflight.set_codes({"*": "200", "CHALLENGE": "upper"})
result = preflight.run()
assert result.returncode == 0, result.stderr
assert preflight.token_calls() == [
"http://registry.test/v2/token?scope=repository:timothy/ersatztv-ci:pull&service=container_registry auth="
], preflight.calls()
assert preflight.authenticated_manifest_calls(), preflight.calls()
@pytest.mark.parametrize("behaviour", ["empty", "unreachable"])
def test_a_token_endpoint_that_yields_NO_TOKEN_refuses(preflight, behaviour):
"""No bearer is not "carry on unauthenticated" — it is could-not-establish, and it must refuse.
Both shapes leave the script without a token: an answer carrying no `token` field, and an
endpoint that will not answer at all. Falling through to a second unauthenticated read would
surface as an ordinary 401 with no cause named, which is the diagnosis this arm exists to give.
"""
preflight.set_codes({"*": "200", "TOKEN": behaviour})
result = preflight.run()
assert result.returncode != 0, "a run that never obtained a token must not report success"
assert "could NOT OBTAIN an anonymous pull token" in result.stderr, (
"a failed token leg must be worded apart from a refusal that survived a GOOD token: one "
"sends an operator to the registry's token endpoint, the other to the package's visibility. "
f"stderr={result.stderr}"
)
assert "still PUBLIC" not in result.stderr, "nothing here establishes the package went private"
assert preflight.authenticated_manifest_calls() == [], "no read can have carried a token"
@pytest.mark.parametrize("behaviour", ["unreachable", "server-error"])
def test_a_token_endpoint_that_SAID_NOTHING_is_retried_like_any_other_unknown(preflight, behaviour):
"""The two legs of one read must not have opposite flake tolerances.
A token endpoint that cannot be reached (or answers 5xx) has told the run nothing — it is the
same transport blip a flaky manifest read gets three tries to survive. Measured 2026-09-05 on
the predecessor of this commit, it got ZERO: `token_leg_done` was set once per RUN, the failed
leg left the first read's `401` standing, and `401` was in the retry loop's break list, so the
job reddened after one token call. A red here denies a merge (the consent hook reads the
COMBINED status, ersatztv#598), so that was a one-second outage blocking a merge until someone
re-ran the job.
"""
preflight.env["ETV_CI_ATTEMPTS"] = "3"
preflight.set_codes({"*": "200", "TOKEN": behaviour})
result = preflight.run()
assert result.returncode != 0, "a run that never obtained a token must not report success"
assert len(preflight.token_calls()) == 3, (
f"expected the token leg to be retried on the attempt budget, got {preflight.calls()}"
)
assert "could NOT OBTAIN an anonymous pull token" in result.stderr, result.stderr
assert "after 3 token-leg attempt(s)" in result.stderr, (
f"the message must report what the run actually did, not what it might have. stderr={result.stderr}"
)
@pytest.mark.parametrize("codes", [{"TOKEN": "empty"}, {"CHALLENGE": "none", "REFUSAL": "401"}])
def test_a_token_endpoint_that_ANSWERED_is_asked_ONCE(preflight, codes):
"""The other direction, and the property the retry above must not cost.
An endpoint that answered and named no token — or a `401` carrying no challenge to follow — has
said something, and asking again cannot change it. That is a registry genuinely refusing
anonymous reads, and it is asked once per RUN.
"""
preflight.env["ETV_CI_ATTEMPTS"] = "3"
preflight.set_codes({"*": "200", **codes})
result = preflight.run()
assert result.returncode != 0
assert len(preflight.token_calls()) <= 1, f"a settled token leg must not be re-asked, got {preflight.calls()}"
assert len(preflight.manifest_calls()) == 1, (
f"nor may the manifest read be retried behind it, got {preflight.calls()}"
)
assert "could NOT OBTAIN an anonymous pull token" in result.stderr, result.stderr
def test_a_token_endpoint_BLIP_shorter_than_the_budget_RECOVERS(preflight):
"""Retrying is only worth anything if the run can still succeed.
The endpoint is unreachable on the first ask and healthy on the second, so the whole point is
the exit code: the pin resolves, with a bearer that reached the registry.
"""
preflight.env["ETV_CI_ATTEMPTS"] = "3"
preflight.set_codes({"*": "200", "TOKEN": "blip"})
result = preflight.run()
assert result.returncode == 0, f"stdout={result.stdout}\nstderr={result.stderr}"
assert "32747a0 resolves" in result.stdout
assert len(preflight.token_calls()) == 2, f"expected one failed ask and one good one, got {preflight.calls()}"
assert preflight.authenticated_manifest_calls(), "the recovered token never reached the registry"
def test_the_TOKEN_LEG_is_performed_ONCE_across_every_pin(preflight):
"""The token is state that must survive the pin loop.
`probe` assigns it rather than printing it precisely because a `$(…)` command substitution runs
in a subshell whose assignments are discarded — a version that read the answer through one would
re-run the challenge and the exchange for every pin, and pass every other test in this file.
"""
preflight.set_workflow_text(
WORKFLOW_TEMPLATE.format(pin="32747a0") + " image: 192.168.1.95:3000/timothy/ersatztv-ci:15d2439\n"
)
preflight.set_codes({"*": "200"})
assert preflight.run().returncode == 0
assert len(preflight.token_calls()) == 1, f"the token leg ran {len(preflight.token_calls())} times"
assert len(preflight.authenticated_manifest_calls()) == 2
def test_a_workflow_with_no_pin_at_all_is_a_failure(preflight):
"""If the grep stops matching, the honest report is 'I found nothing', not 'all clear'."""
preflight.set_workflow_text("jobs:\n test:\n runs-on: ubuntu-latest\n")
result = preflight.run()
assert result.returncode != 0
assert "no ersatztv-ci pin found" in result.stderr
def test_every_distinct_pin_is_checked_and_one_gone_fails_the_job(preflight):
"""`ci-image-pin` bans a second pin; this must not silently check only the first one anyway."""
preflight.set_workflow_text(
WORKFLOW_TEMPLATE.format(pin="32747a0") + " image: 192.168.1.95:3000/timothy/ersatztv-ci:15d2439\n"
)
preflight.set_codes({"32747a0": "200", "15d2439": "404"})
result = preflight.run()
assert result.returncode != 0
assert "15d2439" in result.stderr
assert len(preflight.authenticated_manifest_calls()) == 2, f"both pins must be queried, got {preflight.calls()}"
def test_the_grep_line_cannot_match_ITSELF(preflight):
"""The pin is found with the same expression `pr-checks.yml::ci-image-pin` uses.
That expression is written into this script's own source, so a careless pattern would find its
own text and 'check' a pin nobody wrote — and the same hazard sits in `pr-checks.yml`, whose
pin-count check greps the file this script's job now lives in. Feed the real script its own
source as the workflow file: the answer must be 'no pin found', not a query for `[0-9a-f]+`.
This also pins the second half of the property — the source carries no literal pin of its own,
so the file cannot go stale against a pin bump it does not participate in.
"""
preflight.set_workflow_text(SCRIPT.read_text())
result = preflight.run()
assert result.returncode != 0
assert "no ersatztv-ci pin found" in result.stderr
assert preflight.calls() == []
def test_the_PRODUCTION_retry_defaults_are_the_ones_that_run(preflight):
"""Every other test overrides the retry knobs, so nothing evaluated `${VAR:-default}` itself.
That matters because the defaults are the argument: "unknown fails" is only affordable if an
ordinary registry blip is absorbed first. Edited to 1 attempt / 0 seconds, this file would stay
green while a single transient 503 reddened every PR. So this one drops both overrides and
measures the real thing — three attempts, and a pause long enough to have actually happened.
"""
del preflight.env["ETV_CI_ATTEMPTS"]
del preflight.env["ETV_CI_RETRY_SECONDS"]
preflight.set_codes({"32747a0": "503"})
started = time.monotonic()
result = preflight.run()
elapsed = time.monotonic() - started
assert result.returncode != 0
assert len(preflight.authenticated_manifest_calls()) == 3, (
f"the default attempt count is not 3 — got {len(preflight.authenticated_manifest_calls())} call(s)"
)
assert elapsed >= 8, (
f"two pauses at the default 5s should clear the 8s floor; took {elapsed:.1f}s, so the pause "
"has been shortened out from under the 'a blip does not redden a PR' argument"
)
Reference in New Issue View Git Blame Copy Permalink