ersatztv/scripts/tests/test_ci_release_path_scan_job.py

"""The `scan` job — the delimiter ban made fail-CLOSED on the release path (ersatztv#767).

WHAT THIS IS PROTECTING. #756 brought `build` into the delimiter ban, because a dropped
`Smoke + IPTV E2E` publishes a release candidate that was never booted and reports the job green.
But the ban was enforced ONLY by `test_the_delimiter_banned_jobs_have_NO_expression_delimiter_in_any_run_body`
in `script-tests` — `on: pull_request`, not a required context. Nothing re-checked it on a `v*` tag
push, which is exactly when the candidate is published.

WHY A JOB AND NOT A STEP IN `build`, and why this file is structural. The first cut of #767 put a
bespoke stdlib scanner in `build` itself. Two independent reviews killed it on two counts, and both
are worth keeping written down because both are easy to re-invent:

  * A guard step inside `build` cannot protect `build`. If the runner drops it, the job carries on
    and publishes — fail-OPEN. The defence offered was "the guard's own body has no opener, so it
    cannot be dropped", but the only thing enforcing THAT was the same PR-only test being
    backstopped. Circular. As a `needs:` of `build`, a red here means `build` never runs at all.
  * The bespoke scanner hand-parsed YAML (to avoid provisioning PyYAML on `build`'s bare runner) and
    had ~10 false NEGATIVES within one review round — flow mappings, a quoted `"run":` key, aliases,
    multiline quoted scalars. It was strictly WEAKER than the check it backstopped, in the only
    direction that matters. The fix was to delete it and run the real PyYAML-based test, which needs
    no second definition of "what is a `run:` body" and so has no drift surface.

So the detection logic is not retested here — it lives in `test_ci_dropped_step_guard.py` and this
job runs that file. What this file holds is the WIRING, which is what makes the ban fail-closed:
the job exists, `build` depends on it, nothing can skip it, its own steps cannot be silently
dropped, and it actually invokes the ban test.

WHAT THIS DOES NOT CLAIM. That no step can ever fail to run for a reason other than the
interpolation drop. This job's own steps carry #756 markers and a trailing assert, so the regress
terminates where the sibling guards' does — to fail open you must now drop the pytest step AND the
assert step, not either one. The end-to-end behaviour (a poisoned `Smoke` body reddens `scan` and
`build` never runs) is a LIVE measurement recorded on the issue, not something a static test here
can establish.
"""

from __future__ import annotations

import os
import re
import subprocess
from pathlib import Path

import pytest
import yaml

REPO_ROOT = Path(__file__).resolve().parents[2]
WORKFLOW = REPO_ROOT / ".gitea" / "workflows" / "docker-build.yml"
SCRIPT = REPO_ROOT / "scripts" / "ci-step-ran.sh"

_DOC = yaml.safe_load(WORKFLOW.read_text())
_OPENER = re.compile(r"\$\{\{")
_MARK = re.compile(r'ci-step-ran\.sh"?\s+mark\s+(\S+)')

JOB = "scan"
BAN_TEST_FILE = "scripts/tests/test_ci_dropped_step_guard.py"


def _job():
    assert JOB in _DOC["jobs"], f"the `{JOB}` job is gone — the release path is unguarded again"
    return _DOC["jobs"][JOB]


def _steps():
    return _job()["steps"]


def _run_steps():
    return [s for s in _steps() if s.get("run")]


def _guard():
    """The trailing assert step, located by CONTENT — never by index, so that
    `test_the_guard_is_the_LAST_step` is not true by construction."""
    hits = [s for s in _run_steps() if "ci-step-ran.sh assert" in s["run"]]
    assert len(hits) == 1, f"expected exactly 1 assert step in `{JOB}`, found {len(hits)}"
    return hits[0]


def _marked():
    out = []
    for s in _run_steps():
        m = _MARK.search(s["run"])
        if m:
            out.append((s, m.group(1)))
    return out


# ------------------------------------------------------------------------------------------------
# WIRING — the properties that make the ban fail-closed
# ------------------------------------------------------------------------------------------------


def test_build_DEPENDS_on_the_scan_job():
    """This single edge is the whole fail-closed property.

    Without it the scan is advisory: it could go red while `build` publishes anyway.
    """
    needs = _DOC["jobs"]["build"]["needs"]
    needs = [needs] if isinstance(needs, str) else needs
    assert JOB in needs, f"`build` no longer needs `{JOB}` — a red scan would not stop a release"


def test_the_scan_job_has_NO_job_level_if():
    """Two failure modes at once, in opposite directions.

    An `if:` that excludes the tag push would leave the release path unguarded — the exact hole
    #767 closed. An `if:` that skipped it for any other reason would SKIP `build` too (a skipped
    dependency skips its dependents), breaking every release. Neither is wanted: it always runs.
    """
    job = _job()
    assert "if" not in job, f"`{JOB}` must carry no job-level `if:`, found {job.get('if')!r}"


def test_the_scan_job_actually_invokes_the_ban_test():
    """Otherwise the job is an expensive no-op that reports green.

    Asserted against the file path the ban test really lives in, so renaming that file without
    updating the workflow is a red here rather than a silently unguarded release path.
    """
    assert (REPO_ROOT / BAN_TEST_FILE).is_file()
    assert any(BAN_TEST_FILE in s["run"] for s in _run_steps()), (
        f"no step in `{JOB}` runs {BAN_TEST_FILE}"
    )


