Unlucky-Launcher/packaging/tests/gpu_shot_ci_selftest.py

#!/usr/bin/env python3
r"""`gpu-evidence.yml` 的**接线自检**（M253 / r1 复审返工）：把「这道门禁怎么才算没被改绿」变成判据。

历史教训（必须留着）：r1 那一版宣称「不可能被改成总是绿」，复审独立造了 19 个「改成总是绿」的变异，
**11 个它照样绿** —— 最要命的是把 `available` 写成恒 `false`：门禁永久关闭，而 push/PR 本来就跳过，
**肉眼与机器都看不出**。原因有三：判据 2/3 本质是字符串存在性检查、判据 5 把自测步整条豁免且不扫
`exit 0`、判据 4 用 `is True` 挡不住 `'true'` / `${{ true }}`。那一版给的是一张**假的安全证书**。

现在六条判据，能结构化的不靠字符串、能跑的不靠读文本：

| # | 判据 | 怎么判 | 抓的是什么 |
|---|------|--------|-----------|
| 1 | 五步必须在场 | 按 `id:` 找（`vectors` / `exitcodes` / `wiring` / `detect` / `judge`），删掉或复制即红；`vectors/exitcodes/wiring/detect` 不许有 `if:`（step 或 job 级）；`exitcodes` 必须真的断言 `expect 0/1/2` 三个码 | 删步、`if: false`、把自测关掉 |
| 2 | 判定必须挂门 | **解析 `if:` 表达式后求值**：`needs.probe.outputs.available` 取 `'true'` 时必须为真、取 `'false'` 与 `''` 时必须为假。认不出的表达式一律判红（宁可假红） | 加 `\|\| always()`、改成 `if: true`、删门、`!` 之类认不出的写法 |
| 3 | 判定用的是探测给的产物 | `run` 里**两种调用形态都必须在场**（`check_gpu_shot.py "$LOG" "$PNG"` 与 `... "$LOG" --expect-refuse`），且 `env` 来自 `needs.probe.outputs.*` | 写死路径、判错文件、把某一分支换成 `true`（那个分支啥都没判就绿） |
| 4 | 跳过路径**真跑** | 把 `detect` 步的 `run` 原文抽到临时目录、用桩 env **实跑 7 种输入**，断言 `$GITHUB_OUTPUT` 里的 `available`、job summary 文案、`::warning::`，以及三条红线必须 `exit≠0` | `available` 恒 false、跳过不留痕、红线退化成跳过 |
| 5 | 禁 `continue-on-error` | 按 **YAML 布尔语义**（`true` / `'true'` / `${{ true }}` / 任意表达式都算开启，只有 `false`/`'false'`/`${{ false }}`/`0` 算关闭） | 带引号、表达式形式 |
| 6 | 不吞退出码 | 扫**所有** `run` 里同时出现 `python`/`python3` 与 `packaging/` 的步（自测步与自检步都在内），`\|\| true` / `; true` / `\|\| exit 0` / 裸 `exit 0` 全算吞 | 自测步与自检步被整条豁免、裸 `exit 0` |

再叠一层：`packaging/tests/gpu_shot_ci_vectors.json` 里 **21 条变异向量**（r1 复审那张表里的 20 条 + 一条
「把判定步的命令换成 `true`」），每条按**解析后的结构**注入（`step_delete` / `job_delete` / `step_set` / `job_set` /
`run_append` / `run_replace`），每条都必须让对应判据当场红、基线必须全绿。

**保证不了什么（别读成免责声明，读成边界）**：
- 这些判据本身不是自守的：能改这个文件的人也能改判据。它防的是**顺手改绿**，不是恶意改绿。
- `if:` 是我自己写的小求值器，不是 GitHub 的语义：只认 `==` / `!=` / `&&` / `||` / 括号 / 字符串 /
  `needs.<job>.outputs.<名>` / `true|false|null` / `always()|success()|failure()|cancelled()`（后四个按
  「本次假定上游成功」取常量）。其余一律认不出 ⇒ 红。`!` 不在支持集里（GitHub 与 Python 的 `!` 优先级
  不同，宁可认不出）。
- 只有 `detect` 那一步是**功能断言**；`vectors` / `exitcodes` 两步只做结构断言（它们的命令在 CI 里作为
  独立步骤真跑，本脚本不重复跑，免得本地没有 `python3` 时把自检本身搞红）。
- 真 Actions 里 `if:` 求值、job skip 的呈现、required status check 与 skipped 的交互，本脚本一概不管。

    python3 packaging/tests/gpu_shot_ci_selftest.py
"""

