diff --git a/.github/workflows/static-analysis.yml b/.github/workflows/static-analysis.yml index 263eb340..d1544755 100644 --- a/.github/workflows/static-analysis.yml +++ b/.github/workflows/static-analysis.yml @@ -30,6 +30,9 @@ jobs: - name: Checkout repository uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 with: + # Need the PR base (or push before SHA) so the E2E wait_for_timeout + # guard can diff only *new* call sites (#189). + fetch-depth: 0 persist-credentials: false - name: Set up Python @@ -46,6 +49,40 @@ jobs: - name: Enforce PRKS engineering invariants run: python scripts/check_invariants.py + - name: Fail on new E2E page.wait_for_timeout + # Diff-aware (#189): historical sleeps under tests/e2e/ do not fail + # unrelated PRs. Only newly added / newly-unexempted call sites fail. + # PR base is the immutable event SHA (not the live base-branch tip). + # Zero-before pushes (new/recreated branch) use the empty tree so the + # full tip is scanned — do not merge-base with origin/master after a + # master push (that ref can equal HEAD and skip newly pushed history). + # workflow_dispatch compares against HEAD so grandfathered historical + # waits stay grandfathered (empty tree would fail the whole corpus). + shell: bash + env: + PRKS_PR_BASE_SHA: ${{ github.event.pull_request.base.sha }} + PRKS_PUSH_BEFORE: ${{ github.event.before }} + run: | + set -euo pipefail + empty_tree="4b825dc642cb6eb9a060e54bf8d6927f8d2765b5" + zero_sha="0000000000000000000000000000000000000000" + if [[ "${{ github.event_name }}" == "pull_request" ]]; then + base="${PRKS_PR_BASE_SHA:?missing pull_request.base.sha}" + git fetch --no-tags --depth=1 origin "$base" + elif [[ "${{ github.event_name }}" == "push" ]]; then + if [[ -n "${PRKS_PUSH_BEFORE:-}" && "${PRKS_PUSH_BEFORE}" != "$zero_sha" ]]; then + base="$(git rev-parse --verify --end-of-options "${PRKS_PUSH_BEFORE}")" + else + # Parentless / new-branch push: scan full tip vs empty tree. + base="$empty_tree" + fi + else + # workflow_dispatch (and any other non-PR/non-push event): no-new- + # change baseline so historical E2E waits remain grandfathered. + base="$(git rev-parse --verify --end-of-options HEAD)" + fi + python scripts/check_e2e_wait_for_timeout.py --base "$base" + pyright: permissions: contents: read diff --git a/scripts/check_e2e_wait_for_timeout.py b/scripts/check_e2e_wait_for_timeout.py new file mode 100644 index 00000000..93c8e681 --- /dev/null +++ b/scripts/check_e2e_wait_for_timeout.py @@ -0,0 +1,637 @@ +#!/usr/bin/env python3 +"""Diff-aware guard against new E2E ``page.wait_for_timeout`` calls (#189). + +Historical ``wait_for_timeout`` debt under ``tests/e2e/`` must not fail unrelated +PRs. The checker compares wait call sites on touched paths (and paths removed +from the allowlist) against the comparison base: + +- **New** unapproved calls fail. +- Calls that were **exempt in the base** (marker or path allowlist) but are + no longer exempt fail — even when the sleep line itself is unchanged + (exemption-removal ratchet). +- Historical unexempted calls on untouched paths, or still matched as + base-unexempted on a touched path, pass. + +Call detection uses ``ast.Call`` (so multiline / backslash-continued calls are +found; string literals are not). Exemption markers are accepted only from +Python ``COMMENT`` tokens. + +Exemptions (narrow): + +1. Adjacent marker with a non-empty reason in a comment on the same line or the + previous non-blank line:: + + # prks-allow-wait-for-timeout: absence window for no-request proof + page.wait_for_timeout(250) + +2. Tiny path allowlist below for known intentional timing helpers. Prefer the + marker for one-off cases; keep this set empty unless a helper file is the + documented home for elapsed-time assertions. + +Failure output cites the PRKS E2E "No arbitrary sleeps" policy and points at +observable waits (locator / ``wait_for_function`` / ``wait_for_async`` / route +sync). +""" +from __future__ import annotations + +import argparse +import ast +import io +import os +import re +import subprocess +import sys +import tokenize +from collections import Counter +from dataclasses import dataclass +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] + +E2E_PREFIX = "tests/e2e/" +CHECKER_RELPATH = "scripts/check_e2e_wait_for_timeout.py" +MARKER = "prks-allow-wait-for-timeout:" + +# Tiny allowlist of relative paths (posix) whose entire file may introduce +# ``wait_for_timeout`` without a per-call marker. Ratchet down; no globs. +PATH_ALLOWLIST: frozenset[str] = frozenset() + +# Well-known empty tree — used when a push has no predecessor SHA so the full +# tip is inspected once (greenfield / new-ref) rather than diffing HEAD^HEAD. +EMPTY_TREE_SHA = "4b825dc642cb6eb9a060e54bf8d6927f8d2765b5" + +# CLI/env revision tokens become git argv (list form, not shell). Accept only +# HEAD, hex SHAs, and simple ref names — no whitespace, ranges, or option-like +# values — so argparse/env input cannot inject extra git arguments (S8705). +_SAFE_GIT_REV_RE = re.compile( + r"\A(?:" + r"HEAD" + r"|[0-9a-fA-F]{7,40}" + r"|[A-Za-z0-9][A-Za-z0-9._/-]*" + r")\Z" +) +_FULL_SHA_RE = re.compile(r"\A[0-9a-fA-F]{40}\Z") + + +@dataclass(frozen=True) +class Finding: + path: str + line: int + snippet: str + reason: str + + def render(self) -> str: + return ( + f"E2E-WAIT-001 {self.path}:{self.line}: {self.reason}\n" + f" snippet: {self.snippet.strip()}\n" + f" policy: tests/e2e/AGENTS.md — \"No arbitrary sleeps in E2E\". " + f"Do not add page.wait_for_timeout(...) for post-mutation readiness. " + f"Wait on a locator, wait_for_function (sync predicate), " + f"wait_for_async (async predicate), route/network sync, or other " + f"observable application state instead.\n" + f" exemption: adjacent `{MARKER} ` comment on the same or " + f"previous non-blank line, or a reviewed PATH_ALLOWLIST entry in " + f"scripts/check_e2e_wait_for_timeout.py." + ) + + +@dataclass(frozen=True) +class WaitSite: + lineno: int + snippet: str + key: str + + +class DiscoveryError(RuntimeError): + """Git/base resolution failed; must not look like a clean pass.""" + + +def _git(repo: Path, args: list[str], what: str) -> str: + cmd = ["git", "-C", str(repo), *args] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, check=False) + except OSError as exc: + raise DiscoveryError( + f"{what} failed: could not run `git {' '.join(args)}` " + f"({exc.__class__.__name__})" + ) from exc + if proc.returncode != 0: + detail = [ + line.strip() + for line in (proc.stderr or "").splitlines() + if line.strip() + ] + raise DiscoveryError( + f"{what} failed: `git {' '.join(args)}` exited {proc.returncode}" + + (f" — {detail[0]}" if detail else "") + ) + return proc.stdout or "" + + +def sanitize_git_revision(raw: str) -> str: + """Return a charset-validated git revision token, or raise DiscoveryError. + + ``re.fullmatch`` + returning the match group is the sanitizer boundary for + CLI/env input before it is placed on a git argv list. + """ + value = (raw or "").strip() + if not value: + raise DiscoveryError("invalid --base: empty revision") + if value.startswith("-"): + raise DiscoveryError( + f"invalid --base {value!r}: a revision cannot start with '-' " + "(git would read it as an option)" + ) + if ".." in value: + raise DiscoveryError( + f"invalid --base {value!r}: ranges are not allowed " + "(pass a single revision)" + ) + matched = _SAFE_GIT_REV_RE.fullmatch(value) + if matched is None: + raise DiscoveryError( + f"invalid --base {value!r}: only HEAD, a hex SHA, or a simple " + "ref name is accepted" + ) + return matched.group(0) + + +def resolve_base(repo: Path, explicit: str | None) -> str: + """Pick the comparison revision and resolve it to a 40-char commit SHA. + + Precedence: ``--base`` / argv, ``PRKS_E2E_WAIT_TIMEOUT_BASE``, then ``HEAD`` + (local dirty-tree check). CI for pull requests should pass + ``github.event.pull_request.base.sha``. The empty-tree SHA is accepted as a + synthetic baseline for greenfield pushes. + """ + if explicit: + raw = explicit + else: + raw = (os.environ.get("PRKS_E2E_WAIT_TIMEOUT_BASE") or "").strip() or "HEAD" + safe = sanitize_git_revision(raw) + if safe == EMPTY_TREE_SHA: + return EMPTY_TREE_SHA + out = _git( + repo, + ["rev-parse", "--verify", "--end-of-options", safe], + f"base revision check vs {safe}", + ) + sha = (out.strip().splitlines() or [""])[0].strip() + if _FULL_SHA_RE.fullmatch(sha) is None: + raise DiscoveryError( + f"base revision check vs {safe} failed: rev-parse did not return " + f"a 40-character commit SHA (got {sha!r})" + ) + return sha + + +def iter_wait_sites(source: str) -> list[WaitSite]: + """AST ``.wait_for_timeout(...)`` call sites (ignores strings/comments).""" + try: + tree = ast.parse(source) + except SyntaxError: + return [] + lines = source.splitlines() + sites: list[WaitSite] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if not isinstance(func, ast.Attribute) or func.attr != "wait_for_timeout": + continue + lineno = int(node.lineno) + end = int(getattr(node, "end_lineno", None) or lineno) + snippet = "\n".join(lines[lineno - 1 : end]) + try: + key = ast.unparse(node) + except Exception: + key = snippet.strip() + sites.append(WaitSite(lineno=lineno, snippet=snippet, key=key)) + sites.sort(key=lambda s: s.lineno) + return sites + + +def iter_wait_sites_from_lines(lines: list[str]) -> list[WaitSite]: + return iter_wait_sites("\n".join(lines) + ("\n" if lines else "")) + + +def _marker_reason_from_comment(comment_text: str) -> str | None: + """Parse ``MARKER`` + reason from a tokenize COMMENT string (includes ``#``).""" + idx = comment_text.find(MARKER) + if idx < 0: + return None + reason = comment_text[idx + len(MARKER) :].strip() + return reason or None + + +def comment_marker_reason(line: str) -> str | None: + """Return exemption reason only when MARKER appears in a COMMENT token.""" + if MARKER not in line: + return None + try: + tokens = tokenize.generate_tokens(io.StringIO(line).readline) + for tok in tokens: + if tok.type == tokenize.COMMENT: + reason = _marker_reason_from_comment(tok.string) + if reason is not None: + return reason + except tokenize.TokenError: + return None + return None + + +def line_is_exempt(lines: list[str], lineno_1based: int) -> bool: + """True when a COMMENT marker with reason is on the call or previous non-blank line.""" + if lineno_1based < 1 or lineno_1based > len(lines): + return False + if comment_marker_reason(lines[lineno_1based - 1]) is not None: + return True + for i in range(lineno_1based - 2, -1, -1): + prev = lines[i] + if not prev.strip(): + continue + return comment_marker_reason(prev) is not None + return False + + +def path_is_allowlisted(relpath: str, allowlist: frozenset[str] | None = None) -> bool: + return relpath in (PATH_ALLOWLIST if allowlist is None else allowlist) + + +def _is_e2e_python(path: str | None) -> bool: + return bool( + path + and path.startswith(E2E_PREFIX) + and path.endswith(".py") + ) + + +def parse_path_allowlist_from_source(source: str) -> frozenset[str]: + """Extract ``PATH_ALLOWLIST`` string entries from checker source via AST.""" + try: + tree = ast.parse(source) + except SyntaxError: + return frozenset() + for node in tree.body: + target = None + value = None + if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + target = node.target.id + value = node.value + elif isinstance(node, ast.Assign) and len(node.targets) == 1: + t0 = node.targets[0] + if isinstance(t0, ast.Name): + target = t0.id + value = node.value + if target != "PATH_ALLOWLIST" or value is None: + continue + if not isinstance(value, ast.Call): + return frozenset() + if not isinstance(value.func, ast.Name) or value.func.id != "frozenset": + return frozenset() + if not value.args: + return frozenset() + arg0 = value.args[0] + if not isinstance(arg0, (ast.Set, ast.Tuple, ast.List)): + return frozenset() + out: set[str] = set() + for elt in arg0.elts: + if isinstance(elt, ast.Constant) and isinstance(elt.value, str): + out.add(elt.value) + return frozenset(out) + return frozenset() + + +def load_base_path_allowlist(repo: Path, base_sha: str) -> frozenset[str]: + """PATH_ALLOWLIST as committed at ``base_sha``, or empty if the file is absent.""" + sha = sanitize_git_revision(base_sha) + if sha == EMPTY_TREE_SHA: + return frozenset() + cmd = ["git", "-C", str(repo), "show", f"{sha}:{CHECKER_RELPATH}"] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, check=False) + except OSError as exc: + raise DiscoveryError( + f"base allowlist load failed: could not run git show ({exc.__class__.__name__})" + ) from exc + if proc.returncode != 0: + return frozenset() + return parse_path_allowlist_from_source(proc.stdout or "") + + +def _safe_repo_relpath(relpath: str) -> str: + """Reject path traversal before interpolating into ``git show`` pathspecs.""" + value = (relpath or "").strip().replace("\\", "/") + if ( + not value + or value.startswith("/") + or value.startswith("../") + or "/../" in value + or value.endswith("/..") + or "\0" in value + ): + raise DiscoveryError(f"unsafe repository relative path: {relpath!r}") + return value + + +def read_file_at_revision(repo: Path, base_sha: str, relpath: str) -> list[str] | None: + """Return lines of ``relpath`` at ``base_sha``, or None if it did not exist.""" + sha = sanitize_git_revision(base_sha) + if sha == EMPTY_TREE_SHA: + return None + safe_path = _safe_repo_relpath(relpath) + cmd = ["git", "-C", str(repo), "show", f"{sha}:{safe_path}"] + try: + proc = subprocess.run(cmd, capture_output=True, text=True, check=False) + except OSError as exc: + raise DiscoveryError( + f"read {safe_path} at {sha} failed: could not run git show " + f"({exc.__class__.__name__})" + ) from exc + if proc.returncode != 0: + return None + return (proc.stdout or "").splitlines() + + +def paths_from_name_status_z(raw: str) -> list[tuple[str, str | None]]: + """Parse ``git diff --name-status -z`` into ``(path, rename_source_or_None)``. + + For renames/copies, ``path`` is the post-image and ``rename_source`` is the + pre-image. Other statuses yield ``(path, None)``. + """ + if not raw: + return [] + parts = raw.split("\0") + out: list[tuple[str, str | None]] = [] + i = 0 + n = len(parts) + while i < n: + status = parts[i] + i += 1 + if not status: + continue + if i >= n: + break + first = parts[i] + i += 1 + if not first: + continue + if status[0] in ("R", "C"): + if i >= n: + break + second = parts[i] + i += 1 + if second: + out.append((second, first)) + else: + out.append((first, None)) + return out + + +@dataclass(frozen=True) +class ChangedE2EPath: + path: str + """Path to read at base for historical matching (None ⇒ treat as brand-new).""" + base_read_path: str | None + + +def list_changed_e2e_python(repo: Path, base_sha: str) -> list[ChangedE2EPath]: + """E2E ``*.py`` changes vs ``base_sha``, with rename source tracking.""" + sha = sanitize_git_revision(base_sha) + if sha == EMPTY_TREE_SHA: + # Greenfield: every tracked E2E module is "new" against the empty tree. + raw = _git( + repo, + ["ls-files", "--", E2E_PREFIX], + "E2E tracked-path discovery vs empty tree", + ) + paths = [] + for line in raw.splitlines(): + path = line.strip() + if _is_e2e_python(path): + paths.append(ChangedE2EPath(path=path, base_read_path=None)) + else: + raw = _git( + repo, + [ + "diff", + "--name-status", + "-z", + "--diff-filter=ACMR", + "--end-of-options", + sha, + "--", + E2E_PREFIX, + ], + f"E2E changed-path discovery vs {sha}", + ) + # Also include renames *into* tests/e2e from outside (pre-image is not + # under the pathspec, so name-status scoped to E2E_PREFIX can miss them). + raw_all = _git( + repo, + [ + "diff", + "--name-status", + "-z", + "--diff-filter=ACMR", + "--end-of-options", + sha, + "--", + ], + f"full changed-path discovery vs {sha}", + ) + by_path: dict[str, ChangedE2EPath] = {} + for path, source in paths_from_name_status_z(raw): + if not _is_e2e_python(path): + continue + by_path[path] = ChangedE2EPath(path=path, base_read_path=path) + for path, source in paths_from_name_status_z(raw_all): + if not _is_e2e_python(path): + continue + if source is None: + by_path.setdefault( + path, ChangedE2EPath(path=path, base_read_path=path) + ) + continue + if _is_e2e_python(source): + # Rename within E2E: match historical sites from the old path. + by_path[path] = ChangedE2EPath(path=path, base_read_path=source) + else: + # Rename/copy into E2E from outside: entirely new to the policy. + by_path[path] = ChangedE2EPath(path=path, base_read_path=None) + paths = list(by_path.values()) + + seen = {c.path for c in paths} + for path in list_untracked_e2e_python(repo): + if path not in seen: + paths.append(ChangedE2EPath(path=path, base_read_path=None)) + seen.add(path) + paths.sort(key=lambda c: c.path) + return paths + + +def list_untracked_e2e_python(repo: Path) -> list[str]: + raw = _git( + repo, + ["ls-files", "--others", "--exclude-standard", "--", E2E_PREFIX], + "untracked E2E discovery", + ) + out: list[str] = [] + for line in raw.splitlines(): + path = line.strip() + if path.endswith(".py") and path.startswith(E2E_PREFIX): + out.append(path) + return out + + +def _base_exempt_counters( + base_lines: list[str] | None, + *, + base_allowlisted: bool, +) -> tuple[Counter[str], Counter[str]]: + """Return ``(unexempted, exempted)`` multisets of wait-call keys.""" + unexempted: Counter[str] = Counter() + exempted: Counter[str] = Counter() + if not base_lines: + return unexempted, exempted + for site in iter_wait_sites_from_lines(base_lines): + if base_allowlisted or line_is_exempt(base_lines, site.lineno): + exempted[site.key] += 1 + else: + unexempted[site.key] += 1 + return unexempted, exempted + + +def collect_findings( + repo: Path, + base_sha: str, + *, + path_allowlist: frozenset[str] | None = None, + base_path_allowlist: frozenset[str] | None = None, +) -> list[Finding]: + """Return violations for new or newly-unexempted ``wait_for_timeout`` sites. + + ``base_sha`` must be a 40-character commit SHA from ``resolve_base`` (or the + empty-tree SHA). Optional allowlist kwargs override module/base defaults. + """ + sha = sanitize_git_revision(base_sha) + if _FULL_SHA_RE.fullmatch(sha) is None: + raise DiscoveryError( + f"collect_findings requires a 40-character commit SHA (got {sha!r})" + ) + + current_allow = ( + PATH_ALLOWLIST if path_allowlist is None else frozenset(path_allowlist) + ) + if base_path_allowlist is None: + base_allow = load_base_path_allowlist(repo, sha) + else: + base_allow = frozenset(base_path_allowlist) + + removed_allow = base_allow - current_allow + changed = list(list_changed_e2e_python(repo, sha)) + by_path = {c.path: c for c in changed} + for rel in sorted(removed_allow): + if _is_e2e_python(rel) and rel not in by_path: + by_path[rel] = ChangedE2EPath(path=rel, base_read_path=rel) + + findings: list[Finding] = [] + for rel, changed_path in sorted(by_path.items()): + abs_path = repo / rel + if abs_path.is_file(): + current_text = abs_path.read_text(encoding="utf-8") + current_lines = current_text.splitlines() + else: + continue + if changed_path.base_read_path is None: + base_lines = None + else: + base_lines = read_file_at_revision( + repo, sha, changed_path.base_read_path + ) + unexempted, exempted = _base_exempt_counters( + base_lines, + base_allowlisted=(changed_path.base_read_path or rel) in base_allow, + ) + currently_allowlisted = rel in current_allow + + for site in iter_wait_sites(current_text): + if currently_allowlisted or line_is_exempt(current_lines, site.lineno): + continue + key = site.key + if unexempted[key] > 0: + unexempted[key] -= 1 + continue + if exempted[key] > 0: + exempted[key] -= 1 + findings.append( + Finding( + path=rel, + line=site.lineno, + snippet=site.snippet, + reason=( + "page.wait_for_timeout(...) lost its approved " + "exemption (marker or PATH_ALLOWLIST) while the " + "call remains" + ), + ) + ) + continue + findings.append( + Finding( + path=rel, + line=site.lineno, + snippet=site.snippet, + reason=( + "new page.wait_for_timeout(...) under tests/e2e/ without " + "an approved exemption" + ), + ) + ) + + findings.sort(key=lambda f: (f.path, f.line)) + return findings + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--root", + type=Path, + default=REPO_ROOT, + help="repository root (defaults to this script's parent repository)", + ) + parser.add_argument( + "--base", + default=None, + help=( + "git revision to diff against (default: $PRKS_E2E_WAIT_TIMEOUT_BASE " + "or HEAD). Pull-request CI should pass github.event.pull_request.base.sha." + ), + ) + args = parser.parse_args(argv) + root = args.root.resolve() + + try: + base = resolve_base(root, args.base) + findings = collect_findings(root, base) + except DiscoveryError as exc: + print(f"E2E wait_for_timeout check failed closed: {exc}", file=sys.stderr) + return 2 + + if findings: + for finding in findings: + print(finding.render()) + print() + print( + f"E2E wait_for_timeout check failed: {len(findings)} new " + f"unapproved call site(s) vs {base}" + ) + return 1 + + print(f"E2E wait_for_timeout check: OK (no new unapproved sites vs {base})") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_e2e_wait_for_timeout_guard.py b/tests/test_e2e_wait_for_timeout_guard.py new file mode 100644 index 00000000..99d0c459 --- /dev/null +++ b/tests/test_e2e_wait_for_timeout_guard.py @@ -0,0 +1,555 @@ +"""Regression tests for scripts/check_e2e_wait_for_timeout.py (#189).""" +from __future__ import annotations + +import importlib.util +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +_ROOT = Path(__file__).resolve().parents[1] +_SCRIPT = _ROOT / "scripts" / "check_e2e_wait_for_timeout.py" +_SPEC = importlib.util.spec_from_file_location("prks_check_e2e_wait_for_timeout", _SCRIPT) +assert _SPEC and _SPEC.loader +checker = importlib.util.module_from_spec(_SPEC) +sys.modules[_SPEC.name] = checker +_SPEC.loader.exec_module(checker) + + +def _git(repo: Path, *args: str, check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run( + ["git", "-C", str(repo), *args], + check=check, + capture_output=True, + text=True, + ) + + +def _init_repo(root: Path) -> None: + _git(root, "init") + _git(root, "config", "user.email", "prks-test@example.com") + _git(root, "config", "user.name", "PRKS Test") + _git(root, "checkout", "-b", "master") + + +def _commit_tree(root: Path, files: dict[str, str], message: str) -> None: + for rel, content in files.items(): + path = root / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + _git(root, "add", "--", rel) + _git(root, "commit", "-m", message) + + +HISTORICAL_E2E = ( + "from playwright.sync_api import Page\n" + "\n" + "def test_historical(page: Page):\n" + " page.wait_for_timeout(300)\n" + " assert True\n" +) + + +class AstDetectionTests(unittest.TestCase): + def test_detects_page_and_self_page_calls(self): + src = ( + "page.wait_for_timeout(250)\n" + "self.page.wait_for_timeout(50)\n" + ) + sites = checker.iter_wait_sites(src) + self.assertEqual([s.lineno for s in sites], [1, 2]) + + def test_detects_multiline_backslash_call(self): + src = "page.wait_for_timeout\\\n(250)\n" + sites = checker.iter_wait_sites(src) + self.assertEqual(len(sites), 1) + self.assertEqual(sites[0].lineno, 1) + + def test_detects_parenthesized_multiline_call(self): + src = "page.wait_for_timeout(\n 250\n)\n" + sites = checker.iter_wait_sites(src) + self.assertEqual(len(sites), 1) + self.assertEqual(sites[0].lineno, 1) + + def test_ignores_string_literals_and_comments(self): + src = ( + 'example = ".wait_for_timeout("\n' + '"""page.wait_for_timeout(1)"""\n' + "# page.wait_for_timeout(500)\n" + "page.wait_for_timeout_ms(1)\n" + "wait_for_timeout(1)\n" + ) + self.assertEqual(checker.iter_wait_sites(src), []) + + def test_hash_inside_string_does_not_hide_real_call(self): + src = 'page.locator("#save").click(); page.wait_for_timeout(500)\n' + sites = checker.iter_wait_sites(src) + self.assertEqual(len(sites), 1) + + +class MarkerExemptionTests(unittest.TestCase): + def test_same_line_comment_marker_with_reason(self): + lines = [ + " page.wait_for_timeout(250) # prks-allow-wait-for-timeout: absence window" + ] + self.assertTrue(checker.line_is_exempt(lines, 1)) + + def test_previous_line_comment_marker_with_reason(self): + lines = [ + " # prks-allow-wait-for-timeout: debounce under test", + " page.wait_for_timeout(100)", + ] + self.assertTrue(checker.line_is_exempt(lines, 2)) + + def test_marker_without_reason_is_not_exempt(self): + lines = [ + " # prks-allow-wait-for-timeout:", + " page.wait_for_timeout(100)", + ] + self.assertFalse(checker.line_is_exempt(lines, 2)) + + def test_blank_line_does_not_break_previous_marker(self): + lines = [ + " # prks-allow-wait-for-timeout: settle window", + "", + " page.wait_for_timeout(100)", + ] + self.assertTrue(checker.line_is_exempt(lines, 3)) + + def test_marker_in_assignment_string_does_not_exempt(self): + lines = [ + ' reason = "prks-allow-wait-for-timeout: temporary"', + " page.wait_for_timeout(100)", + ] + self.assertFalse(checker.line_is_exempt(lines, 2)) + self.assertIsNone(checker.comment_marker_reason(lines[0])) + + def test_marker_in_same_line_string_does_not_exempt(self): + lines = [ + ' page.wait_for_timeout(100); x = "prks-allow-wait-for-timeout: no"' + ] + # No COMMENT token carries the marker. + self.assertFalse(checker.line_is_exempt(lines, 1)) + + +class RepoScenarioTests(unittest.TestCase): + def test_new_unapproved_timeout_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + updated = HISTORICAL_E2E + "\n page.wait_for_timeout(999)\n" + (root / "tests/e2e/test_hist.py").write_text(updated, encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual([f.path for f in findings], ["tests/e2e/test_hist.py"]) + self.assertIn("new page.wait_for_timeout", findings[0].reason) + rendered = findings[0].render() + self.assertIn("No arbitrary sleeps", rendered) + self.assertIn("wait_for_async", rendered) + + def test_multiline_new_call_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + updated = HISTORICAL_E2E + "\n page.wait_for_timeout\\\n (999)\n" + (root / "tests/e2e/test_hist.py").write_text(updated, encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual(len(findings), 1) + self.assertIn("new page.wait_for_timeout", findings[0].reason) + + def test_string_literal_mention_does_not_fail(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + updated = HISTORICAL_E2E + '\n note = "page.wait_for_timeout(1)"\n' + (root / "tests/e2e/test_hist.py").write_text(updated, encoding="utf-8") + self.assertEqual(checker.collect_findings(root, base), []) + + def test_historical_unchanged_passes(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + self.assertEqual(checker.collect_findings(root, base), []) + + def test_unrelated_edit_near_historical_timeout_passes(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + edited = HISTORICAL_E2E.replace("assert True", "assert True # unrelated") + (root / "tests/e2e/test_hist.py").write_text(edited, encoding="utf-8") + self.assertEqual(checker.collect_findings(root, base), []) + + def test_approved_marker_passes(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + approved = ( + HISTORICAL_E2E + + "\n" + + " # prks-allow-wait-for-timeout: absence window\n" + + " page.wait_for_timeout(400)\n" + ) + (root / "tests/e2e/test_hist.py").write_text(approved, encoding="utf-8") + self.assertEqual(checker.collect_findings(root, base), []) + + def test_string_marker_does_not_approve_new_call(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + fake = ( + HISTORICAL_E2E + + "\n" + + ' reason = "prks-allow-wait-for-timeout: temporary"\n' + + " page.wait_for_timeout(400)\n" + ) + (root / "tests/e2e/test_hist.py").write_text(fake, encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual(len(findings), 1) + self.assertIn("new page.wait_for_timeout", findings[0].reason) + + def test_removed_marker_while_addition_remains_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + without_marker = HISTORICAL_E2E + "\n page.wait_for_timeout(400)\n" + (root / "tests/e2e/test_hist.py").write_text(without_marker, encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual(len(findings), 1) + self.assertIn("new page.wait_for_timeout", findings[0].reason) + + def test_later_pr_removes_marker_call_unchanged_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + approved = ( + HISTORICAL_E2E + + "\n" + + " # prks-allow-wait-for-timeout: absence window\n" + + " page.wait_for_timeout(400)\n" + ) + _commit_tree( + root, + {"tests/e2e/test_hist.py": approved, "README": "x\n"}, + "approved sleep landed", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + without_marker = HISTORICAL_E2E + "\n" + " page.wait_for_timeout(400)\n" + (root / "tests/e2e/test_hist.py").write_text(without_marker, encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual(len(findings), 1) + self.assertIn("lost its approved exemption", findings[0].reason) + self.assertIn("wait_for_timeout(400)", findings[0].snippet) + + def test_later_pr_removes_same_line_marker_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + approved = ( + "page.wait_for_timeout(250) " + "# prks-allow-wait-for-timeout: debounce under test\n" + ) + _commit_tree( + root, + {"tests/e2e/test_timing.py": approved, "README": "x\n"}, + "same-line marker", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + (root / "tests/e2e/test_timing.py").write_text( + "page.wait_for_timeout(250)\n", + encoding="utf-8", + ) + findings = checker.collect_findings(root, base) + self.assertEqual(len(findings), 1) + self.assertIn("lost its approved exemption", findings[0].reason) + + def test_path_allowlist_then_removal_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/helper_timing.py": "page.wait_for_timeout(1)\n", "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + (root / "tests/e2e/helper_timing.py").write_text( + "page.wait_for_timeout(1)\npage.wait_for_timeout(2)\n", + encoding="utf-8", + ) + helper = frozenset({"tests/e2e/helper_timing.py"}) + self.assertEqual( + checker.collect_findings( + root, + base, + path_allowlist=helper, + base_path_allowlist=helper, + ), + [], + ) + findings = checker.collect_findings( + root, + base, + path_allowlist=frozenset(), + base_path_allowlist=helper, + ) + self.assertEqual(len(findings), 2) + self.assertTrue(all(f.path == "tests/e2e/helper_timing.py" for f in findings)) + reasons = sorted(f.reason for f in findings) + self.assertTrue(any("lost its approved exemption" in r for r in reasons)) + self.assertTrue(any("new page.wait_for_timeout" in r for r in reasons)) + + def test_later_pr_removes_allowlist_call_unchanged_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + { + "tests/e2e/helper_timing.py": "page.wait_for_timeout(1)\n", + "tests/e2e/test_hist.py": HISTORICAL_E2E, + "README": "x\n", + }, + "allowlisted helper landed", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + helper = frozenset({"tests/e2e/helper_timing.py"}) + findings = checker.collect_findings( + root, + base, + path_allowlist=frozenset(), + base_path_allowlist=helper, + ) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0].path, "tests/e2e/helper_timing.py") + self.assertIn("lost its approved exemption", findings[0].reason) + self.assertTrue(all(f.path != "tests/e2e/test_hist.py" for f in findings)) + + def test_rename_into_e2e_from_outside_scans_full_file(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + { + "tests/unit/sleep.py": "page.wait_for_timeout(10)\n", + "README": "x\n", + }, + "outside e2e", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + (root / "tests" / "e2e").mkdir(parents=True, exist_ok=True) + _git(root, "mv", "tests/unit/sleep.py", "tests/e2e/sleep.py") + findings = checker.collect_findings(root, base) + self.assertEqual([f.path for f in findings], ["tests/e2e/sleep.py"]) + self.assertIn("new page.wait_for_timeout", findings[0].reason) + + def test_rename_within_e2e_keeps_historical_match(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/old_name.py": HISTORICAL_E2E, "README": "x\n"}, + "inside e2e", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + _git(root, "mv", "tests/e2e/old_name.py", "tests/e2e/new_name.py") + self.assertEqual(checker.collect_findings(root, base), []) + + def test_rename_within_e2e_allowlist_drop_fails(self): + """Dropping PATH_ALLOWLIST after a within-E2E rename must still fail. + + Base membership is keyed by ``base_read_path`` (pre-rename), not the + post-rename path, so allowlist-drop is not lost across the rename. + """ + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + { + "tests/e2e/old_name.py": "page.wait_for_timeout(1)\n", + "README": "x\n", + }, + "allowlisted e2e helper", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + _git(root, "mv", "tests/e2e/old_name.py", "tests/e2e/new_name.py") + old_allow = frozenset({"tests/e2e/old_name.py"}) + findings = checker.collect_findings( + root, + base, + path_allowlist=frozenset(), + base_path_allowlist=old_allow, + ) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0].path, "tests/e2e/new_name.py") + self.assertIn("lost its approved exemption", findings[0].reason) + + def test_untracked_e2e_module_with_timeout_fails(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree(root, {"README": "x\n"}, "base") + base = _git(root, "rev-parse", "HEAD").stdout.strip() + new_mod = root / "tests" / "e2e" / "test_brand_new.py" + new_mod.parent.mkdir(parents=True, exist_ok=True) + new_mod.write_text("page.wait_for_timeout(10)\n", encoding="utf-8") + findings = checker.collect_findings(root, base) + self.assertEqual([f.path for f in findings], ["tests/e2e/test_brand_new.py"]) + + def test_empty_tree_baseline_scans_tracked_e2e(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "tip", + ) + findings = checker.collect_findings(root, checker.EMPTY_TREE_SHA) + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0].path, "tests/e2e/test_hist.py") + + def test_main_ok_on_clean_tree(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + self.assertEqual(checker.main(["--root", str(root), "--base", "HEAD"]), 0) + + def test_main_fails_on_new_timeout(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree( + root, + {"tests/e2e/test_hist.py": HISTORICAL_E2E, "README": "x\n"}, + "base", + ) + base = _git(root, "rev-parse", "HEAD").stdout.strip() + (root / "tests/e2e/test_hist.py").write_text( + HISTORICAL_E2E + "\npage.wait_for_timeout(1)\n", + encoding="utf-8", + ) + self.assertEqual(checker.main(["--root", str(root), "--base", base]), 1) + + def test_invalid_base_fails_closed(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _init_repo(root) + _commit_tree(root, {"README": "x\n"}, "base") + self.assertEqual( + checker.main(["--root", str(root), "--base", "origin/does-not-exist"]), + 2, + ) + + +class SanitizeRevisionTests(unittest.TestCase): + def test_accepts_head_sha_and_ref_names(self): + self.assertEqual(checker.sanitize_git_revision("HEAD"), "HEAD") + self.assertEqual( + checker.sanitize_git_revision("origin/master"), "origin/master" + ) + sha = "a" * 40 + self.assertEqual(checker.sanitize_git_revision(sha), sha) + self.assertEqual( + checker.resolve_base(Path("."), checker.EMPTY_TREE_SHA), + checker.EMPTY_TREE_SHA, + ) + + def test_rejects_options_ranges_and_metacharacters(self): + for bad in ( + "--upload-pack=evil", + "a..b", + "HEAD;rm", + "origin/master space", + "", + ): + with self.subTest(bad=bad): + with self.assertRaises(checker.DiscoveryError): + checker.sanitize_git_revision(bad) + + +class AllowlistParseTests(unittest.TestCase): + def test_parses_empty_and_populated_frozenset(self): + self.assertEqual( + checker.parse_path_allowlist_from_source( + "PATH_ALLOWLIST: frozenset[str] = frozenset()\n" + ), + frozenset(), + ) + self.assertEqual( + checker.parse_path_allowlist_from_source( + 'PATH_ALLOWLIST: frozenset[str] = frozenset({"tests/e2e/a.py"})\n' + ), + frozenset({"tests/e2e/a.py"}), + ) + + +class CurrentRepoSmokeTests(unittest.TestCase): + def test_current_repo_vs_head_is_clean(self): + sha = checker.resolve_base(_ROOT, "HEAD") + findings = checker.collect_findings(_ROOT, sha) + self.assertEqual( + findings, + [], + "\n".join(f.render() for f in findings), + ) + + +if __name__ == "__main__": + unittest.main()