def test_no_step_in_the_scan_job_is_advisory():
    """`continue-on-error: true` would make the whole gate a no-op while every other test here
    stayed green — it is the cheapest way to accidentally disarm this."""
    offenders = [s.get("name") for s in _steps() if s.get("continue-on-error")]
    assert not offenders, f"advisory step(s) in `{JOB}`: {offenders}"


@pytest.mark.parametrize(
    "step_name",
    [s.get("name", "?") for s in yaml.safe_load(WORKFLOW.read_text())["jobs"][JOB]["steps"]
     if s.get("run")],
)
def test_every_run_body_in_the_scan_job_is_delimiter_free(step_name):
    """The guard must not be vulnerable to the defect it guards against.

    Not a proof that it always runs — a construction argument about ONE mechanism, the same axiom
    the sibling guards rest on. It is asserted per step so a failure names which step regressed.
    """
    step = next(s for s in _run_steps() if s.get("name", "?") == step_name)
    assert not _OPENER.search(step["run"]), (
        f"step {step_name!r} of `{JOB}` contains an expression delimiter; the runner would rewrite "
        "the whole body and DROP the step while reporting success (ersatztv#751). Pass values "
        "through `env:`, which is interpolated per value."
    )


# ------------------------------------------------------------------------------------------------
# THE JOB'S OWN DROPPED-STEP GUARD
# ------------------------------------------------------------------------------------------------


def test_every_consequential_step_marks_itself():
    """Every `run:` step except the guard records that it executed."""
    marked = {s.get("name") for s, _ in _marked()}
    expected = {s.get("name") for s in _run_steps() if s is not _guard()}
    assert marked == expected, f"unmarked step(s) in `{JOB}`: {expected - marked}"


def test_the_guard_expectations_match_the_markers_exactly():
    """The set the guard waits for IS the set the steps write — derived from the workflow, not
    restated here, so adding a step without a marker is a red."""
    argv = _guard()["run"].split()
    assert "--always" in argv, argv
    always = argv[argv.index("--always") + 1:]
    assert "--gated" not in argv, "every step in this job is unconditional; there is nothing to gate"
    assert sorted(always) == sorted(k for _, k in _marked())


def test_the_guard_is_the_LAST_step():
    assert _steps()[-1] is _guard(), "the assert must run after the steps it checks"


def test_the_guard_has_no_if():
    """Same reasoning as the sibling guards: the default `success()` is wanted, because a genuine
    early failure legitimately skips later steps and already fails the job."""
    assert "if" not in _guard()


# ------------------------------------------------------------------------------------------------
# BEHAVIOURAL — the guard's REAL command line, against the steps' REAL marker lines
# ------------------------------------------------------------------------------------------------


def _mark_line(step) -> str:
    """The step's own marker line, verbatim from the workflow — never rebuilt in Python, so a
    drift between the workflow and the script cannot hide behind a test that composed its own."""
    return next(ln for ln in step["run"].splitlines() if _MARK.search(ln)).strip()


def _env(tmp_path, **extra):
    env = {
        "PATH": os.environ["PATH"],
        "GITHUB_WORKSPACE": str(REPO_ROOT),
        "RUNNER_TEMP": str(tmp_path),
        "GITHUB_JOB": JOB,
        "GITHUB_RUN_ID": "424242",
        "GITHUB_RUN_ATTEMPT": "7",
    }
    env.update(extra)
    return {k: v for k, v in env.items() if v is not None}


def _run(script: str, env):
    return subprocess.run(["bash", "-c", script], cwd=REPO_ROOT, env=env,
                          capture_output=True, text=True)


def test_the_guard_PASSES_when_every_step_ran(tmp_path):
    env = _env(tmp_path)
    for step, _ in _marked():
        assert _run(_mark_line(step), env).returncode == 0
    res = _run(_guard()["run"], env)
    assert res.returncode == 0, res.stderr


@pytest.mark.parametrize("dropped", [k for _, k in _marked()])
def test_the_guard_FAILS_when_a_step_was_dropped(tmp_path, dropped):
    """The positive control. Drop each key in turn — the guard must go red and NAME it.

    A guard only ever exercised on the happy path is indistinguishable from one that passes
    unconditionally, which is the failure this whole mechanism exists to remove.
    """
    env = _env(tmp_path)
    for step, key in _marked():
        if key != dropped:
            assert _run(_mark_line(step), env).returncode == 0
    res = _run(_guard()["run"], env)
    assert res.returncode != 0, f"guard passed despite '{dropped}' never running: {res.stdout}"
    # BOTH streams: the script's `::error::` lands on stdout here while other diagnostics go to
    # stderr, and a test that picked the wrong one would assert on an empty string and pass for the
    # wrong reason on any message change.
    assert dropped in (res.stdout + res.stderr), (res.stdout, res.stderr)


def test_the_guard_REFUSES_to_pass_with_no_expectations(tmp_path):
    """`assert` with an empty expectation set would report success having checked nothing."""
    res = _run(f"{SCRIPT} assert --always", _env(tmp_path))
    assert res.returncode != 0