from __future__ import annotations

import copy
import json
import re
import sys
from pathlib import Path

PACKAGING = Path(__file__).resolve().parent.parent
WORKFLOW = PACKAGING.parent / ".github/workflows/gpu-evidence.yml"
VECTORS_PATH = Path(__file__).resolve().parent / "gpu_shot_ci_vectors.json"
sys.path.insert(0, str(PACKAGING))
sys.path.insert(0, str(Path(__file__).resolve().parent))
from check_workflows_yaml import load_yaml  # noqa: E402
from gpu_shot_ci_probe import probe_problems  # noqa: E402

#: 必须在场、且必须无条件的步骤（按 `id:` 认，不按步骤名认）。
REQUIRED_STEPS = {
    "vectors": "判据实现自测（`--self-test`，19 条向量）",
    "exitcodes": "退出码语义（真 CLI 走 0/1/2）",
    "wiring": "接线自检（本脚本，必须在场）",
    "detect": "产物探测（probe 的唯一步骤）",
    "judge": "真实产物判据（evidence 的步骤）",
}
UNCONDITIONAL = ("vectors", "exitcodes", "wiring", "detect")
AVAILABLE_IDENT = "needs.probe.outputs.available"
CALL_FUNCS = {"always": True, "success": True, "failure": False, "cancelled": False}
SWALLOW_MARKS = ("|| true", "; true", "|| exit 0", "exit 0")
PYTHON_RE = re.compile(r"\bpython3?\b")
_TOKEN_RE = re.compile(r"(?P<string>'[^']*')|(?P<number>\d+)|(?P<ident>[A-Za-z_][\w.\-]*)|(?P<op>==|!=|&&|\|\||[()!])")


class ConditionError(Exception):
    """`if:` 表达式认不出 —— 认不出即判红，绝不默认放过。"""


def coerce_flag(value) -> bool:
    """YAML 里的开关按**真语义**判：`true` / `'true'` / `${{ true }}` 都算开启。"""
    if value is True:
        return True
    if value is False or value is None:
        return False
    if isinstance(value, (int, float)):
        return bool(value)
    return str(value).strip().lower() not in {"false", "${{ false }}", "0", "no", "off", ""}


def evaluate_condition(raw, available: str) -> bool:
    """把 GitHub 的 `if:` 表达式求值。只认一小撮语法，其余抛 `ConditionError`（= 判红）。"""
    text = str(raw).strip()
    if text.startswith("${{") and text.endswith("}}"):
        text = text[3:-2].strip()
    for name, value in CALL_FUNCS.items():
        text = re.sub(rf"\b{name}\s*\(\s*\)", "true" if value else "false", text)
    if re.search(r"\b(" + "|".join(CALL_FUNCS) + r")\s*\(", text):
        raise ConditionError("函数只支持不带参数的形式")
    parts: list[str] = []
    pos = 0
    while pos < len(text):
        if text[pos].isspace():
            pos += 1
            continue
        match = _TOKEN_RE.match(text, pos)
        if not match:
            raise ConditionError(f"认不出的字符 {text[pos]!r}")
        pos = match.end()
        token = match.group()
        if match.lastgroup in ("string", "number"):
            parts.append(token)
        elif match.lastgroup == "ident":
            if token == AVAILABLE_IDENT:
                parts.append("__available")
            elif token in ("true", "false", "null"):
                parts.append({"true": "True", "false": "False", "null": "None"}[token])
            else:
                raise ConditionError(f"认不出的标识符 `{token}`")
        else:
            if token == "!":
                raise ConditionError("不支持 `!`（GitHub 与 Python 的优先级不同，宁可认不出）")
            parts.append({"&&": "and", "||": "or"}.get(token, token))
    source = " ".join(parts)
    try:
        return bool(eval(source, {"__builtins__": {}}, {"__available": available}))  # noqa: S307 - 词表是白名单，见上
    except Exception as error:  # noqa: BLE001 - 任何求值失败都按「认不出」处理
        raise ConditionError(f"求值失败（{error}）：`{source}`") from None


def steps(doc: dict) -> list[tuple[str, dict, dict]]:
    found: list[tuple[str, dict, dict]] = []
    for job_id, job in (doc.get("jobs") or {}).items():
        for step in (job or {}).get("steps") or []:
            if isinstance(step, dict):
                found.append((job_id, job, step))
    return found


