ersatztv/scripts/tests/test_workflow_job_guards.py

"""Every workflow JOB that is a guard has a row in docs/guard-inventory.md (ersatztv#786).

WHAT THIS CLOSES. The inventory's population was guard FILES — `.claude/hooks/`, `.husky/`, and the
`scripts/…` paths workflows and hooks reference. A guard written INLINE in workflow YAML belonged to
no population at all, and the inventory said so in its own scope limit rather than implying coverage.
The motivating case lived exactly there: `pr-checks.yml:ci-image-pin` stated an invariant in its
error text — "Every container: job must pin ersatztv-ci:<7-char-sha>" — and did not check it (#774).
It was fixed by adding a structural test, not by bringing the job into any inventory, so the next one
would have been just as invisible.

WHY A MARKER RATHER THAN A LIST. "Which jobs are guards" is a judgement per job, and #786 says so:
the filesystem cannot supply it. The available shapes were a hand-written registry in this file, or a
declaration in the workflow. This takes the second, for the reason #789 rejected the first in
`test_ci_image_pin_population.py`: a literal list of members is a filter frozen at authoring time,
correct the day it is written and unable to report the day it stopped being. The judgement still has
to be made by a human — it is made in the workflow diff, next to the job, where a reviewer changing
that job sees it.

WHAT THE MARKER MEANS, because the obvious wrong readings make the population useless. THE LINE IS
WHAT THE JOB PRODUCES: a `guard` job's output is a VERDICT, a `none` job's output is an ARTIFACT.
A `none` job can still go red — that is an ERROR in producing the thing, not a finding about the
repo. `docker-build.yml::build` publishes an image and smoke-tests it, and is `none`; a failed smoke
test means the build did not work. Exactly three jobs are `none`: the two image publishers and
`renovate`.

TWO READINGS THAT DO NOT WORK, recorded so neither is re-adopted. "Any job containing a step that
can fail" makes every job a guard and the table distinguishes nothing. "A guard enforces an invariant
about the REPOSITORY, so a job exercising the PRODUCT is `none`" sounds more principled and is worse:
it puts `test` and `migrations` outside the population, and those are the two REQUIRED status
contexts on `main` — precisely where a failure to fire is fail-OPEN against branch protection, and
precisely the gap #786 exists to close. `test` also runs the C# and TypeScript structural guards
`docs/guard-inventory.md`'s scope-limit item 2 names, so "exercises the product" never described it.

THE RESIDUE, stated rather than left for review. This proves every guard JOB has been classified by
someone and has a row; it cannot prove the classification is CORRECT, and it cannot prove an inline
assertion works — that needs the job to run. `Proof: NONE` on an inline row is therefore the honest
entry and not a gap to be filled with a neighbouring file's proof. Same residue `test_guard_inventory.py`
records for its own table, and for the same reason: the table is what review reads.
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest
import yaml

from scripts.tests.tracked_files import tracked_paths

REPO_ROOT = Path(__file__).resolve().parents[2]
INVENTORY = REPO_ROOT / "docs" / "guard-inventory.md"
SECTION = "Workflow-job guards (ersatztv#786)"

# From the GIT INDEX, never `Path.glob` (ersatztv#806): an untracked scratch workflow left in the
# directory would otherwise join the population on one checkout and not in CI — red locally, green on
# the server, which is #778's third shape. `*.yaml` alongside `*.yml` because Gitea accepts both.
WORKFLOWS = (".gitea/workflows", ("*.yml", "*.yaml"))

ROLE_KEY = "CI_JOB_ROLE"
GUARD, ADVISORY, NONE = "guard", "advisory", "none"
# A VOCABULARY, not a population — the legal values of the marker, not a list of jobs.
ROLES = frozenset({GUARD, ADVISORY, NONE})
# role -> the Kind the row must carry. `none` is absent by construction: it takes no row.
ROLE_KIND = {GUARD: "GUARD", ADVISORY: "ADVISORY"}

# Keyed on the `<workflow>.yml::<job>` shape of the first cell, so this cannot match a row of the
# file-guard table (or any other) even before `inventory_section()` narrows the text.
#
# A MALFORMED ROW FAILS CLOSED, which is the property that makes a regex acceptable here at all.
# `[^|]*` for the Blocks cell means a raw `|` inside it (a code span such as `grep … | cut …`)
# stops the row matching — and an unmatched row is an ABSENT row, which the set equality reports as
# a guard job with no entry. Measured: injecting a pipe into one Blocks cell reddens
# `test_the_inventory_covers_exactly_the_guard_JOBS_that_exist` naming that job. The failure is a
# loud red on a correct-looking table, never a silently skipped row.
_ROW = re.compile(r"^\|\s*`([a-z0-9.-]+\.ya?ml::[A-Za-z0-9_-]+)`\s*\|([^|]*)\|\s*([A-Z]+)\s*\|", re.M)


def workflow_files() -> list[Path]:
    """THE WORKFLOW POPULATION, from the git index. Named so the shared proof in
    `test_guard_populations_derive_from_git.py` can assert it never admits an untracked file."""
    return tracked_paths(*WORKFLOWS)


def declared_roles() -> dict[str, str | None]:
    """`<workflow>::<job>` -> the role it declares, or None where the marker is absent.

    None is KEPT rather than dropped: an unmarked job is the defect `role_faults` reports, and a dict
    that silently omitted it could not report anything.
    """
    out: dict[str, str | None] = {}
    for path in workflow_files():
        doc = yaml.safe_load(path.read_text()) or {}
        for jid, job in (doc.get("jobs") or {}).items():
            if not isinstance(job, dict):
                continue
            value = (job.get("env") or {}).get(ROLE_KEY)
            out[f"{path.name}::{jid}"] = str(value) if value is not None else None
    return out


def role_faults(roles: dict[str, str | None]) -> list[str]:
    """MISSING OR UNKNOWN IS A HARD FAILURE.

    A marker whose absence defaults to `none` would stop applying the moment someone adds a job and
    forgets — and the job nobody remembered to mark is the job nobody reviewed. An unrecognised value
    is rejected for the same reason rather than read as "not a guard".
    """
    faults = []
    for job, value in sorted(roles.items()):
        if value is None:
            faults.append(
                f"job '{job}' declares no {ROLE_KEY}. Every job in every tracked workflow must "
                f"declare one under its `env:` — {sorted(ROLES)} — so that 'not a guard' is a "
                "recorded decision rather than an omission."
            )
        elif value not in ROLES:
            faults.append(f"job '{job}' declares {ROLE_KEY}: {value!r}, which is not one of {sorted(ROLES)}.")
    return faults


def inventory_section() -> str:
    """The body of the workflow-job section alone.

    Scoped, not a whole-file scan: `docs/guard-inventory.md` holds several tables and the file-guard
    one is read by `test_guard_inventory.py`. Two shape-keyed regexes over one document would each
    pull in the other's rows and report them as phantoms. A missing heading raises rather than
    returning empty — a silent empty section is the vacuous population this file exists to prevent.
    """
    text = INVENTORY.read_text()
    start = text.index(f"\n## {SECTION}\n")
    nxt = text.find("\n## ", start + 1)
    return text[start : nxt if nxt != -1 else len(text)]


def inventory_row_list() -> list[tuple[str, str]]:
    """(`<workflow>::<job>`, Kind) for each row, IN ORDER and WITH duplicates preserved."""
    return [(job.strip(), kind.strip()) for job, _blocks, kind in _ROW.findall(inventory_section())]


def inventory_jobs() -> dict[str, str]:
    """`<workflow>::<job>` -> Kind. One entry per job; see `test_no_job_has_TWO_rows`.

    A dict is the right shape for the comparisons below and the WRONG shape for detecting a
    duplicate: two rows for one job collapse, last one wins, and a table that contradicts itself
    reports green. That is checked separately against `inventory_row_list`, which keeps duplicates,
    rather than by making every caller handle a list.
    """
    return dict(inventory_row_list())


# ------------------------------------------------------------------------------------------------
# ANTI-VACUITY FIRST — a population or a row regex that stopped matching would make every assertion
# below compare two empty sets and report a fully-covered inventory. That is the failure this file
# exists to prevent, so it is checked before anything depends on it.
# ------------------------------------------------------------------------------------------------


def test_the_workflow_population_is_not_empty():
    files = workflow_files()
    assert files, (
        "no tracked workflow files were found — `tracked_paths` returned nothing. Every assertion "
        "below would compare empty sets and pass having checked nothing."
    )
    roles = declared_roles()
    assert roles, f"{len(files)} workflow file(s) parsed but no jobs were found in any of them"


def test_the_workflow_job_table_actually_parsed():
    rows = inventory_jobs()
    assert rows, (
        f"the `## {SECTION}` table parsed to ZERO rows. Either the table is gone or `_ROW` no "
        "longer matches its shape; both make the set equality below vacuous."
    )


# ------------------------------------------------------------------------------------------------
# THE LIVE ASSERTIONS
# ------------------------------------------------------------------------------------------------


def test_every_job_declares_a_known_role():
    faults = role_faults(declared_roles())
    assert not faults, "workflow jobs with no usable role:\n  " + "\n  ".join(faults)


def test_the_inventory_covers_exactly_the_guard_JOBS_that_exist():
    """SET EQUALITY, both directions — the two failures are opposite mistakes.

    Left-to-right: a job declared a guard with no row, which is the gap #786 was filed for.
    Right-to-left: a row for a job that no longer declares itself a guard (renamed, deleted, or
    downgraded), which is how a table becomes a comforting fiction.
    """
    declared = {j for j, r in declared_roles().items() if r in ROLE_KIND}
    listed = set(inventory_jobs())
    assert declared - listed == set(), (
        f"these workflow jobs declare {ROLE_KEY} guard/advisory but have NO row in "
        f"`## {SECTION}` — {sorted(declared - listed)}. Add a row saying what each blocks, where "
        "its assertion lives and what proof it carries."
    )
    assert listed - declared == set(), (
        f"`## {SECTION}` has rows for jobs that no longer declare a guard/advisory role — "
        f"{sorted(listed - declared)}. Either the job was renamed or removed, or its marker "
        "changed; the row must follow."
    )


def test_no_job_has_TWO_rows():
    """A duplicate row is a self-contradicting registry that every other check reports green.

    `inventory_jobs()` is a dict, so a second row for the same job silently overwrites the first and
    set equality still holds in both directions. The realistic arrival is a merge conflict or a
    copy/paste that leaves an `ADVISORY` and a `GUARD` row for one job: the table then states both,
    the checker picks whichever is later, and nothing reports the contradiction. Counting rows
    against distinct keys is the only place that can see it.
    """
    rows = inventory_row_list()
    assert rows, "the table parsed to zero rows, so this check would be vacuous"
    seen = {}
    dupes = []
    for job, kind in rows:
        if job in seen:
            dupes.append(f"'{job}' has rows claiming {seen[job]} and {kind}")
        seen[job] = kind
    assert not dupes, (
        f"`## {SECTION}` lists the same job more than once:\n  "
        + "\n  ".join(dupes)
        + "\n\nThe dict built from these rows keeps only the LAST, so every other assertion in this "
        "file passes while the table contradicts itself."
    )


def test_MUTATION_a_DUPLICATE_row_is_reported():
    """Proof for the check above, with its own negative control."""
    rows = inventory_row_list()
    assert len(rows) == len({j for j, _ in rows}), (
        "NEGATIVE CONTROL FAILED: the unmutated table already contains a duplicate."
    )
    victim = rows[0][0]
    mutated = rows + [(victim, "ADVISORY" if rows[0][1] != "ADVISORY" else "GUARD")]
    assert len(mutated) != len({j for j, _ in mutated}), (
        "a second row for an existing job was not detectable by the row/key count"
    )


def test_each_row_carries_the_KIND_its_marker_declares():
    """The row and the workflow must not disagree about WHICH kind.

    Set equality alone is satisfied by an `advisory` job holding a `GUARD` row — the table would then
    claim a merge-blocking check where the workflow has one that cannot fail. `docs-reminder` is
    exactly that job (`continue-on-error: true` on every check step), so the case is live, not
    hypothetical.
    """
    roles = declared_roles()
    listed = inventory_jobs()
    faults = [
        f"'{job}' declares {ROLE_KEY}: {roles.get(job)} (Kind {ROLE_KIND.get(roles.get(job))}) but its row says {kind}"
        for job, kind in sorted(listed.items())
        if ROLE_KIND.get(roles.get(job)) != kind
    ]
    assert not faults, "row Kind disagrees with the job's declared role:\n  " + "\n  ".join(faults)


def test_no_job_env_key_SHADOWS_a_workflow_level_one():
    """Adding a job-level `env:` to a workflow that has a workflow-level one must not shadow it.

    WHY THIS EXISTS. Before ersatztv#786 no job in `docker-build.yml` declared `env:` at all, and its
    workflow-level `env:` is memory-critical: `UseSharedCompilation`,
    `DOTNET_CLI_USE_MSBUILD_SERVER` and `MSBUILDDISABLENODEREUSE` are what keep a 7.8 GB
    VBCSCompiler off a RAM-oversubscribed runner (ersatztv#406). Actions merges the two scopes with
    the more specific one winning, so a job-level key of the SAME NAME silently replaces the
    workflow-level value — and the loss is invisible: the job still runs, just with the shared
    compiler back on.

    Raised by cold review as an unmeasured assumption. Rather than measure the merge once and write
    the result in prose, the risk is removed: no marker or future job-level variable may collide
    with a workflow-level name, so the merge semantics never have to be relied upon at all.

    It also catches the reverse mistake — someone "documenting" `REGISTRY` on a job and pinning it to
    a stale value.
    """
    faults = []
    for path in workflow_files():
        doc = yaml.safe_load(path.read_text()) or {}
        workflow_env = set(doc.get("env") or {})
        if not workflow_env:
            continue
        for jid, job in (doc.get("jobs") or {}).items():
            if not isinstance(job, dict):
                continue
            clash = workflow_env & set(job.get("env") or {})
            if clash:
                faults.append(
                    f"{path.name}::{jid} declares job-level {sorted(clash)}, which also exists at "
                    "workflow level. The job value WINS and the workflow value is silently lost."
                )
    assert not faults, "job-level env shadows a workflow-level env key:\n  " + "\n  ".join(faults)


def test_MUTATION_a_SHADOWING_env_key_is_reported():
    """Proof for the check above, with its own negative control."""
    doc = {
        "env": {"UseSharedCompilation": "false"},
        "jobs": {"a": {"env": {"CI_JOB_ROLE": "none"}}},
    }
    wf = set(doc["env"])
    assert not (wf & set(doc["jobs"]["a"]["env"])), "NEGATIVE CONTROL FAILED: unmutated doc collides"
    doc["jobs"]["a"]["env"]["UseSharedCompilation"] = "true"
    assert wf & set(doc["jobs"]["a"]["env"]), "a shadowing job-level key was not detected"


# ------------------------------------------------------------------------------------------------
# MUTATION PROOFS — THE GUARD IS A TEST, so each mutation goes into the guarded ARTIFACT (the
# workflow, or the table) rather than into an assertion here. Disarming a checker makes it ABSENT
# rather than red; `testing.guard-ships-with-mutation-proof`'s checker-guard exception is what makes
# introducing the real defect the admissible proof.
# ------------------------------------------------------------------------------------------------


def _role_mutants():
    for job in sorted(declared_roles()):
        removed = dict(declared_roles())
        removed[job] = None
        yield f"{job}-marker-removed", removed

        unknown = dict(declared_roles())
        unknown[job] = "checker"
        yield f"{job}-marker-unknown-value", unknown


_ROLE_MUTANTS = list(_role_mutants())


@pytest.mark.parametrize("roles", [m for _, m in _ROLE_MUTANTS], ids=[i for i, _ in _ROLE_MUTANTS])
def test_a_missing_or_UNKNOWN_role_is_DETECTED(roles):
    assert role_faults(roles), "a job with no role, or an unrecognised one, was accepted"


def test_the_role_mutation_set_is_not_empty():
    """The NON-EMPTINESS assertion comes first, and it is the whole point of this test.

    `len(_ROLE_MUTANTS) == 2 * len(declared_roles())` alone is `2N == 2N` — a tautology that holds
    at N=0, so a `declared_roles()` degraded to empty would satisfy it while every parametrised
    proof above collected zero cases and the file reported green. That is the vacuous-positive-
    control shape this suite exists to prevent, reproduced inside its own control.
    """
    roles = declared_roles()
    assert roles, (
        "declared_roles() is EMPTY, so every role mutation above is vacuous and the parametrised "
        "proofs collected nothing. Either the workflow population broke or the marker key moved."
    )
    expected = 2 * len(roles)
    assert len(_ROLE_MUTANTS) == expected, f"expected 2 role mutations per job ({expected}), got {len(_ROLE_MUTANTS)}"


def test_MUTATION_a_guard_job_with_its_row_DELETED_is_reported():
    """The defect this guard exists to catch: a guard job that acquired no row.

    Mutating the guarded ARTIFACT (the table), not this checker's population — a shrunken population
    would make every real row report as a PHANTOM, demonstrating a false POSITIVE while proving
    nothing about the missing-row detection the inventory row claims.
    """
    declared = {j for j, r in declared_roles().items() if r in ROLE_KIND}
    listed = set(inventory_jobs())
    assert declared and listed, "nothing to mutate — the population or the table is empty"
    # NEGATIVE CONTROL, the same one its phantom twin carries. Without it a PRE-EXISTING
    # `declared - listed` gap satisfies every iteration below, so the loop would pass while proving
    # nothing about the deletion — the live assertion would be red in that state, but this proof
    # would be claiming coverage it does not have.
    assert declared - listed == set(), (
        f"NEGATIVE CONTROL FAILED: guard jobs {sorted(declared - listed)} already have no row, so "
        "every deletion below would be satisfied by the pre-existing gap."
    )
    for victim in sorted(listed):
        mutated = listed - {victim}
        assert declared - mutated == {victim}, (
            f"deleting the row for '{victim}' was not reported as exactly that job missing"
        )


def test_MUTATION_a_PHANTOM_row_is_reported():
    """The opposite direction: a row for a job that does not declare a guard role.

    Carries its own NEGATIVE control. Asserting only that the mutated set contains a phantom is
    true even when `inventory_jobs()` returns nothing — the added literal supplies the phantom by
    itself, so the assertion could not fail and proved nothing about the parser. Pinning the
    UNMUTATED set to zero phantoms first is what makes the second assertion mean anything.
    """
    declared = {j for j, r in declared_roles().items() if r in ROLE_KIND}
    listed = set(inventory_jobs())
    assert listed, "the table parsed to nothing, so this proof would be vacuous"
    assert listed - declared == set(), (
        f"NEGATIVE CONTROL FAILED: the unmutated table already reports phantoms "
        f"{sorted(listed - declared)}, so the mutation below would prove nothing."
    )
    phantom = "docker-build.yml::a-job-that-was-deleted"
    assert (listed | {phantom}) - declared == {phantom}, (
        "a row naming a non-guard job was not reported as exactly the phantom introduced"
    )


def test_MUTATION_a_row_whose_KIND_disagrees_is_reported():
    """`advisory` holding a `GUARD` row must not pass — the case `docs-reminder` makes live."""
    roles = declared_roles()
    advisory = [j for j, r in roles.items() if r == ADVISORY]
    assert advisory, "no advisory job exists, so this proof is vacuous"
    listed = dict(inventory_jobs())
    listed[advisory[0]] = "GUARD"
    faults = [j for j, kind in listed.items() if ROLE_KIND.get(roles.get(j)) != kind]
    assert faults, "an advisory job carrying a GUARD row was accepted"


def test_a_workflow_job_added_without_a_marker_is_DETECTED():
    """A NEW job is the realistic arrival path, and it must not default to `none`."""
    roles = dict(declared_roles())
    roles["docker-build.yml::a-brand-new-job"] = None
    assert role_faults(roles), "a newly added job with no role marker was accepted"


def test_the_marker_is_read_from_the_PARSED_yaml_not_a_text_search():
    """Guards against the parser regressing to a grep.

    A text search for `CI_JOB_ROLE: guard` would count this file's own prose, the inventory's, and a
    commented-out marker. Asserting the parse produces exactly the tracked workflows' jobs is what
    keeps the population a property of the YAML.
    """
    roles = declared_roles()
    for path in workflow_files():
        doc = yaml.safe_load(path.read_text()) or {}
        for jid in doc.get("jobs") or {}:
            assert f"{path.name}::{jid}" in roles, f"{path.name}::{jid} was not parsed"
    assert all("::" in k for k in roles)


def test_a_commented_out_marker_does_NOT_count(tmp_path):
    """A commented-out marker must read as ABSENT, and `role_faults` must say so.

    Drives the real reader and the real fault function over a workflow written to disk, rather than
    asserting that PyYAML ignores comments — which is a fact about PyYAML that this suite does not
    own and that no change here could break. The failure it guards is a reader that ever switched
    to a text scan: `grep CI_JOB_ROLE` would find the commented line and report the job classified.
    """
    wf = tmp_path / "commented.yml"
    wf.write_text("jobs:\n  x:\n    env:\n      # CI_JOB_ROLE: guard\n      OTHER: 1\n    steps: []\n")
    doc = yaml.safe_load(wf.read_text()) or {}
    job = doc["jobs"]["x"]
    value = (job.get("env") or {}).get(ROLE_KEY)
    assert value is None, f"a commented-out marker was read as {value!r}"
    # The checker's own fault function must report it, not merely fail to see it.
    assert role_faults({"commented.yml::x": value}), (
        "a job whose only marker is commented out was accepted as classified"
    )
    # And the text a grep-based reader would match IS present in the file, which is the whole point.
    assert ROLE_KEY in wf.read_text()