felhom.eu/scripts/test_script_tests_gate.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Decoys for scripts/script_tests_gate.py (R-885, R-421). Run: python3 scripts/test_script_tests_gate.py

Each decoy is a fake repo with a scripts/ tree. The gate must REFUSE: a suite that prints "OK" but exits 1 (the label
without the fact); a failing suite three directories deep (scope is a walk, not a list); an empty scripts/ (a gate that
checked nothing). It must PASS the genuine article, and step aside when nested.
"""
import os
import subprocess
import sys
import tempfile

HERE = os.path.dirname(os.path.abspath(__file__))
GATE = os.path.join(HERE, "script_tests_gate.py")

COVERS = {
    "script-tests": ("a suite that PRINTS OK and exits 1; a failing suite three levels deep; an empty scripts/ "
                     "tree; plus the genuine article (must pass) and the nested-run mark (must step aside); R-1297: a "
                     "failing heavy suite on a push that touched its own path, with no list, an empty or a missing list"),
}


def fake_repo(files):
    root = tempfile.mkdtemp()
    os.makedirs(os.path.join(root, "scripts"))
    for rel, body in files.items():
        p = os.path.join(root, "scripts", rel)
        os.makedirs(os.path.dirname(p), exist_ok=True)
        open(p, "w").write(body)
    return root


def gate(root, nested=False):
    env = {k: v for k, v in os.environ.items() if k != "FELHOM_SCRIPT_TESTS_GATE"}
    if nested:
        env["FELHOM_SCRIPT_TESTS_GATE"] = "1"
    return subprocess.run([sys.executable, GATE, root], env=env, stdout=subprocess.PIPE,
                          stderr=subprocess.STDOUT).returncode


CASES = [
    ("prints OK, exits 1", {"test_a.py": "print('OK')\nraise SystemExit(1)\n"}, False, 1),
    ("failing suite 3 levels deep", {"test_ok.py": "pass\n", "x/y/z/test_deep.py": "raise SystemExit(2)\n"}, False, 1),
    ("empty scripts/", {"README": "no tests here\n"}, False, 1),
    ("genuine: two passing suites", {"test_a.py": "pass\n", "sub/test_b.py": "print('fine')\n"}, False, 0),
    ("nested run steps aside", {"test_a.py": "raise SystemExit(1)\n"}, True, 0),
]


def timeouts_ok():
    """R-1209: only the decoy suite gets the longer limit; every other suite keeps 300 s, and a time-out still fails."""
    sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
    import script_tests_gate as g
    return (g.timeout_for("scripts/test_gate_decoys.py") == 600 and g.timeout_for("scripts/test_other.py") == g.TIMEOUT_S == 300)


# R-1297 (A): a heavy suite runs on a CI push only when one of ITS paths changed; with no list (a tag, the nightly run,
# the pre-push hook) or an unreadable or empty list, every suite runs. Each case plants a FAILING heavy suite, so
# „it ran" is rc 1 and „it was skipped" is rc 0 — the verdict, not a log line, says which happened.
CHANGED = "FELHOM_CHANGED_FILES"


def gate_changed(root, changed):
    """changed: None (variable unset), "MISSING" (points at no file), or a list of pushed paths."""
    env = {k: v for k, v in os.environ.items() if k not in ("FELHOM_SCRIPT_TESTS_GATE", CHANGED)}
    if changed == "MISSING":
        env[CHANGED] = os.path.join(root, "no-such-list.txt")
    elif changed is not None:
        lst = os.path.join(root, "pushed.txt")
        open(lst, "w").write("".join(f + "\n" for f in changed))
        env[CHANGED] = lst
    p = subprocess.run([sys.executable, GATE, root], env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    return p.returncode, p.stdout.decode("utf-8", "replace")


HEAVY = {"test_ok.py": "pass\n", "test_gate_decoys.py": "raise SystemExit(1)\n",
         "dooplex-offsite/test_dooplex_offsite.py": "raise SystemExit(1)\n",
         "test_due_checks_gate.py": "raise SystemExit(1)\n"}
FILTER_CASES = [
    # (name, files, changed, want rc, text the output must hold)
    ("filter: no list -> every suite runs", HEAVY, None, 1, "test_gate_decoys.py"),
    ("filter: docs-only push skips the 3", HEAVY, ["documentation/x.md", "STATUS.md"], 0, "SKIP"),
    ("filter: a gate script runs decoys", {"test_ok.py": "pass\n", "test_gate_decoys.py": "raise SystemExit(1)\n"},
     ["scripts/site_gates.py"], 1, "test_gate_decoys.py"),
    ("filter: the due gate runs its suite", {"test_ok.py": "pass\n", "test_due_checks_gate.py": "raise SystemExit(1)\n"},
     ["scripts/due_checks_gate.py"], 1, "test_due_checks_gate.py"),
    ("filter: offsite path runs offsite", {"test_ok.py": "pass\n", "dooplex-offsite/test_dooplex_offsite.py": "raise SystemExit(1)\n"},
     ["scripts/dooplex-offsite/dooplex-offsite.sh"], 1, "test_dooplex_offsite.py"),
    ("filter: hub push skips offsite", {"test_ok.py": "pass\n", "dooplex-offsite/test_dooplex_offsite.py": "raise SystemExit(1)\n"},
     ["hub/internal/web/hosts.go"], 0, "SKIP"),
    ("filter: empty list -> every suite", HEAVY, [], 1, "test_gate_decoys.py"),
    ("filter: missing list -> every suite", HEAVY, "MISSING", 1, "test_gate_decoys.py"),
    ("filter: other suites never skipped", {"test_other.py": "raise SystemExit(1)\n"}, ["documentation/x.md"], 1, "test_other.py"),
]


def workflow_pins():
    """R-1297: the filter is safe only while the nightly run exists and only a branch push sets the list. Read from the
    real workflow and hook, so removing the schedule, or setting the list on every event, turns this red."""
    root = os.path.dirname(HERE)
    wf = open(os.path.join(root, ".gitea", "workflows", "gates.yml"), encoding="utf-8").read()
    hook = open(os.path.join(root, ".githooks", "pre-push"), encoding="utf-8").read()
    problems = []
    if "schedule:" not in wf or "- cron:" not in wf:
        problems.append("gates.yml has no nightly schedule — the filtered suites would run only when touched")
    arm = wf.find("push:refs/heads/*)")
    sets = [i for i in range(len(wf)) if wf.startswith('FELHOM_CHANGED_FILES=', i)]
    if not sets or arm < 0 or any(i < arm or i > wf.find(";;", arm) for i in sets):
        problems.append("gates.yml sets FELHOM_CHANGED_FILES outside the branch-push arm")
    if "total_commits" not in wf:
        problems.append("gates.yml does not treat a trimmed commits array as an unknown list")
    if "FELHOM_CHANGED_FILES" in hook:
        problems.append("the pre-push hook sets the suite filter — the hook must run every suite")
    for p in problems:
        print("FAIL workflow pin: " + p)
    if not problems:
        print("ok   workflow pins (nightly run, branch-push-only list, trimmed payload, hook unfiltered)")
    return not problems


def main():
    bad = 0
    bad += not workflow_pins()
    for name, files, changed, want, text in FILTER_CASES:
        got, out = gate_changed(fake_repo(files), changed)
        ok = got == want and text in out
        bad += not ok
        print("%s %-38s want rc=%d got rc=%d%s" % ("ok  " if ok else "FAIL", name, want, got,
                                                  "" if text in out else " (no %r in output)" % text))
    t_ok = timeouts_ok()
    bad += not t_ok
    print("%s %-32s" % ("ok  " if t_ok else "FAIL", "per-suite time limits (R-1209)"))
    for name, files, nested, want in CASES:
        got = gate(fake_repo(files), nested)
        ok = got == want
        bad += not ok
        print("%s %-32s want rc=%d got rc=%d" % ("ok  " if ok else "FAIL", name, want, got))
    print("script-tests decoys: %d failed of %d" % (bad, len(CASES) + len(FILTER_CASES) + 1))
    return 1 if bad else 0


if __name__ == "__main__":
    sys.exit(main())