def step_by_id(doc: dict, sid: str) -> tuple[str, dict, dict] | None:
    found = [entry for entry in steps(doc) if entry[2].get("id") == sid]
    return found[0] if len(found) == 1 else None


def check_presence(doc: dict) -> list[str]:
    problems: list[str] = []
    counts = {sid: len([e for e in steps(doc) if e[2].get("id") == sid]) for sid in REQUIRED_STEPS}
    for sid, label in REQUIRED_STEPS.items():
        if counts[sid] != 1:
            problems.append(f"`id: {sid}`（{label}）必须恰好一步，实际 {counts[sid]} 步 —— 删掉或复制都算红")
            continue
        job_id, job, step = step_by_id(doc, sid)
        if sid in UNCONDITIONAL and ("if" in step or "if" in job):
            problems.append(f"`id: {sid}`（{label}）带 `if:`（步骤或 job 级）—— 这四步必须无条件跑，否则门禁可被静默关掉")
    entry = step_by_id(doc, "exitcodes")
    if entry:
        asserted = set(re.findall(r"\bexpect\s+([0-9]+)\b", entry[2].get("run") or ""))
        missing = {"0", "1", "2"} - asserted
        if missing:
            problems.append(f"`id: exitcodes` 没有断言退出码 {'/'.join(sorted(missing))}（`expect <码>`）—— 三个码就是门禁的接口")
    return problems


def check_judge_gate(doc: dict) -> list[str]:
    entry = step_by_id(doc, "judge")
    if entry is None:
        return []  # presence 已经报过
    job_id, job, step = entry
    gates: list[tuple[str, object]] = []
    if "if" in job:
        gates.append((f"job `{job_id}` 的 `if`", job.get("if")))
    if "if" in step:
        gates.append(("`id: judge` 步的 `if`", step.get("if")))
    if not gates:
        return ["`judge` 步与它的 job 都没有 `if:` —— 无产物时也会照跑，「跳过」与「通过」分不开"]
    problems: list[str] = []
    for label, raw in gates:
        for available, want in (("true", True), ("false", False), ("", False)):
            try:
                got = evaluate_condition(raw, available)
            except ConditionError as error:
                problems.append(f"{label} 认不出（{error}）—— 认不出即红：`{raw}`")
                break
            if got is not want:
                problems.append(f"{label} 在 `available={available!r}` 时求值为 {got}，应为 {want}：`{raw}`")
    return problems


def check_judge_invocation(doc: dict) -> list[str]:
    entry = step_by_id(doc, "judge")
    if entry is None:
        return []
    run = entry[2].get("run") or ""
    env = entry[2].get("env") or {}
    problems: list[str] = []
    # 两种调用形态都必须在场：正例（日志 + PNG）与负例（--expect-refuse，PNG 可选）。
    # 少一种 ⇒ 那个分支可以「什么都没判」就绿（r1 复审 M20：把正例那一行换成 `true`，旧判据照样绿）。
    for form in ('check_gpu_shot.py "$LOG" "$PNG"', 'check_gpu_shot.py "$LOG" --expect-refuse'):
        if form not in run:
            problems.append(f"`id: judge` 的 `run` 里缺调用形态 `{form}` —— 该分支可以「什么都没判」就绿")
    for name, ref in (("LOG", "needs.probe.outputs.log"), ("PNG", "needs.probe.outputs.png")):
        if ref not in str(env.get(name, "")):
            problems.append(f"`judge` 的 `{name}` 不是来自 `{ref}`（写死的路径会在跳过那条路上判错东西）")
    return problems


def check_skip_path(doc: dict) -> list[str]:
    entry = step_by_id(doc, "detect")
    if entry is None:
        return []  # presence 已经报过
    return probe_problems(entry[2].get("run") or "")


def check_continue_on_error(doc: dict) -> list[str]:
    problems: list[str] = []
    for job_id, job in (doc.get("jobs") or {}).items():
        if coerce_flag((job or {}).get("continue-on-error")):
            problems.append(f"job `{job_id}` 带 `continue-on-error: {job.get('continue-on-error')!r}` —— 判据红也会变绿")
        for step in (job or {}).get("steps") or []:
            if isinstance(step, dict) and coerce_flag(step.get("continue-on-error")):
                problems.append(f"job `{job_id}` 的 `{step.get('id') or step.get('name')}` 带 `continue-on-error: {step.get('continue-on-error')!r}`")
    return problems


