scripts.gate_parallel_safety
Reference › scripts › gate_parallel_safety
scripts/gate_parallel_safety.py is a deterministic static analysis of the test tree for patterns that make parallel test runs flaky. It is a CLI gate run by a human or a gate step; nothing imports it. Its only internal import is orchestrata_ultimate.core.gate_waivers, and _is_waived honours a gate-ok waiver comment on the offending line or the line above it.
It contains numbered gates, collected in the GATES table. The ones visible in the API are: Gate 01 (gate_01_cypher_create), Cypher CREATE on uniqueness-constrained labels must carry a test_id; Gate 02 (gate_02_detach_delete), DETACH DELETE and global relationship deletes must be scoped to the test; Gate 03 (gate_03_session_fixtures), session-scope fixtures that mutate shared state must be worker-scoped; and Gate 04, a Redis key scan. Gate 04 is the largest part of the file. It parses each test module with ast, tracks which names are Redis handles and which values derive from a test id through a _Scope alias table, follows seeds from call sites into callee parameters (_seeds_from_call, _resolve_module_calls) and asks whether each Redis key expression derives from the test id. The older line-based detector is kept as _gate_04_line_based_scan.
Options: --gate runs one gate, --json prints machine-readable results, --strict makes advisory gates blocking, and --tests-dir scans another tests tree (used to test the gate against a seeded sandbox). It reads the repository test files and writes nothing. It prints PASS, WARN or FAIL per gate with file:line details, and exits non-zero when blocking violations remain. Advisory gates only warn unless --strict is given.
At a glance
| Source | scripts/gate_parallel_safety.py |
| Lines of code | 1546 |
| Classes | 1 |
| Functions | 32 |
| Imports (internal) | orchestrata_ultimate.core.gate_waivers |
| Imported by | none |
| Tests that import it | none |
Class diagram
Dependency diagram
API
Constants
| Name | Type | Value |
|---|---|---|
_COMPOUND_BODY_FIELDS |
dict[type, tuple[str, ...]] |
{ast.If: ('body', 'orelse'), ast.For: ('body', 'orelse'), ast.AsyncFor: ('body'… |
Classes
_Scope
Line 441
One function's alias tables. redis_aliases/testid_aliases hold bare identifiers bound to a Name target (r = redis_client -> "r"); the *_qualified_* sets hold dotted paths…
| Method | Signature | Summary |
|---|---|---|
__init__ |
__init__(parent: _Scope | None = None) -> None |
|
snapshot |
snapshot() -> _Scope |
An independent copy of this scope's CURRENT alias state. |
Functions
| Function | Signature | Summary |
|---|---|---|
_is_waived |
_is_waived(lines: list[str], lineno: int) -> bool |
True if the 1-based lineno (or the line above) carries a gate-ok waiver. |
_scan_cypher_create |
_scan_cypher_create(content: str, filename: str) -> list[dict] |
Return CREATE-without-test_id violations on constrained labels in content. |
gate_01_cypher_create |
gate_01_cypher_create() |
Cypher CREATE on uniqueness-constrained labels must include test_id. |
_docstring_line_numbers |
_docstring_line_numbers(source: str) -> set[int] |
1-based line numbers occupied by docstrings / bare string-expression statements. |
_scan_detach_delete |
_scan_detach_delete(source: str, filename: str) -> list[dict] |
Return unscoped DETACH / global relationship DELETE violations in source. |
gate_02_detach_delete |
gate_02_detach_delete() |
DETACH DELETE / global relationship DELETE must be scoped to this test. |
gate_03_session_fixtures |
gate_03_session_fixtures() |
Session-scope fixtures that mutate shared state must be worker-scoped. |
_chain_path |
_chain_path(expr: ast.expr) -> list[str] | None |
The dotted-name path of a pure Name/Attribute chain (holder.conn -> ["holder", "conn"]), or None if the chain bottoms out in anything else (a Call, a Subscript, a literal, ...). |
_is_redis_expr |
_is_redis_expr(expr: ast.expr, scope: _Scope) -> bool |
True if expr denotes a Redis client/handle, resolved through scope. |
_is_testid_derived |
_is_testid_derived(expr: ast.expr, scope: _Scope) -> bool |
True if expr — a Redis KEY expression — actually derives from test_id, resolved through scope: directly, through a local variable/attribute bound to something test_id-derived, through an… |
_seed_params_from_arguments |
_seed_params_from_arguments(args: ast.arguments) -> tuple[set[str], set[str]] |
(redis-seeded param names, test_id-seeded param names) from a bare ast.arguments node alone — test_id itself, and any parameter literally named like a known Redis fixture. |
_seed_params |
_seed_params(func: ast.FunctionDef | ast.AsyncFunctionDef) -> tuple[set[str], set[str]] |
(redis-seeded params, test_id-seeded params) from func's own signature alone — test_id itself, and any parameter literally named like a known Redis fixture. |
_expr_calls |
_expr_calls(node: ast.AST) -> list[ast.Call] |
Call nodes reachable from node itself and its descendants, without crossing into a NESTED statement (those are walked separately, in physical order, by _process_scope's recursion) or a nested… |
_stmt_own_calls |
_stmt_own_calls(stmt: ast.stmt) -> list[ast.Call] |
Call nodes that live directly in stmt's own expression fields (its condition, its RHS, its decorators-free body-less parts) — NOT inside any NESTED statement (those are walked separately, in… |
_find_lambdas |
_find_lambdas(stmt: ast.stmt) -> list[ast.Lambda] |
Every Lambda reachable from stmt's own expression fields — including a lambda nested inside another lambda — without crossing into a nested FunctionDef/AsyncFunctionDef (those get their own… |
_bind_alias_target |
_bind_alias_target(target: ast.expr, value: ast.expr, scope: _Scope) -> None |
Grow scope from one target = value-shaped binding — a bare Name target becomes a plain alias, a dotted Attribute target becomes a qualified alias (holder.conn = redis_client). |
_process_scope |
_process_scope(body: list[ast.stmt], scope: _Scope, calls_out: list[tuple[ast.Call, _Scope]], scopes_by_node: dict[ast.AST, _Scope]) -> None |
Walk body's statements in physical order, growing scope's alias tables from assignments and recording every call site — SNAPSHOTTED at the moment it is seen, never a live reference to the… |
_seeds_from_call |
_seeds_from_call(call: ast.Call, caller_scope: _Scope, callee: ast.FunctionDef | ast.AsyncFunctionDef) -> tuple[frozenset[str], frozenset[str]] |
The (redis-seeded, test_id-seeded) parameter names this ONE call site supplies to callee, resolved purely against caller_scope. |
_resolve_module_calls |
_resolve_module_calls(tree: ast.Module) -> list[tuple[ast.Call, _Scope]] |
Every Call in tree, paired with the scope active at its call site — resolved through a CALL-SITE-ISOLATED interprocedural analysis (2026-08-12 rewrite, replacing a per-parameter UNION across all… |
_gate_04_line_based_scan |
_gate_04_line_based_scan(lines: list[str]) -> list[tuple[int, str]] |
The ORIGINAL (pre-AST) Gate 0.4 detector, MINUS the TEST_ID_NEARBY text-window (deleted everywhere in this gate — see the block comment above the Gate 0.4 constants) — the fallback for a… |
_write_target |
_write_target(call: ast.Call) -> tuple[str, ast.expr, int] | None |
(verb, receiver-expr, physical write-line) for a REDIS-shaped write call, covering two receiver shapes: the ordinary <expr>.<verb>(...) static attribute access, and ``getattr(<expr>… |
_key_exprs |
_key_exprs(call: ast.Call, verb: str) -> list[ast.expr] |
The expression(s) that determine the Redis KEY(S) call touches, given its already-resolved write verb (_write_target's first element — never re-derived from call.func here, since that… |
gate_04_redis_keys |
gate_04_redis_keys() |
Redis write operations must namespace keys with test_id. |
_enclosing_region |
_enclosing_region(lines: list[str], i: int) -> tuple[str, str] | None |
(signature, body) of the function enclosing line i, or None. |
_enclosing_uses_tmp |
_enclosing_uses_tmp(lines: list[str], i: int) -> bool |
True if the function enclosing line i is tmp-scoped: it takes a Path parameter (tmp_path or a wrapper like root/temp_project, or any parameter explicitly annotated : Path — a… |
_build_fixture_tmp_index |
_build_fixture_tmp_index() -> re.Pattern | None |
Compile a word-matcher for every fixture whose value is tmp-backed. |
_pure_string_content_lines |
_pure_string_content_lines(source: str) -> set[int] |
1-based line numbers whose ENTIRE content is string-literal text. |
gate_05_filesystem |
gate_05_filesystem() |
Filesystem writes must use the isolation fixture, tmp_path, or a tmp-derived path. |
_waits_for_port_to_be_free |
_waits_for_port_to_be_free(node: ast.AST) -> bool |
True for connect_ex(...) != 0 — "nothing is accepting, so it is mine". |
_wait_for_port_loops |
_wait_for_port_loops(source: str, filename: str) -> list[dict] |
Loops that poll a port for availability, rather than taking one. |
gate_06_wait_for_port |
gate_06_wait_for_port() |
A test must take a free port, not wait for a shared one to look free. |
main |
main() -> int |
Source
#!/usr/bin/env python3
"""Gate: Parallel Safety Audit — deterministic static analysis for test flakiness.
Five gates that detect guaranteed parallel collisions without running tests.
Each gate finds patterns that WILL cause failures under parallel execution
(threadpool runner, multiple workers sharing the graph/Redis/filesystem).
Usage:
python scripts/gate_parallel_safety.py # all gates, exit 1 on violation
python scripts/gate_parallel_safety.py --gate 0.1 # single gate
python scripts/gate_parallel_safety.py --json # JSON output for CI
Exit code 0 = all gates pass, 1 = violations found.
Waivers: a finding on a line carrying (or directly below) a ``# gate-ok:
<reason>`` comment is suppressed. This exists for tests whose FIXTURE DATA is
deliberately-unsafe Cypher/source fed to a tool under test (preflight checker,
isolation auditor, code reviewer) — the string is analyzed, never executed. A
reason is required; a bare ``# gate-ok:`` does not waive.
"""
import argparse
import ast
import io
import json
import re
import sys
import tokenize
from pathlib import Path
from orchestrata_ultimate.core.gate_waivers import waived_lines
REPO_ROOT = Path(__file__).resolve().parent.parent
TESTS_DIR = REPO_ROOT / "tests"
SRC_DIR = REPO_ROOT / "src" / "orchestrata_ultimate"
# ── Gate 0.1: Cypher CREATE on constrained labels without test_id ───────────
CONSTRAINED_LABELS = [
"File",
"Project",
"Module",
"Intent",
"Session",
"Spec",
"Artifact",
"Portfolio",
"WorkItem",
"Checkpoint",
"Verification",
"Run",
"Requirement",
]
CREATE_RE = re.compile(r"CREATE\s*\(\s*\w*\s*:(\w+)\s*\{([^}]*)\}", re.IGNORECASE)
TEST_ID_IN_PROPS = re.compile(r"test_id\s*:")
# ``# gate-ok: <reason>`` on the flagged line (or the line directly above it)
# waives that one finding. The reason is mandatory — ``\S`` refuses bare tags.
_WAIVER_RE = re.compile(r"#\s*gate-ok:\s*\S")
def _is_waived(lines: list[str], lineno: int) -> bool:
"""True if the 1-based *lineno* (or the line above) carries a gate-ok waiver."""
for idx in (lineno - 1, lineno - 2):
if 0 <= idx < len(lines) and _WAIVER_RE.search(lines[idx]):
return True
return False
def _scan_cypher_create(content: str, filename: str) -> list[dict]:
"""Return CREATE-without-test_id violations on constrained labels in *content*.
Matches inside docstrings / bare string-expression statements are prose, not
Cypher, and are skipped (Cypher value-strings passed to ``.run(...)`` are
still scanned). See :func:`_docstring_line_numbers`.
"""
doc_lines = _docstring_line_numbers(content)
content_lines = content.split("\n")
found: list[dict] = []
for match in CREATE_RE.finditer(content):
label = match.group(1)
props = match.group(2)
if label in CONSTRAINED_LABELS and not TEST_ID_IN_PROPS.search(props):
lineno = content[: match.start()].count("\n") + 1
if lineno in doc_lines or _is_waived(content_lines, lineno):
continue
found.append(
{
"file": filename,
"line": lineno,
"label": label,
"detail": f"CREATE ({label}) without test_id in props",
}
)
return found
def gate_01_cypher_create():
"""Cypher CREATE on uniqueness-constrained labels must include test_id."""
violations = []
for f in sorted(TESTS_DIR.rglob("test_*.py")):
try:
content = f.read_text()
except Exception:
continue
violations.extend(_scan_cypher_create(content, str(f.relative_to(REPO_ROOT))))
return violations
# ── Gate 0.2: Cypher DETACH DELETE / global edge DELETE without scope ────────
# A delete is "scoped" if the statement (or the params dict on the following
# lines) references a Cypher $param, test_id, _TEST_, or project_id. An unscoped
# `MATCH (a:Action) DETACH DELETE a` or `MATCH ()-[r]->() DELETE r` wipes other
# concurrent tests' data on the shared graph.
_DELETE_LINE_RE = re.compile(r"DETACH\s+DELETE|\)\s*DELETE\s+\w", re.IGNORECASE)
_SCOPE_TOKEN_RE = re.compile(r"\$\w+|test_id|_TEST_|project_id|tmp_", re.IGNORECASE)
def _docstring_line_numbers(source: str) -> set[int]:
"""1-based line numbers occupied by docstrings / bare string-expression
statements. These are prose, not Cypher, so they must not be scanned for
delete patterns — a docstring that *mentions* "DETACH DELETE" is not a delete.
String *values* (e.g. a Cypher query passed to ``.run("...")``) are NOT
docstrings — they are ``Call`` args, not ``Expr`` statements — so they stay
scannable.
"""
try:
tree = ast.parse(source)
except SyntaxError:
return set()
doc_lines: set[int] = set()
for node in ast.walk(tree):
if (
isinstance(node, ast.Expr)
and isinstance(node.value, ast.Constant)
and isinstance(node.value.value, str)
):
start = node.value.lineno
end = getattr(node.value, "end_lineno", start)
doc_lines.update(range(start, end + 1))
return doc_lines
def _scan_detach_delete(source: str, filename: str) -> list[dict]:
"""Return unscoped DETACH / global relationship DELETE violations in *source*.
``#`` comments and docstring prose are skipped; everything else (including
Cypher value-strings) is scanned. A delete is scoped when a
$param/test_id/_TEST_/project_id/tmp_ token appears in the context window.
"""
lines = source.split("\n")
doc_lines = _docstring_line_numbers(source)
found: list[dict] = []
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("#") or (i + 1) in doc_lines:
continue
if not _DELETE_LINE_RE.search(line):
continue
if _is_waived(lines, i + 1):
continue
context = "\n".join(lines[max(0, i - 6) : i + 4])
if not _SCOPE_TOKEN_RE.search(context):
found.append(
{
"file": filename,
"line": i + 1,
"detail": f"Unscoped delete (no $param/test_id/project_id): {stripped[:90]}",
}
)
return found
def gate_02_detach_delete():
"""DETACH DELETE / global relationship DELETE must be scoped to this test.
Catches single-line (`MATCH (...) DETACH DELETE x`) and multi-line forms, and
includes the lines AFTER the delete so a `{"prefix": test_id}` params dict on
the next line counts as scoping. Docstrings/comments are skipped so prose
mentioning the keywords is not a false positive.
"""
violations = []
for f in sorted(TESTS_DIR.rglob("test_*.py")):
try:
source = f.read_text()
except Exception:
continue
violations.extend(_scan_detach_delete(source, str(f.relative_to(REPO_ROOT))))
return violations
# ── Gate 0.3: Session-scope fixtures audit ──────────────────────────────────
def gate_03_session_fixtures():
"""Session-scope fixtures that mutate shared state must be worker-scoped."""
violations = []
conftest_path = TESTS_DIR / "conftest.py"
if not conftest_path.exists():
return violations
content = conftest_path.read_text()
# The _cleanup_test_prefix fixture must use ORCHESTRATA_WORKER_SLOT
if "_cleanup_test_prefix" in content:
if "ORCHESTRATA_WORKER_SLOT" not in content or "worker_slot" not in content:
violations.append(
{
"file": "tests/conftest.py",
"line": 253,
"detail": "_cleanup_test_prefix must scope cleanup to ORCHESTRATA_WORKER_SLOT",
}
)
# Check knowledge_graph/conftest.py for pooled drivers
kg_conftest = TESTS_DIR / "knowledge_graph" / "conftest.py"
if kg_conftest.exists():
kg_content = kg_conftest.read_text()
if (
'scope="session"' in kg_content
and "max_connection_pool_size" not in kg_content
):
violations.append(
{
"file": "tests/knowledge_graph/conftest.py",
"line": 1,
"detail": "Session-scope graph driver without pool size limit",
}
)
return violations
# ── Gate 0.4: Redis writes without test_id namespacing ──────────────────────
#
# AST data-flow, not text-window matching (2026-08-12 rewrite). The prior
# TEST_ID_NEARBY window matched any of a dozen loosely-related tokens
# anywhere in a ~13-line span around the write — including inside the
# test's OWN NAME (``test_resume_session_inval|id`` matches the window's own
# ``test_[a-z_]*id`` alternative) and regardless of whether the token had
# anything to do with the key being written. That is a presence check, not a
# safety proof: tree-wide it suppressed every one of 25 real candidates
# (measured: old rule 7 flagged, new-with-window 0). A second presence check
# one level down — "is ``test_id`` a parameter of, or ``ast.walk``-reachable
# from, the enclosing ``test_*`` function" — had the same defect: it exempted
# a hardcoded write 50+ lines from an unrelated ``test_id`` mention, and it
# exempted a write merely because a NESTED closure referenced ``test_id``
# somewhere else in the function.
#
# The replacement resolves, per call site: (a) is the RECEIVER a Redis
# object — via local alias tracking (``r = redis_client``), attribute-target
# tracking (``holder.conn = redis_client``), pipeline construction
# (``pipe = redis_client.pipeline()``), constructor-argument tainting
# (``GovernanceCache(redis=redis_state_manager)``), ``await``, and one hop of
# interprocedural propagation into helpers called with a Redis-rooted
# argument (``_seed(redis_client)`` taints ``_seed``'s ``client`` parameter);
# and (b) does the KEY EXPRESSION actually derive from ``test_id`` — through
# a local variable, an f-string/concatenation, or a helper argument — using
# the SAME alias machinery, seeded from ``test_id`` instead of a Redis root.
# A write is a violation unless (a) is false, (b) is true, or the write line
# carries a reasoned ``# gate-ok:`` waiver.
# Bare names known (from tests/conftest.py's redis_client / async_redis_client
# / redis_state_manager fixtures, plus the raw redis-py entry points a test
# occasionally imports directly) to BE a Redis client/handle — the roots that
# seed alias tracking below. Matched by resolved identifier, never by regex
# over source text.
REDIS_ROOT_NAMES = frozenset(
{
"redis_client",
"async_redis_client",
"redis_state_manager",
"redis",
"redis_lib",
"async_redis",
"rc",
"_redis",
"redis_mgr",
# Added 2026-08-12 from a live-tree scan: 21 write sites were invisible
# because their receiver was a bare name or attribute leaf absent here.
# `flushdb` had 0% coverage for exactly this reason — 5 live sites, all
# spelled `client.flushdb()` / `s._client.flushdb()`.
"client",
"_client",
"aclient",
"conn",
"_conn",
}
)
# An attribute chain whose LAST segment is one of these is Redis-tainted
# regardless of what it is rooted at — ``ctx.redis_client``,
# ``session_manager.redis_client``, ``engine_extended.redis``,
# ``handler._redis``, ``sla_tracker._redis`` are all real receiver shapes
# measured in the live tree (tests/session, tests/kernel, tests/orchestrator,
# tests/governance).
REDIS_ATTR_LEAVES = frozenset(
{
"redis_client",
"async_redis_client",
"redis",
"_redis",
"redis_mgr",
"redis_conn",
# See the note on REDIS_ROOT_NAMES: `rb.client`, `gw._ring_buffer.client`,
# `queue._client`, `s._client` are all real receivers in tests/.
"client",
"_client",
"conn",
"_conn",
}
)
# Every Redis command that mutates keyspace state: the canonical set plus
# ``hsetnx`` — not in the canonical list, but called on a genuine
# ``redis_client`` receiver in the live tree (tests/factory/
# test_dispatch_gates.py) — added from that measurement, not assumed
# (enumerated with a small AST-based recon script over every test_*.py Call
# node, not guessed; ``append``/``delete``/... are safe to include broadly
# because a match ALSO requires the receiver to resolve as Redis-tainted —
# ``calls.append(x)`` on an ordinary list never does).
REDIS_WRITE_VERBS = frozenset(
{
"set",
"setex",
"setnx",
"mset",
"getset",
"append",
"incr",
"decr",
"hset",
"hmset",
"hsetnx",
# Added 2026-08-12: every one below was reproduced as an UNFLAGGED
# hazard against a receiver the gate already resolves, so the verb list
# alone was the discriminator. `psetex` was captured live on db0.
"psetex",
"setrange",
"setbit",
"lset",
"linsert",
"smove",
"zincrby",
"getdel",
"hincrbyfloat",
"incrby",
"decrby",
"incrbyfloat",
"xdel",
"pfadd",
"geoadd",
"restore",
"move",
"bitfield",
"rpoplpush",
"lmove",
"zpopmin",
"zpopmax",
"zremrangebyrank",
"zremrangebyscore",
"sinterstore",
"sunionstore",
"sdiffstore",
"zunionstore",
"zinterstore",
"hdel",
"hincrby",
"lpush",
"rpush",
"lpop",
"rpop",
"lrem",
"ltrim",
"sadd",
"srem",
"spop",
"zadd",
"zrem",
"xadd",
"xtrim",
"delete",
"unlink",
"expire",
"pexpire",
"persist",
"rename",
"copy",
"flushdb",
"flushall",
"execute_command",
"eval",
"evalsha",
}
)
# Bounded interprocedural recursion depth (see ``_resolve_module_calls``):
# how many CALL-SITE-ISOLATED hops a "this parameter received a Redis-rooted
# / test_id-derived argument" derivation is allowed to chain through
# module-level helper calls before the recursion stops (also the memoization
# net's last line of defense against a (mutually) recursive helper). Test
# files are small; this is generous headroom for a multi-hop delegation
# chain, not a performance knob.
_TAINT_FIXED_POINT_ROUNDS = 6
# The SyntaxError/ValueError fallback (``_gate_04_line_based_scan``) cannot
# build an AST, so it cannot resolve aliases — it derives its regex from the
# SAME canonical name/verb sets above rather than hand-copying a second
# receiver list, so there is exactly one place that says what counts as a
# Redis write.
_REDIS_LINE_SCAN_RE = re.compile(
r"\b(?:"
+ "|".join(re.escape(n) for n in sorted(REDIS_ROOT_NAMES | REDIS_ATTR_LEAVES))
+ r")\b\s*\.\s*(?:"
+ "|".join(re.escape(v) for v in sorted(REDIS_WRITE_VERBS))
+ r")\s*\(",
re.IGNORECASE,
)
def _chain_path(expr: ast.expr) -> list[str] | None:
"""The dotted-name path of a pure Name/Attribute chain (``holder.conn``
-> ``["holder", "conn"]``), or ``None`` if the chain bottoms out in
anything else (a Call, a Subscript, a literal, ...). Used both to read a
chain's own root name / leaf attribute, and to match a known qualified
alias recorded by an attribute-target assignment."""
parts: list[str] = []
node = expr
while isinstance(node, ast.Attribute):
parts.append(node.attr)
node = node.value
if isinstance(node, ast.Name):
parts.append(node.id)
parts.reverse()
return parts
return None
class _Scope:
"""One function's alias tables. ``redis_aliases``/``testid_aliases`` hold
bare identifiers bound to a ``Name`` target (``r = redis_client`` ->
``"r"``); the ``*_qualified_*`` sets hold dotted paths bound to an
``Attribute`` target (``holder.conn = redis_client`` ->
``"holder.conn"``). A nested function's scope starts as a COPY of its
enclosing function's tables — real closure semantics, a nested ``def``
genuinely can read the outer test's ``test_id`` — so growth inside the
nested function never leaks back out to the parent."""
__slots__ = (
"redis_aliases",
"redis_qualified_aliases",
"testid_aliases",
"testid_qualified_aliases",
)
def __init__(self, parent: _Scope | None = None) -> None:
self.redis_aliases: set[str] = set(parent.redis_aliases) if parent else set()
self.redis_qualified_aliases: set[str] = (
set(parent.redis_qualified_aliases) if parent else set()
)
self.testid_aliases: set[str] = set(parent.testid_aliases) if parent else set()
self.testid_qualified_aliases: set[str] = (
set(parent.testid_qualified_aliases) if parent else set()
)
def snapshot(self) -> _Scope:
"""An independent copy of this scope's CURRENT alias state. Used to
freeze the scope paired with a recorded call site — ``_Scope`` is
mutated in place as ``_process_scope`` keeps walking LATER
statements in the same function, so a call recorded by reference
would silently pick up alias bindings created after it (flow
INsensitivity: a key variable rebound to a test_id-derived value
later in the function would retroactively exempt an earlier write
that used the same name while it still held a hardcoded value)."""
return _Scope(parent=self)
def _is_redis_expr(expr: ast.expr, scope: _Scope) -> bool:
"""True if *expr* denotes a Redis client/handle, resolved through
*scope*. Structural, not textual: a Name/Attribute chain is tainted if
its root name is a known Redis fixture or a locally-tracked alias, if a
PREFIX of its dotted path was bound by an attribute-target assignment, or
if its own leaf attribute is a known Redis-holding name; a Subscript
defers to its base; an Await defers to its inner expression; a Call is
tainted if it is ``<redis-expr>.pipeline(...)`` or if ANY argument passed
to it is itself Redis-tainted — covering both
``RedisRingBuffer(redis_client)``-style wrapping construction and one hop
of ``_seed(redis_client)`` delegation (the latter via the interprocedural
seeding in ``_resolve_module_calls``, not this function alone)."""
if isinstance(expr, ast.Await):
return _is_redis_expr(expr.value, scope)
if isinstance(expr, ast.Subscript):
return _is_redis_expr(expr.value, scope)
path = _chain_path(expr)
if path is not None:
if path[0] in REDIS_ROOT_NAMES or path[0] in scope.redis_aliases:
return True
for i in range(2, len(path) + 1):
if ".".join(path[:i]) in scope.redis_qualified_aliases:
return True
return isinstance(expr, ast.Attribute) and expr.attr in REDIS_ATTR_LEAVES
if isinstance(expr, ast.Call):
func = expr.func
if isinstance(func, ast.Attribute) and func.attr == "pipeline":
return _is_redis_expr(func.value, scope)
args = (*expr.args, *(kw.value for kw in expr.keywords))
return any(_is_redis_expr(a, scope) for a in args)
return False
def _is_testid_derived(expr: ast.expr, scope: _Scope) -> bool:
"""True if *expr* — a Redis KEY expression — actually derives from
``test_id``, resolved through *scope*: directly, through a local
variable/attribute bound to something test_id-derived, through an
f-string / ``%``-format / ``+``-concatenation / ``.format(...)`` call
that embeds one, or (for a PARAMETER, at a specific call site) through
the interprocedural seeding in ``_resolve_module_calls``. The mere
PRESENCE of the token ``test_id`` somewhere in the enclosing function
does not derive any particular key — that presence check was the
shipped defect this replaces. Deliberately narrow for an ARBITRARY
function call used AS a key expression (``_build_key(test_id)``): only
the known-safe ``str.format(...)`` shape is trusted to embed its
arguments in its result — trusting ANY call that merely RECEIVES
something test_id-derived (with no proof the callee's return value
actually uses it) was itself a shipped defect: a helper that receives
``test_id`` and returns a hardcoded key would exempt the write."""
if isinstance(expr, ast.Await):
return _is_testid_derived(expr.value, scope)
if isinstance(expr, ast.Name):
return expr.id == "test_id" or expr.id in scope.testid_aliases
if isinstance(expr, ast.Attribute):
path = _chain_path(expr)
if path is not None:
if path[0] == "test_id" or path[0] in scope.testid_aliases:
return True
return any(
".".join(path[:i]) in scope.testid_qualified_aliases
for i in range(2, len(path) + 1)
)
return _is_testid_derived(expr.value, scope)
if isinstance(expr, ast.Subscript):
return _is_testid_derived(expr.value, scope)
if isinstance(expr, ast.JoinedStr):
return any(
isinstance(v, ast.FormattedValue) and _is_testid_derived(v.value, scope)
for v in expr.values
)
if isinstance(expr, ast.BinOp):
return _is_testid_derived(expr.left, scope) or _is_testid_derived(
expr.right, scope
)
if isinstance(expr, (ast.Tuple, ast.List)):
return any(_is_testid_derived(e, scope) for e in expr.elts)
if isinstance(expr, ast.Call):
func = expr.func
if isinstance(func, ast.Attribute) and func.attr == "format":
args = (*expr.args, *(kw.value for kw in expr.keywords))
return any(_is_testid_derived(a, scope) for a in args) or (
_is_testid_derived(func.value, scope)
)
return False
return False
def _seed_params_from_arguments(args: ast.arguments) -> tuple[set[str], set[str]]:
"""(redis-seeded param names, test_id-seeded param names) from a bare
``ast.arguments`` node alone — ``test_id`` itself, and any parameter
literally named like a known Redis fixture. Shared by ``_seed_params``
(a ``def``'s signature) and lambda-scope seeding (a ``lambda``'s
signature is the SAME node type)."""
names = {a.arg for a in (*args.posonlyargs, *args.args, *args.kwonlyargs)}
if args.vararg is not None:
names.add(args.vararg.arg)
if args.kwarg is not None:
names.add(args.kwarg.arg)
return (names & REDIS_ROOT_NAMES, {"test_id"} & names)
def _seed_params(
func: ast.FunctionDef | ast.AsyncFunctionDef,
) -> tuple[set[str], set[str]]:
"""(redis-seeded params, test_id-seeded params) from *func*'s own
signature alone — ``test_id`` itself, and any parameter literally named
like a known Redis fixture."""
return _seed_params_from_arguments(func.args)
def _expr_calls(node: ast.AST) -> list[ast.Call]:
"""Call nodes reachable from *node* itself and its descendants, without
crossing into a NESTED statement (those are walked separately, in
physical order, by ``_process_scope``'s recursion) or a nested
FunctionDef/AsyncFunctionDef/Lambda (those get their own scope — a
Lambda's body is walked separately by ``_find_lambdas`` + its own
per-lambda scope, never folded into the enclosing statement's). *node*
itself is checked too — not just its children — so this also serves a
Lambda's single-expression BODY directly (``lambda: redis_client.set(...)``
has a bare ``Call`` as its whole body, with no enclosing statement of its
own to recurse from)."""
calls: list[ast.Call] = []
def _collect(n: ast.AST) -> None:
if isinstance(n, ast.Call):
calls.append(n)
for child in ast.iter_child_nodes(n):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda)):
continue
if isinstance(child, ast.stmt):
continue
_collect(child)
_collect(node)
return calls
def _stmt_own_calls(stmt: ast.stmt) -> list[ast.Call]:
"""Call nodes that live directly in *stmt*'s own expression fields (its
condition, its RHS, its decorators-free body-less parts) — NOT inside any
NESTED statement (those are walked separately, in physical order, by
``_process_scope``'s recursion) and NOT inside a nested
FunctionDef/AsyncFunctionDef/Lambda (those get their own scope)."""
return _expr_calls(stmt)
def _find_lambdas(stmt: ast.stmt) -> list[ast.Lambda]:
"""Every ``Lambda`` reachable from *stmt*'s own expression fields —
including a lambda nested inside another lambda — without crossing into
a nested FunctionDef/AsyncFunctionDef (those get their own, separate
statement-level walk when ``_process_scope`` recurses into them).
Companion to ``_stmt_own_calls``: a Lambda is invisible to that walk (it
stops at the boundary, same as a ``def``) so a write inside a lambda body
— e.g. ``pytest.fixture(lambda: redis_client.flushdb())`` — would
otherwise never be visited at all, not merely unseeded."""
found: list[ast.Lambda] = []
def _collect(n: ast.AST) -> None:
for child in ast.iter_child_nodes(n):
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if isinstance(child, ast.stmt):
continue
if isinstance(child, ast.Lambda):
found.append(child)
_collect(child)
_collect(stmt)
return found
# Compound statements whose nested statement-list fields get walked in the
# SAME (flat) scope as their parent — an `if`/`for`/`while`/`with` block does
# not open a new alias scope, only a `def` does. ``ast.Try`` is handled
# separately (it has more than two list fields).
_COMPOUND_BODY_FIELDS: dict[type, tuple[str, ...]] = {
ast.If: ("body", "orelse"),
ast.For: ("body", "orelse"),
ast.AsyncFor: ("body", "orelse"),
ast.While: ("body", "orelse"),
ast.With: ("body",),
ast.AsyncWith: ("body",),
}
def _bind_alias_target(target: ast.expr, value: ast.expr, scope: _Scope) -> None:
"""Grow *scope* from one ``target = value``-shaped binding — a bare
``Name`` target becomes a plain alias, a dotted ``Attribute`` target
becomes a qualified alias (``holder.conn = redis_client``). Shared by
``Assign`` and ``With``/``AsyncWith`` (``with EXPR as TARGET:``) — the
same binding shape, just spelled differently."""
if isinstance(target, ast.Name):
if _is_redis_expr(value, scope):
scope.redis_aliases.add(target.id)
if _is_testid_derived(value, scope):
scope.testid_aliases.add(target.id)
return
path = _chain_path(target)
if path is not None and len(path) >= 2:
key = ".".join(path)
if _is_redis_expr(value, scope):
scope.redis_qualified_aliases.add(key)
if _is_testid_derived(value, scope):
scope.testid_qualified_aliases.add(key)
def _process_scope(
body: list[ast.stmt],
scope: _Scope,
calls_out: list[tuple[ast.Call, _Scope]],
scopes_by_node: dict[ast.AST, _Scope],
) -> None:
"""Walk *body*'s statements in physical order, growing *scope*'s alias
tables from assignments and recording every call site — SNAPSHOTTED at
the moment it is seen, never a live reference to the still-mutating
*scope* (flow sensitivity: a rebinding later in the function must not
retroactively exempt an earlier write) — into *calls_out*. A nested
FunctionDef/AsyncFunctionDef gets its OWN child scope (a copy of *scope*
at this point, per real closure semantics) and is recursed into
separately; a Lambda gets the same treatment via ``_find_lambdas`` (its
single-expression body cannot itself contain a nested statement, so no
further recursion is needed); a nested ClassDef is walked flat (to
reach its methods) without seeding ``self``; ``with EXPR as TARGET:``
binds TARGET the same way an ``Assign`` would; ``match``/``case`` bodies
are walked in the same (flat) scope as their ``match`` statement.
"""
for stmt in body:
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
child = _Scope(parent=scope)
redis_p, testid_p = _seed_params(stmt)
child.redis_aliases |= redis_p
child.testid_aliases |= testid_p
scopes_by_node[stmt] = child
_process_scope(stmt.body, child, calls_out, scopes_by_node)
continue
if isinstance(stmt, ast.ClassDef):
_process_scope(stmt.body, scope, calls_out, scopes_by_node)
continue
for call in _stmt_own_calls(stmt):
calls_out.append((call, scope.snapshot()))
for lam in _find_lambdas(stmt):
lam_scope = _Scope(parent=scope)
redis_p, testid_p = _seed_params_from_arguments(lam.args)
lam_scope.redis_aliases |= redis_p
lam_scope.testid_aliases |= testid_p
for call in _expr_calls(lam.body):
calls_out.append((call, lam_scope))
if isinstance(stmt, ast.Assign):
for target in stmt.targets:
_bind_alias_target(target, stmt.value, scope)
if isinstance(stmt, (ast.With, ast.AsyncWith)):
for item in stmt.items:
if item.optional_vars is not None:
_bind_alias_target(item.optional_vars, item.context_expr, scope)
if isinstance(stmt, ast.Try):
_process_scope(stmt.body, scope, calls_out, scopes_by_node)
for handler in stmt.handlers:
_process_scope(handler.body, scope, calls_out, scopes_by_node)
_process_scope(stmt.orelse, scope, calls_out, scopes_by_node)
_process_scope(stmt.finalbody, scope, calls_out, scopes_by_node)
continue
if isinstance(stmt, ast.Match):
for case in stmt.cases:
_process_scope(case.body, scope, calls_out, scopes_by_node)
continue
for field in _COMPOUND_BODY_FIELDS.get(type(stmt), ()):
_process_scope(getattr(stmt, field), scope, calls_out, scopes_by_node)
def _seeds_from_call(
call: ast.Call,
caller_scope: _Scope,
callee: ast.FunctionDef | ast.AsyncFunctionDef,
) -> tuple[frozenset[str], frozenset[str]]:
"""The (redis-seeded, test_id-seeded) parameter names *this ONE call
site* supplies to *callee*, resolved purely against *caller_scope*.
Deliberately never merged with any OTHER call site's seeds — that merge
was the unsoundness the 2026-08-12 rewrite removed: one caller passing a
test_id-derived key would retroactively exempt every OTHER caller of the
same shared helper, including one passing a hardcoded key. Call-site
isolation is enforced by the caller, ``_resolve_module_calls``: each
call site drives its OWN, independent re-walk of the callee's body."""
positional_params = [*callee.args.posonlyargs, *callee.args.args]
kw_eligible = {a.arg for a in (*callee.args.args, *callee.args.kwonlyargs)}
redis_seeds: set[str] = set()
testid_seeds: set[str] = set()
def _seed(pname: str, arg_expr: ast.expr) -> None:
if _is_redis_expr(arg_expr, caller_scope):
redis_seeds.add(pname)
if _is_testid_derived(arg_expr, caller_scope):
testid_seeds.add(pname)
for i, arg_expr in enumerate(call.args):
if i >= len(positional_params):
break
_seed(positional_params[i].arg, arg_expr)
for kw in call.keywords:
if kw.arg is not None and kw.arg in kw_eligible:
_seed(kw.arg, kw.value)
return frozenset(redis_seeds), frozenset(testid_seeds)
def _resolve_module_calls(tree: ast.Module) -> list[tuple[ast.Call, _Scope]]:
"""Every Call in *tree*, paired with the scope active at its call site —
resolved through a CALL-SITE-ISOLATED interprocedural analysis
(2026-08-12 rewrite, replacing a per-parameter UNION across all of a
helper's call sites — the defect where one caller's test_id-derived
argument retroactively exempted every other caller of the same helper).
A top-level function with NO same-module caller (nothing else in the
file names it in a call) is a ROOT — its only real invocation is
external (pytest calling a ``test_*`` function or fixture directly), or
it is dead code either way — and is analyzed exactly once, seeded from
nothing but its own signature (``_seed_params``).
A top-level function WITH one or more same-module callers is instead
analyzed ONCE PER DISTINCT CALL SITE: each call site's own resolved
(redis-seeded, test_id-seeded) parameters (``_seeds_from_call``) drive
an INDEPENDENT re-walk of the callee's body, memoized by (function
name, seed set) to skip an identical context reached twice and to bound
recursion on (mutually) recursive helpers. A write inside the callee is
therefore judged against the SPECIFIC calling context that reaches it —
never against a different, unrelated call site's seeds — while a chain
of delegations (``A`` calls ``B`` calls ``C``) is still resolved, one
hop at a time, by the same recursion, up to ``_TAINT_FIXED_POINT_ROUNDS``
hops deep. This is what lets ``_seed(redis_client)`` taint ``_seed``'s
``client`` parameter, and ``_do_write(redis_client, test_id)`` taint
BOTH of ``_do_write``'s parameters — while a DIFFERENT call site of the
SAME ``_do_write`` that passes a hardcoded key stays flagged, exactly as
it should.
A called (non-root) function's OWN baseline walk (no external seeds) is
used only to discover the call-graph edges into it — it is provisional
structure, discarded once real callers are known, and never itself
contributes findings (a caller-supplied argument the callee's own
signature alone cannot see would otherwise be invisible to that
baseline walk, producing a false positive for a genuinely-scoped write).
"""
module_functions: dict[str, ast.FunctionDef | ast.AsyncFunctionDef] = {
stmt.name: stmt
for stmt in tree.body
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef))
}
# Calls made OUTSIDE any top-level function's own body — at module top
# level, or inside a class body (pytest test-class methods; a method is
# never itself an entry of `module_functions`, so it is not subject to
# the root/called distinction below — it is always analyzed, exactly
# like a root, from its own signature alone). Always final.
root_scope = _Scope()
outer_calls: list[tuple[ast.Call, _Scope]] = []
for stmt in tree.body:
if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
if isinstance(stmt, ast.ClassDef):
_process_scope(stmt.body, root_scope, outer_calls, {})
else:
for call in _stmt_own_calls(stmt):
outer_calls.append((call, root_scope.snapshot()))
for lam in _find_lambdas(stmt):
lam_scope = _Scope(parent=root_scope)
redis_p, testid_p = _seed_params_from_arguments(lam.args)
lam_scope.redis_aliases |= redis_p
lam_scope.testid_aliases |= testid_p
for call in _expr_calls(lam.body):
outer_calls.append((call, lam_scope))
# Every top-level function's OWN baseline walk, kept SEPARATE per
# function so a function later found to have a same-module caller can
# have its baseline verdicts discarded wholesale — never merged with
# anything else, and never merged with each other.
per_function_baseline: dict[str, list[tuple[ast.Call, _Scope]]] = {}
for name, func in module_functions.items():
scope = _Scope()
own_redis, own_testid = _seed_params(func)
scope.redis_aliases |= own_redis
scope.testid_aliases |= own_testid
local_calls: list[tuple[ast.Call, _Scope]] = []
_process_scope(func.body, scope, local_calls, {})
per_function_baseline[name] = local_calls
called_names: set[str] = set()
for local_calls in per_function_baseline.values():
for call, _scope in local_calls:
if isinstance(call.func, ast.Name) and call.func.id in module_functions:
called_names.add(call.func.id)
for call, _scope in outer_calls:
if isinstance(call.func, ast.Name) and call.func.id in module_functions:
called_names.add(call.func.id)
all_calls: list[tuple[ast.Call, _Scope]] = list(outer_calls)
for name, local_calls in per_function_baseline.items():
if name not in called_names:
all_calls.extend(local_calls) # a genuine root — baseline is final
visited: set[tuple[str, frozenset[str], frozenset[str]]] = set()
def _analyze(
func: ast.FunctionDef | ast.AsyncFunctionDef,
extra_redis: frozenset[str],
extra_testid: frozenset[str],
depth: int,
) -> None:
key = (func.name, extra_redis, extra_testid)
if key in visited or depth > _TAINT_FIXED_POINT_ROUNDS:
return
visited.add(key)
own_redis, own_testid = _seed_params(func)
scope = _Scope()
scope.redis_aliases |= own_redis | extra_redis
scope.testid_aliases |= own_testid | extra_testid
local_calls: list[tuple[ast.Call, _Scope]] = []
_process_scope(func.body, scope, local_calls, {})
all_calls.extend(local_calls)
_recurse(local_calls, depth)
def _recurse(calls: list[tuple[ast.Call, _Scope]], depth: int) -> None:
for call, caller_scope in calls:
target = call.func
if not isinstance(target, ast.Name):
continue
callee = module_functions.get(target.id)
if callee is None:
continue
redis_seeds, testid_seeds = _seeds_from_call(call, caller_scope, callee)
_analyze(callee, redis_seeds, testid_seeds, depth + 1)
# Every call site into a CALLED (non-root) function spawns its own,
# independent re-analysis of the callee — found either at module top
# level / in a class body (`outer_calls`, already final), or inside a
# ROOT function's own baseline body (also final: a root's baseline
# calling context is real, since nothing else in the module calls it).
# A NON-root function's own baseline call sites are intentionally
# excluded here — its baseline was already discarded above as
# non-final, and recursing from it too would reintroduce the exact
# union-across-callers defect this rewrite fixes.
_recurse(outer_calls, depth=0)
for name, local_calls in per_function_baseline.items():
if name not in called_names:
_recurse(local_calls, depth=0)
return all_calls
def _gate_04_line_based_scan(lines: list[str]) -> list[tuple[int, str]]:
"""The ORIGINAL (pre-AST) Gate 0.4 detector, MINUS the TEST_ID_NEARBY
text-window (deleted everywhere in this gate — see the block comment
above the Gate 0.4 constants) — the fallback for a ``test_*.py`` file the
AST parser cannot read. A file the gate cannot parse must not become an
invisible hole in the audit, so it still gets the weaker,
string-literal-blind line scan rather than being skipped; it can no
longer be talked out of a finding by an unrelated token sitting nearby —
only by the same narrow same-line ``def test_...test_id`` signature check
the original rule always used, or a reasoned ``# gate-ok:`` waiver (read
through the shared ``core.gate_waivers`` module, same as the AST path)."""
found: list[tuple[int, str]] = []
waived = waived_lines("\n".join(lines))
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("#") or stripped.startswith('"""'):
continue
if not _REDIS_LINE_SCAN_RE.search(line):
continue
if (i + 1) in waived:
continue
func_has_test_id = False
# ``max(-1, ...)`` (not ``max(0, ...)``) so the walk's own stop bound
# is exclusive-of-index-0 correctly: a range whose stop is 0 never
# visits index 0 at all, making a ``def test_...test_id`` sitting on
# a fixture's literal FIRST line invisible to this check (found while
# rewriting this fallback: same defect class as the one this whole
# gate exists to catch — a boundary excused from the very check
# meant to examine it).
for j in range(i, max(-1, i - 40), -1):
if "def test_" in lines[j] and "test_id" in lines[j]:
func_has_test_id = True
break
if not func_has_test_id:
found.append((i + 1, line))
return found
def _write_target(call: ast.Call) -> tuple[str, ast.expr, int] | None:
"""(verb, receiver-expr, physical write-line) for a REDIS-shaped write
call, covering two receiver shapes: the ordinary ``<expr>.<verb>(...)``
static attribute access, and ``getattr(<expr>, "<verb>")(...)`` — the
exact same mutation reached through a string literal instead of a
static attribute (undetected before this rewrite: ``call.func`` there is
itself a ``Call``, not an ``Attribute``). Returns ``None`` for any other
call shape. For the attribute shape, the write line is the ``.verb``
Attribute's ``end_lineno`` — not the receiver's line, since for
``(\\n redis_client\\n .set(...)\\n)`` the ``Call`` node's own
``lineno`` is the ``redis_client`` line (Python's AST anchors a Call at
its callee's start). For the getattr shape there is no Attribute node to
anchor on, so the outer call's own ``end_lineno`` — the closing paren of
``(...)`` — is used instead."""
func = call.func
if isinstance(func, ast.Attribute):
return func.attr, func.value, (func.end_lineno or call.lineno)
if (
isinstance(func, ast.Call)
and isinstance(func.func, ast.Name)
and func.func.id == "getattr"
and len(func.args) >= 2
and isinstance(func.args[1], ast.Constant)
and isinstance(func.args[1].value, str)
):
return func.args[1].value, func.args[0], (call.end_lineno or call.lineno)
return None
def _key_exprs(call: ast.Call, verb: str) -> list[ast.expr]:
"""The expression(s) that determine the Redis KEY(S) *call* touches,
given its already-resolved write *verb* (``_write_target``'s first
element — never re-derived from ``call.func`` here, since that node is
NOT always an ``Attribute``: the ``getattr(...)`` write shape has no
``.attr`` to read). ``execute_command("SET", key, value)`` carries the
command name as ``args[0]``, so the key is ``args[1]``; every other
verb's key is the first positional argument, or — no positional args at
all, e.g. ``hset(name=..., mapping=...)`` / ``mset(mapping=...)`` — its
``name``/``mapping`` keyword, falling back to every keyword value so
nothing with a derivable key is silently treated as bare."""
if verb == "execute_command":
return [call.args[1]] if len(call.args) > 1 else list(call.args)
if call.args:
return [call.args[0]]
named = [kw.value for kw in call.keywords if kw.arg in ("name", "mapping")]
return named or [kw.value for kw in call.keywords]
def gate_04_redis_keys():
"""Redis write operations must namespace keys with test_id.
AST data-flow (see the block comment above the Gate 0.4 constants for
the full design, and why the prior TEST_ID_NEARBY text-window and
"test_id anywhere in the enclosing function" presence check are gone). A
write is a violation unless: (a) a reasoned ``# gate-ok: <reason>``
waiver — read through the shared ``core.gate_waivers`` module, never a
local regex copy — covers the write's own line; or (b) the KEY
expression provably derives from ``test_id`` through the alias /
interprocedural machinery in ``_resolve_module_calls``. Detection itself
is AST-only: only a genuine ``ast.Call`` whose receiver resolves to a
Redis object (``_is_redis_expr``) and whose verb mutates keyspace state
(``REDIS_WRITE_VERBS``) is ever a candidate — text that merely *looks*
like a write inside a string literal, f-string prose, or a comment never
parses to a Call at all. A file the parser cannot read (SyntaxError, or
a null-byte ``ValueError``) falls back to the original line-based scan
for that one file, and the violation detail says so, rather than the
file being silently skipped.
The SAME physical write line can be visited more than once — a write
inside a shared helper is now re-analyzed once per distinct calling
context (see ``_resolve_module_calls``), so it is deduplicated by
(file, line): flagged if UNSAFE under any of those contexts, silent
only if EVERY one of them proves the key test_id-derived.
"""
violations = []
for f in sorted(TESTS_DIR.rglob("test_*.py")):
try:
content = f.read_text()
except Exception:
continue
lines = content.split("\n")
try:
tree = ast.parse(content)
except SyntaxError, ValueError:
for lineno, line in _gate_04_line_based_scan(lines):
violations.append(
{
"file": str(f.relative_to(REPO_ROOT)),
"line": lineno,
"detail": (
"Redis write without test_id (unparseable file, "
f"line-based fallback): {line.strip()[:100]}"
),
}
)
continue
waived = waived_lines(content)
seen_lines: set[int] = set()
for call, scope in _resolve_module_calls(tree):
target = _write_target(call)
if target is None:
continue
verb, receiver, write_line = target
if verb not in REDIS_WRITE_VERBS:
continue
if not _is_redis_expr(receiver, scope):
continue
if write_line in waived or write_line in seen_lines:
continue
if any(_is_testid_derived(k, scope) for k in _key_exprs(call, verb)):
continue
seen_lines.add(write_line)
line_text = (
lines[write_line - 1] if 0 <= write_line - 1 < len(lines) else ""
)
violations.append(
{
"file": str(f.relative_to(REPO_ROOT)),
"line": write_line,
"detail": f"Redis write without test_id: {line_text.strip()[:100]}",
}
)
return violations
# ── Gate 0.5: Filesystem writes outside isolation ───────────────────────────
FILE_WRITE_RE = re.compile(
r"(?:\.write_text\s*\(|open\s*\([^)]*[\"']w|Path\s*\([^)]*\)\.write|"
r"shutil\.(?:copy|move)|\.mkdir\s*\()",
)
# tmp_path itself, its conventional wrapper-fixture names, the tempfile API, and
# write targets named like tmp artifacts (``*_path``/``*_dir``/``*_file``/etc.),
# which in test code are overwhelmingly tmp_path-derived.
ISOLATION_RE = re.compile(
r"isolation|tmp_path|tmp_path_factory|tmpdir|tempfile|temp_dir|"
r"temp_project|temp_root|temp_repo|project_root|project_dir|"
r"git_project|git_repo|\bmkdtemp\b|tmp_|_tmpdir|"
r"\w+_(?:path|dir|file|root|repo)\b",
re.IGNORECASE,
)
# A function/fixture/helper that receives the write target as a Path parameter
# (the pytest tmp_path convention, possibly via a wrapper fixture) writes under a
# caller-controlled dir -> isolated.
_PATH_PARAM_RE = re.compile(
r"\b(?:tmp_path|tmp_path_factory|tmpdir|isolation|root|[a-z_]*_root|"
r"path|[a-z_]*_path|[a-z_]*_dir|project|[a-z_]*_project|repo|[a-z_]*_repo|temp_\w*)\s*[:,)=]",
re.IGNORECASE,
)
def _enclosing_region(lines: list[str], i: int) -> tuple[str, str] | None:
"""(signature, body) of the function enclosing line ``i``, or None."""
start = None
for j in range(i, -1, -1):
if re.match(r"\s*(?:async\s+)?def\s", lines[j]):
start = j
break
if start is None:
return None
# Signature (may span several lines until the closing ')').
sig: list[str] = []
for j in range(start, min(start + 10, len(lines))):
sig.append(lines[j])
if re.search(r"\)\s*(?:->|:)", lines[j]):
break
indent = len(lines[start]) - len(lines[start].lstrip())
end = len(lines)
for j in range(start + 1, len(lines)):
s = lines[j]
if (
s.strip()
and (len(s) - len(s.lstrip())) <= indent
and re.match(r"\s*(?:def|class|@)", s)
):
end = j
break
return "\n".join(sig), "\n".join(lines[start:end])
def _enclosing_uses_tmp(lines: list[str], i: int) -> bool:
"""True if the function enclosing line ``i`` is tmp-scoped: it takes a Path
parameter (tmp_path or a wrapper like ``root``/``temp_project``, or any
parameter explicitly annotated ``: Path`` — a caller-controlled directory
by contract) or references tmp_path/tempfile anywhere in its body."""
region = _enclosing_region(lines, i)
if region is None:
return False
sig, body = region
m = re.search(r"def\s+\w+\s*\((.*?)\)\s*(?:->|:)", sig, re.DOTALL)
if m and (
_PATH_PARAM_RE.search(m.group(1))
or re.search(r":\s*(?:pathlib\.)?Path\b", m.group(1))
):
return True
return bool(ISOLATION_RE.search(body))
def _build_fixture_tmp_index() -> re.Pattern | None:
"""Compile a word-matcher for every fixture whose value is tmp-backed.
Resolves wrapper-fixture CHAINS transitively (``git_controller(git_repo)``
-> ``git_repo(temp_root)`` -> ``temp_root(tmp_path)``): a fixture is
tmp-backed if its body references tmp_path/tempfile, its signature carries
a tmp-style/`: Path` parameter, or any of its parameters is itself a
tmp-backed fixture. Same-named fixtures across files are merged — for an
advisory suppression index that bias is the safe direction. Returns None
when no fixture qualifies (so callers can skip the check)."""
defs: dict[str, tuple[set[str], str]] = {}
for f in list(TESTS_DIR.rglob("test_*.py")) + list(TESTS_DIR.rglob("conftest.py")):
try:
lines = f.read_text().split("\n")
except Exception:
continue
for i, line in enumerate(lines):
if not re.match(r"@(?:pytest\.)?fixture\b", line.strip()):
continue
j = i + 1
while j < len(lines) and not re.match(r"\s*(?:async\s+)?def\s", lines[j]):
j += 1
if j >= len(lines):
continue
region = _enclosing_region(lines, j)
if region is None:
continue
sig, body = region
m = re.search(r"def\s+(\w+)\s*\((.*?)\)\s*(?:->|:)", sig, re.DOTALL)
if not m:
continue
name = m.group(1)
params = {
p.split(":")[0].split("=")[0].strip()
for p in m.group(2).split(",")
if p.strip() and not p.strip().startswith("*")
}
prev = defs.get(name)
defs[name] = (
params | (prev[0] if prev else set()),
((prev[1] + "\n") if prev else "") + sig + "\n" + body,
)
tmp_backed: set[str] = set()
changed = True
rounds = 0
while changed and rounds < 8: # bounded transitive closure
changed = False
rounds += 1
for name, (params, text) in defs.items():
if name in tmp_backed:
continue
if (
ISOLATION_RE.search(text)
or _PATH_PARAM_RE.search("," + ",".join(params) + ",")
or params & tmp_backed
):
tmp_backed.add(name)
changed = True
if not tmp_backed:
return None
return re.compile(r"\b(?:" + "|".join(map(re.escape, sorted(tmp_backed))) + r")\b")
def _pure_string_content_lines(source: str) -> set[int]:
"""1-based line numbers whose ENTIRE content is string-literal text.
Gate 0.5 needs a stricter exclusion than :func:`_docstring_line_numbers`
and a narrower one than "any line holding a literal".
Measured 2026-08-08, both directions in one sitting:
* TOO NARROW — the gate warned on ``"p.write_text(p.read_text())\\n"``,
one line of a multi-line payload that tests/scripts/
test_single_gate_entrypoint.py feeds to a hook on stdin. That string
sits in an ASSIGNMENT, so it is not a bare string-expression statement
and _docstring_line_numbers never covered it -- yet nothing there
touches the filesystem.
* TOO WIDE — excluding every line that CONTAINS a literal blinded the
gate completely, because a real write is
``open("/etc/x.conf", "w")``: the path and the mode ARE literals. The
first attempt at this helper did exactly that and three existing tests
went red.
So the rule is positional: blank out the literal spans and keep the line
only if something executable survives. A line that is nothing but string
content cannot perform a write; a line with code around its literals can.
Gates 0.1/0.2 deliberately do NOT use this -- for them the string IS the
payload (a Cypher query passed to ``.run("...")`` must stay scannable).
"""
lines = source.split("\n")
grid = [list(line) for line in lines]
blank = {tokenize.STRING}
for name in ("FSTRING_START", "FSTRING_MIDDLE", "FSTRING_END"):
if hasattr(tokenize, name):
blank.add(getattr(tokenize, name))
try:
for tok in tokenize.generate_tokens(io.StringIO(source).readline):
if tok.type not in blank:
continue
(sr, sc), (er, ec) = tok.start, tok.end
for r in range(sr, er + 1):
if not (1 <= r <= len(grid)):
continue
row = grid[r - 1]
c0 = sc if r == sr else 0
c1 = ec if r == er else len(row)
for c in range(c0, min(c1, len(row))):
row[c] = " "
except Exception:
# Un-tokenizable source: exclude nothing, so the gate stays strict.
return set()
# What survives blanking decides. Whitespace and pure STRUCTURE --
# separators, brackets, the implicit/explicit concatenation operator, a
# line continuation -- cannot open, write to, or unlink anything. A write
# needs a NAME (open, Path, shutil, .write_text), and a name survives.
#
# Structure has to count as inert or the rule misses the common shape: a
# string element of a list is `"...",` and that trailing comma alone was
# enough to flag two lines of this gate\'s own test corpus.
inert = set(" \t,()[]{}+\\")
pure: set[int] = set()
for i, row in enumerate(grid, 1):
if not lines[i - 1].strip():
continue
if not set("".join(row)) - inert:
pure.add(i)
return pure
def gate_05_filesystem():
"""Filesystem writes must use the isolation fixture, tmp_path, or a tmp-derived
path. A write is isolated if the nearby context OR the enclosing function body
references tmp_path/tempfile (so wrapper fixtures like ``temp_project`` and
``project_root`` are recognised)."""
violations = []
tmp_fixture_re = _build_fixture_tmp_index()
for f in sorted(TESTS_DIR.rglob("test_*.py")):
try:
content = f.read_text()
except Exception:
continue
lines = content.split("\n")
literal_lines = _pure_string_content_lines(content)
for i, line in enumerate(lines):
stripped = line.strip()
if stripped.startswith("#") or (i + 1) in literal_lines:
continue
if _is_waived(lines, i + 1):
continue
if FILE_WRITE_RE.search(line):
ctx_start = max(0, i - 10)
ctx_end = min(len(lines), i + 3)
context = "\n".join(lines[ctx_start:ctx_end])
if ISOLATION_RE.search(context) or _enclosing_uses_tmp(lines, i):
continue
# Wrapper-fixture resolution: the write target came through a
# fixture whose value is PROVEN tmp-backed (transitive chain
# to tmp_path/tempfile) — referenced in the enclosing function
# or the nearby context.
if tmp_fixture_re is not None:
region = _enclosing_region(lines, i)
scope = context + ("\n" + region[1] if region else "")
if tmp_fixture_re.search(scope):
continue
# Instance-attribute path (``self.workspace``): constructor-
# injected by the test that built the object — the
# construction sites carry their own tmp_path context.
if re.search(
r"(?:os\.path\.join\(\s*self\.|\bself\.\w+\s*/|open\(\s*self\.)",
context,
):
continue
violations.append(
{
"file": str(f.relative_to(REPO_ROOT)),
"line": i + 1,
"detail": f"Filesystem write outside isolation: {line.strip()[:100]}",
}
)
return violations
# ── CLI ─────────────────────────────────────────────────────────────────────
# ── Gate 0.6: wait-for-port guards (a shared port wearing a disguise) ────────
#
# A loop that polls a port until it looks free is only ever written because a
# test is sharing a fixed port with the other seven workers, and it cannot make
# that safe: `connect()` asks "is anything accepting?" while the server asks
# "may I own this address?". A listener on a specific address (127.0.0.2, a host
# IP) or a socket in TIME_WAIT answers no to both, so the guard passes and the
# bind still fails.
#
# tests/knowledge_graph/test_server.py collided on 0.0.0.0:50052 (2026-07-10),
# gained exactly this guard, then collided again on 0.0.0.0:50154 (2026-08-13)
# and cost an 8.8-minute authoritative run. Rule T5 already forbade the shared
# port; nothing enforced it. This gate does.
#
# Fix: Ephemeral Port Injection (docs/ephemeral-port-injection.md) — allocate
# with bind(port 0) and inject the port through the entrypoint's `env` mapping.
def _waits_for_port_to_be_free(node: ast.AST) -> bool:
"""True for ``connect_ex(...) != 0`` — "nothing is accepting, so it is mine".
The DIRECTION of the comparison is the whole distinction, and it separates
two loops that otherwise look identical:
* ``connect_ex(...) == 0`` waits for a server to come UP. Legitimate: the
test started that server and is waiting to talk to it.
* ``connect_ex(...) != 0`` waits for a port to fall FREE, which is the
shared-port guard this gate exists to refuse.
Only the direct comparison form is recognised. A loop that assigns the
result to a name first is not matched — a narrower rule that never cries
wolf beats a broad one that gets waived into irrelevance.
"""
for sub in ast.walk(node):
if not isinstance(sub, ast.Compare):
continue
if not (
isinstance(sub.left, ast.Call)
and isinstance(sub.left.func, ast.Attribute)
and sub.left.func.attr == "connect_ex"
):
continue
for op, comparator in zip(sub.ops, sub.comparators):
if isinstance(op, ast.NotEq) and (
isinstance(comparator, ast.Constant) and comparator.value == 0
):
return True
return False
def _wait_for_port_loops(source: str, filename: str) -> list[dict]:
"""Loops that poll a port for availability, rather than taking one."""
try:
tree = ast.parse(source)
except SyntaxError:
return []
lines = source.splitlines()
found: list[dict] = []
for node in ast.walk(tree):
if not isinstance(node, (ast.While, ast.For)):
continue
if not _waits_for_port_to_be_free(node):
continue
if _is_waived(lines, node.lineno):
continue
found.append(
{
"file": filename,
"line": node.lineno,
"detail": (
"polls a port until it looks free — a shared fixed port. "
"connect_ex cannot see a listener on a specific address or "
"a socket in TIME_WAIT, so this guard passes and the bind "
"still fails. Use Ephemeral Port Injection: bind(port 0) "
"plus an explicit env mapping "
"(docs/ephemeral-port-injection.md)"
),
}
)
return found
def gate_06_wait_for_port():
"""A test must take a free port, not wait for a shared one to look free."""
violations = []
for f in sorted(TESTS_DIR.rglob("test_*.py")):
try:
source = f.read_text()
except Exception:
continue
violations.extend(_wait_for_port_loops(source, str(f.relative_to(REPO_ROOT))))
return violations
GATES = {
"0.1": ("Cypher CREATE without test_id", gate_01_cypher_create),
"0.2": ("Cypher DETACH DELETE without test_id", gate_02_detach_delete),
"0.3": ("Session-scope fixture audit", gate_03_session_fixtures),
"0.4": ("Redis key namespacing", gate_04_redis_keys),
"0.5": ("Filesystem isolation", gate_05_filesystem),
"0.6": ("Wait-for-port guard (shared port)", gate_06_wait_for_port),
}
# Gate 0.5 (filesystem) is ADVISORY: static analysis cannot prove a write target
# is tmp-backed when it comes through a wrapper fixture, so it is reported but does
# not fail the build. The graph/Redis/fixture gates are precise and BLOCKING.
ADVISORY_GATES = {"0.5"}
def main() -> int:
parser = argparse.ArgumentParser(
description="Parallel safety static analysis gates"
)
parser.add_argument("--gate", choices=list(GATES), help="Run a single gate")
parser.add_argument("--json", action="store_true", help="JSON output")
parser.add_argument(
"--strict", action="store_true", help="Fail on advisory gates too"
)
parser.add_argument(
"--tests-dir",
type=Path,
default=None,
help="Scan this tests tree instead of the repo's (deterministic "
"CLI-behaviour tests seed a sandbox tree with known violations)",
)
args = parser.parse_args()
if args.tests_dir is not None:
global TESTS_DIR, REPO_ROOT
TESTS_DIR = args.tests_dir.resolve()
REPO_ROOT = TESTS_DIR.parent
gates_to_run = [args.gate] if args.gate else list(GATES)
all_violations = {}
for gate_id in gates_to_run:
name, fn = GATES[gate_id]
violations = fn()
all_violations[gate_id] = {"name": name, "violations": violations}
if args.json:
print(json.dumps(all_violations, indent=2))
else:
blocking_total = advisory_total = 0
for gate_id in gates_to_run:
info = all_violations[gate_id]
n = len(info["violations"])
advisory = gate_id in ADVISORY_GATES and not args.strict
if advisory:
advisory_total += n
else:
blocking_total += n
tag = " (advisory)" if advisory else ""
status = (
"PASS" if n == 0 else (f"WARN ({n})" if advisory else f"FAIL ({n})")
)
print(f" Gate {gate_id} [{info['name']}]{tag}: {status}")
for v in info["violations"]:
print(f" {v['file']}:{v['line']} — {v.get('detail', v)}")
summary = (
"ALL BLOCKING GATES PASS"
if blocking_total == 0
else f"{blocking_total} BLOCKING VIOLATIONS"
)
if advisory_total:
summary += f" ({advisory_total} advisory warnings)"
print(f"\n{summary}")
def _is_blocking(gid: str) -> bool:
return not (gid in ADVISORY_GATES and not args.strict)
return (
0
if all(
len(info["violations"]) == 0
for gid, info in all_violations.items()
if _is_blocking(gid)
)
else 1
)
if __name__ == "__main__":
sys.exit(main())