open-standards/scripts/pr-touches-paths.py

#!/usr/bin/env python3
"""Does this pull request's own diff touch any path its workflow filters on?

Forgejo matches a workflow's `paths:` filter against the **push delta**, not the
pull request's diff. A rebase moves a branch across every commit main has
gained, so its push delta touches nearly every path and every path-filtered
workflow fires. Measured on this repo: a jcp-only change pushed to a fresh
branch ran 2 workflows; the same change force-pushed after a rebase ran 15,
including JPEG-AI whole decode and wai-pack on three platforms. That is what
fills the queue, and it delays everyone's real jobs behind irrelevant ones.

This reads the path list out of the workflow file itself — one source of truth,
rather than a third copy of the list to drift from the two `on:` blocks — and
compares it against `git diff --name-only BASE...HEAD`.

    pr-touches-paths.py <workflow.yml> <base-sha> <head-sha> [--paths-from DIR]... [--wide GLOB]...

With --paths-from, the files of DIR add the globs their headers declare (leading lines
`# paths: <glob>...`, as the verifier steps of .github/wai-conformance/ do), and each --wide
glob of the workflow's list is dropped: it is the forge's filter for those files' paths, wide
because the forge cannot read them, and the gate replaces it with the paths themselves. So a
step that brings a path adds it in its own file, and the gate still answers for exactly the
paths named.

Prints `true` or `false`. **Fails open**: anything it cannot determine prints
`true`, because a wrong `false` silently skips a test and a wrong `true` only
costs runner time. A gate that guesses must guess toward running.
"""
import re
import subprocess
import sys
from pathlib import Path


def paths_of(workflow: Path) -> list[str]:
    """The union of every `paths:` list under `on:` in this workflow.

    Parsed rather than imported from a library: the runner image has no PyYAML
    guaranteed, and the shape here is a flat `paths:` followed by `- 'glob'`
    lines. A list this parser fails to read yields none, which fails open.
    """
    globs: list[str] = []
    in_paths = False
    for line in workflow.read_text().splitlines():
        if re.match(r"\s*paths(-ignore)?:\s*$", line):
            in_paths = True
            continue
        if in_paths:
            stripped = line.strip()
            # A comment or a blank line inside the list does not end it. (It used to:
            # every glob after a list's first comment was dropped, and a pull request
            # touching only those paths was answered `false`.)
            if not stripped or stripped.startswith("#"):
                continue
            m = re.match(r"\s*-\s*'([^']+)'\s*(#.*)?$", line) or re.match(r'\s*-\s*"([^"]+)"\s*(#.*)?$', line)
            if m:
                globs.append(m.group(1))
                continue
            if stripped.startswith("- "):
                globs.append(re.sub(r"\s+#.*$", "", stripped[2:]).strip())
                continue
            in_paths = False
    return globs


HEADER = re.compile(r"^# ([a-z][a-z-]*): ?(.*)$")


def declared_paths(directory: Path) -> "list[str] | None":
    """The globs the files of `directory` declare: each `# paths:` line of a file's leading
    `# key: value` lines. None when the directory holds no file, which fails open."""
    if not directory.is_dir():
        return None
    files = sorted(p for p in directory.iterdir() if p.is_file() and not p.name.startswith("."))
    if not files:
        return None
    globs: list[str] = []
    for f in files:
        for line in f.read_text(encoding="utf-8").splitlines():
            m = HEADER.match(line)
            if not m:
                break
            if m.group(1) == "paths":
                globs.extend(m.group(2).split())
    return globs


def gate_globs(workflow: Path, paths_from: "list[Path]", wide: "list[str]") -> "list[str] | None":
    """The globs the gate answers for: the workflow's, each --wide one replaced by the paths
    the --paths-from files declare. None when that cannot be read, which fails open."""
    globs = paths_of(workflow)
    if not paths_from and not wide:
        return globs
    if not paths_from or not wide or any(w not in globs for w in wide):
        return None  # a wide glob with nothing to stand for, or one the workflow lacks
    extra: list[str] = []
    for d in paths_from:
        found = declared_paths(d)
        if found is None:
            return None
        extra.extend(found)
    return [g for g in globs if g not in wide] + extra


def matches(path: str, glob: str) -> bool:
    """`glob` in the Actions sense: `**` spans separators, `*` does not."""
    rx = []
    i = 0
    while i < len(glob):
        if glob.startswith("**", i):
            rx.append(".*")
            i += 2
        elif glob[i] == "*":
            rx.append("[^/]*")
            i += 1
        elif glob[i] == "?":
            rx.append("[^/]")
            i += 1
        else:
            rx.append(re.escape(glob[i]))
            i += 1
    return re.fullmatch("".join(rx), path) is not None


def main() -> None:
    args = sys.argv[1:]
    paths_from: list[Path] = []
    wide: list[str] = []
    while len(args) > 3:
        if args[-2] == "--paths-from":
            paths_from.insert(0, Path(args[-1]))
        elif args[-2] == "--wide":
            wide.insert(0, args[-1])
        else:
            print("true")  # an argument this script does not read: run it
            return
        args = args[:-2]
    if len(args) != 3:
        print("true")  # cannot tell: run it
        return
    workflow, base, head = Path(args[0]), args[1], args[2]
    if not workflow.is_file() or not base or not head:
        print("true")
        return
    globs = gate_globs(workflow, paths_from, wide)
    if not globs:
        # No filter, or a shape this parser does not read: the workflow wanted
        # to run on everything, or we cannot say it did not.
        print("true")
        return
    try:
        out = subprocess.run(
            ["git", "diff", "--name-only", f"{base}...{head}"],
            capture_output=True, text=True, check=True,
        ).stdout
    except (subprocess.CalledProcessError, OSError):
        print("true")
        return
    changed = [l for l in out.splitlines() if l.strip()]
    if not changed:
        print("true")
        return
    hit = any(matches(p, g) for p in changed for g in globs)
    print("true" if hit else "false")


if __name__ == "__main__":
    main()