def check_no_swallow(doc: dict) -> list[str]:
    problems: list[str] = []
    for job_id, _job, step in steps(doc):
        run = step.get("run") or ""
        if not (PYTHON_RE.search(run) and "packaging/" in run):
            continue
        hits = [mark for mark in SWALLOW_MARKS if mark in run]
        if hits:
            problems.append(f"job `{job_id}` 的 `{step.get('id') or step.get('name')}` 里出现 {'/'.join(f'`{h}`' for h in hits)} —— 退出码被吞掉，判据红也绿")
    return problems


CHECKS = {
    "1 五步必须在场": check_presence,
    "2 判定必须挂门": check_judge_gate,
    "3 判定用探测给的产物": check_judge_invocation,
    "4 跳过路径真跑": check_skip_path,
    "5 禁 continue-on-error": check_continue_on_error,
    "6 不吞退出码": check_no_swallow,
}


def evaluate(doc: dict) -> dict[str, list[str]]:
    if not isinstance(doc, dict) or not isinstance(doc.get("jobs"), dict):
        return {name: ["workflow 不是一个带 jobs 的 YAML 映射"] for name in CHECKS}
    return {name: check(doc) for name, check in CHECKS.items()}


def apply_vector(doc: dict, vector: dict) -> dict:
    """按**解析后的结构**注入变异（不碰文本格式，改动 workflow 的排版不会让向量过期）。"""
    op = vector["op"]
    target = str(vector.get("target", ""))
    if op in ("step_delete", "step_set", "run_append", "run_replace"):
        entry = step_by_id(doc, target)
        if entry is None:
            raise KeyError(f"找不到唯一的 `id: {target}`")
        step = entry[2]
        if op == "step_delete":
            entry[1]["steps"].remove(step)
        elif op == "step_set":
            step[vector["key"]] = vector["value"]
        else:
            text = step.get("run") or ""
            if op == "run_append":
                step["run"] = text + vector["value"]
            else:
                find, replacement, occurrence = vector["find"], vector["replace"], int(vector.get("occurrence", 1))
                pos = -1
                for _ in range(occurrence):
                    pos = text.find(find, pos + 1)
                    if pos < 0:
                        raise KeyError(f"`id: {target}` 的 run 里找不到第 {occurrence} 处 `{find}`")
                step["run"] = text[:pos] + replacement + text[pos + len(find) :]
    elif op == "job_delete" or op == "job_set":
        if target not in (doc.get("jobs") or {}):
            raise KeyError(f"没有 job `{target}`")
        if op == "job_delete":
            del doc["jobs"][target]
        else:
            doc["jobs"][target][vector["key"]] = vector["value"]
    else:
        raise KeyError(f"不认识的变异操作 `{op}`")
    return doc


def main() -> int:
    yaml = load_yaml()
    if not WORKFLOW.is_file():
        print(f"::error file={WORKFLOW}::接线自检的靶子不存在", file=sys.stderr)
        return 1
    doc = yaml.safe_load(WORKFLOW.read_text(encoding="utf-8"))
    failures = 0
    for name, problems in evaluate(doc).items():
        if problems:
            failures += 1
            print(f"FAIL: 判据「{name}」没过：", file=sys.stderr)
            for problem in problems:
                print(f"  - {problem}", file=sys.stderr)
        else:
            print(f"OK: 判据「{name}」")

    vectors = json.loads(VECTORS_PATH.read_text(encoding="utf-8"))["vectors"]
    for vector in vectors:
        label = f"{vector['id']} {vector['label']}"
        try:
            mutated = apply_vector(copy.deepcopy(doc), vector)
        except Exception as error:  # noqa: BLE001 - 向量失效（锚点没了）同样算红，逼着人更新它
            failures += 1
            print(f"FAIL: 变异「{label}」注入失败（向量失效）：{error}", file=sys.stderr)
            continue
        broken = [name for name, problems in evaluate(mutated).items() if problems]
        if vector["expect"] not in broken:
            failures += 1
            print(f"FAIL: 变异「{label}」没被抓到（应红「{vector['expect']}」，实际红的是 {broken or '无'}）", file=sys.stderr)
        else:
            print(f"OK: 变异「{label}」→ 「{vector['expect']}」当场红")

    if failures:
        print(f"FAIL：{failures} 处没过（{WORKFLOW.name}）", file=sys.stderr)
        return 1
    print(f"ok: {len(CHECKS)} 条接线判据全过，{len(vectors)} 条变异向量全部被抓（{WORKFLOW.name}）")
    print("   边界：本脚本防「顺手改绿」，不防「连判据一起改」；`if:` 语义由本脚本的小求值器模拟，不是 GitHub 本身。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())