rune/tests/test_workflow_script_contract.py

"""Env-contract guard between Gitea workflow steps and scripts/*.py (#192).

Spec-gate finding 1 discharge — the systematic guard: every step that invokes
a ``scripts/*.py`` entrypoint must provide every env var that script reads
with a hard ``os.environ[...]`` subscript. A read counts as satisfied when it
comes from any of:

* the step's own ``env:`` block (plus workflow/job-level ``env:``),
* a ``$GITHUB_ENV`` export from an earlier step in the same job (shell
  ``echo "NAME=..." >> "$GITHUB_ENV"`` or heredoc ``env.write(f"NAME=...")``),
* a plain shell ``export NAME=`` inside the invoking run block,
* a documented script-derived default (``SCRIPT_DERIVED`` below) — keys the
  script derives itself from other env instead of requiring them.

Regression history: ``create_release.py`` subscript-read LIST_API_URL /
RELEASE_API_URL while ``auto-version.yml`` only ever passed
RELEASE_SERVER_URL/RELEASE_REPO — KeyError at runtime on both release paths.
"""

from __future__ import annotations

import re
from pathlib import Path
from typing import Any, Iterator

import pytest
import yaml

ROOT = Path(__file__).resolve().parents[1]
WORKFLOW_DIR = ROOT / ".gitea" / "workflows"
SCRIPTS_DIR = ROOT / "scripts"

WORKFLOWS = ("auto-version.yml", "maintenance.yml")

# Hard env reads: os.environ["KEY"] / os.environ['KEY'] (KeyError risk).
REQUIRED_ENV_RE = re.compile(r"""os\.environ\[["']([A-Za-z_][A-Za-z0-9_]*)["']\]""")
# Bare subscript of an allowlisted derived key — the derivation must be real.
_SUBSCRIPT_TEMPLATES = ('os.environ["{key}"', "os.environ['{key}'")
# GITHUB_ENV exports, shell form: echo "NAME=..." >> "$GITHUB_ENV"
GITHUB_ENV_SHELL_RE = re.compile(r"""([A-Za-z_][A-Za-z0-9_]*)=.*?>>\s*["']?\$\{?GITHUB_ENV\}?["']?""")
# GITHUB_ENV exports, python form: env.write(f"NAME=...") / env.write("NAME=...")
GITHUB_ENV_PY_RE = re.compile(r"""env\.write\(f?["']([A-Za-z_][A-Za-z0-9_]*)=""")
# Shell exports inside the invoking run block: export NAME=...
SHELL_EXPORT_RE = re.compile(r"^\s*export\s+([A-Za-z_][A-Za-z0-9_]*)=", re.MULTILINE)
# scripts/<name>.py invocation inside a run block
SCRIPT_INVOKE_RE = re.compile(r"scripts/([a-z_][a-z0-9_]*)\.py")

# Keys each script derives itself from other env instead of requiring them
# from the workflow. Guarded below: an allowlisted key must NOT be read with
# a bare subscript (that would KeyError when the derivation inputs are absent).
SCRIPT_DERIVED: dict[str, frozenset[str]] = {
    "create_release": frozenset({"LIST_API_URL", "RELEASE_API_URL"}),
    "auto_bump_pr": frozenset(),
    "version_convergence": frozenset(),
    "file_drift_issue": frozenset(),
}


def _load_workflow(name: str) -> dict[str, Any]:
    path = WORKFLOW_DIR / name
    assert path.exists(), f"workflow missing: {path}"
    data = yaml.safe_load(path.read_text(encoding="utf-8"))
    assert isinstance(data, dict), f"workflow did not parse to a mapping: {name}"
    return data


def _invocations(workflow_name: str) -> Iterator[tuple[str, str, frozenset[str]]]:
    """Yield (step_id, script_name, provided_env) per scripts/*.py invocation.

    ``provided_env`` models runtime visibility faithfully: step env + same-run
    shell exports are visible to the invoking step; GITHUB_ENV exports only
    become visible to *subsequent* steps, so they are accumulated after the
    step's own invocations are checked.
    """
    workflow = _load_workflow(workflow_name)
    for job_name, job in (workflow.get("jobs") or {}).items():
        github_env: set[str] = set()
        for idx, step in enumerate(job.get("steps") or []):
            run = step.get("run") or ""
            step_env = frozenset((step.get("env") or {}).keys())
            shell_exports = frozenset(SHELL_EXPORT_RE.findall(run))
            for match in SCRIPT_INVOKE_RE.finditer(run):
                step_id = f"{workflow_name}:{job_name}/step{idx}"
                yield step_id, match.group(1), step_env | shell_exports | frozenset(github_env)
            github_env |= set(GITHUB_ENV_SHELL_RE.findall(run))
            github_env |= set(GITHUB_ENV_PY_RE.findall(run))


def _params() -> list[tuple[str, str, frozenset[str]]]:
    cases: list[tuple[str, str, frozenset[str]]] = []
    for workflow_name in WORKFLOWS:
        cases.extend(_invocations(workflow_name))
    assert cases, "no scripts/*.py invocations found — guard is vacuous"
    return cases


