ersatztv/scripts/tests/test_ci_toolchain_image_resolves.py

"""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 == "unreachable":
        sys.exit(7)
    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":
    # 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")
        pathlib.Path(dump).write_text(
            "HTTP/1.1 401 Unauthorized\r\n"
            'Www-Authenticate: Bearer realm="%s",service="container_registry",scope="*"\r\n' % 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 "PUBLIC" in result.stderr, "the message must name the cause an operator can act on"
    assert preflight.token_calls(), "the token leg must have been attempted before refusing"


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="
    ]


@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"


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"
    )