def _script_source(script: str) -> str:
    return (SCRIPTS_DIR / f"{script}.py").read_text(encoding="utf-8")


@pytest.mark.parametrize(("step_id", "script", "provided"), _params(), ids=lambda v: str(v))
def test_step_provides_every_required_env_read(step_id: str, script: str, provided: frozenset[str]) -> None:
    source = _script_source(script)
    required = set(REQUIRED_ENV_RE.findall(source))
    missing = required - provided - SCRIPT_DERIVED.get(script, frozenset())
    assert not missing, (
        f"{step_id}: scripts/{script}.py hard-reads {sorted(missing)} from os.environ "
        f"but the workflow provides none of them (step/job env, $GITHUB_ENV exports, "
        f"shell exports, or a documented script-derived default)"
    )


@pytest.mark.parametrize(("step_id", "script", "provided"), _params(), ids=lambda v: str(v))
def test_derived_allowlist_is_honest(step_id: str, script: str, provided: frozenset[str]) -> None:
    """An allowlisted derived key must genuinely be derived: no bare subscript
    read of it may remain in the script (that would KeyError at runtime)."""
    source = _script_source(script)
    for key in SCRIPT_DERIVED.get(script, frozenset()):
        for template in _SUBSCRIPT_TEMPLATES:
            assert template.format(key=key) not in source, (
                f"{step_id}: scripts/{script}.py subscripts {key} directly — "
                f"either provide it from the workflow or derive it in-script"
            )


# ---------------------------------------------------------------------------
# Boolean env exports (quality-gate finding 5, #192)
# ---------------------------------------------------------------------------


def _run_steps(workflow_name: str) -> Iterator[tuple[str, str, int]]:
    """Yield (job_name, run_block, step_index) for every run step."""
    workflow = _load_workflow(workflow_name)
    for job_name, job in (workflow.get("jobs") or {}).items():
        for idx, step in enumerate(job.get("steps") or []):
            run = step.get("run")
            if run:
                yield job_name, run, idx


# A GITHUB_ENV export of a boolean-typed key must emit lowercase true/false:
# workflow conditionals compare against the string 'true'/'false', and Python
# str(True) is capitalized — auto-version.yml emitted 'True'/'False' from
# json.load while the steps compared env.CONVERGED == 'false', making the
# safety-net conditionals unreachable.
BOOLEAN_ENV_KEYS = frozenset({"CONVERGED", "AUTO_BUMPED", "DRIFT"})
_LOWERCASE_LITERAL_RE = re.compile(r"""['"](true|false)['"]""")
_JSON_BOOL_ACCESS_RE = re.compile(r"""json\.load\(|\[\s*['"](?:converged|merged)['"]\s*\]""")


def _value_expr(line: str, lines: list[str], key: str) -> str | None:
    """The value-producing expression behind a boolean key export/capture."""
    capture = re.search(rf"^\s*{key}=\$\((.*)\)\s*$", line)
    if capture:
        return capture.group(1)
    write = re.search(rf"""env\.write\(f?["']{key}=(.*?)["']\)$""", line.strip())
    if write:
        return write.group(1)
    echo = re.search(rf'echo\s+"{key}=(\$[A-Za-z_][A-Za-z0-9_]*)"', line)
    if echo:  # indirect: resolve the earlier capture line in the same run block
        ref = echo.group(1).lstrip("${").rstrip("}")
        for other in lines:
            other_capture = re.search(rf"^\s*{re.escape(ref)}=\$\((.*)\)\s*$", other)
            if other_capture:
                return other_capture.group(1)
        return echo.group(1)  # unresolvable — not provably lowercase
    return None


def _emits_lowercase(expr: str) -> bool:
    if ".lower()" in expr:
        return True
    if _LOWERCASE_LITERAL_RE.search(expr):
        return True  # built from explicit lowercase literals
    if re.search(r"\b(True|False)\b", expr):
        return False  # Python bool reprs leak capitalized
    if _JSON_BOOL_ACCESS_RE.search(expr):
        return False  # bool-typed json field without .lower()
    return True  # opaque shell var / workflow expression (rendered lowercase)


@pytest.mark.parametrize(
    ("workflow_name", "job", "run", "idx"),
    [(w, j, r, i) for w in WORKFLOWS for (j, r, i) in _run_steps(w)],
)
def test_boolean_env_exports_are_lowercase(workflow_name: str, job: str, run: str, idx: int) -> None:
    lines = run.splitlines()
    offenders = []
    for key in BOOLEAN_ENV_KEYS:
        for line in lines:
            if not (re.search(rf"\b{key}=", line) or f'"{key}=' in line):
                continue
            expr = _value_expr(line, lines, key)
            if expr is not None and not _emits_lowercase(expr):
                offenders.append(f"[{key}] {line.strip()}")
    assert not offenders, (
        f"{workflow_name}:{job}/step{idx}: boolean env exports must force lowercase "
        f"(str(...).lower()) or use lowercase literals — conditionals compare "
        f"env.CONVERGED == 'false'-style strings: {offenders}"
    )