1236 lines
49 KiB
Python
1236 lines
49 KiB
Python
#!/usr/bin/env python3
|
|
"""Static analysis to detect un-hermetic test patterns that cause slow or flaky tests.
|
|
|
|
This module is used in two ways:
|
|
|
|
1. **As a pytest plugin** (automatic — no configuration needed):
|
|
When devx is installed, pytest auto-discovers this plugin via the
|
|
``pytest11`` entry point. Every ``pytest`` run statically analyzes
|
|
test files for patterns that cause slow, non-deterministic, or
|
|
non-hermetic tests and **fails the test run** if any violations are found.
|
|
|
|
The plugin also wraps ``subprocess.run`` at runtime to catch real
|
|
subprocess calls that leak through transitive call paths (e.g.
|
|
``CliRunner.invoke(main)`` → ``main()`` → ``update_doc_versions()``
|
|
→ ``subprocess.run()``). If a test spawns a real subprocess without
|
|
``@patch``, the test fails.
|
|
|
|
To disable for a specific run: ``--no-test-isolation``.
|
|
|
|
2. **As a standalone CLI** (for CI gates)::
|
|
|
|
python3 -m devx.tools.check_test_isolation [--test-path tests/]
|
|
|
|
Always exits non-zero on any hard violation. Transitive-subprocess
|
|
findings are reported as advisories (exit 0) since static analysis
|
|
can't predict early exits — the runtime audit is authoritative.
|
|
|
|
Patterns detected:
|
|
|
|
1. **Unpatched subprocess calls** — test functions that call
|
|
``subprocess.run/call/Popen/check_call/check_output`` without a
|
|
corresponding ``@patch`` decorator or ``with patch(...)`` context manager.
|
|
2. **Unpatched ``time.sleep``** — test functions that call ``time.sleep``
|
|
without patching it.
|
|
3. **Unpatched known-subprocess-helpers** — functions known to spawn
|
|
subprocesses (e.g. ``update_doc_versions``) called without patching.
|
|
4. **Unpatched I/O functions** — functions known to do filesystem or
|
|
network I/O (e.g. ``get_pat``, ``load_secrets``, ``requests.get``)
|
|
called without patching.
|
|
5. **Excessive iteration loops** — ``for _ in range(N)`` where N > 100.
|
|
6. **Module-level heavy imports** — importing ``httpx``, ``ansible``,
|
|
etc. at module level in test files slows collection for all tests.
|
|
7. **``importlib.reload`` without cleanup** — reloading a module in a
|
|
test mutates global state. Each reload must be paired with a
|
|
cleanup reload (or wrapped in try/finally) to restore defaults.
|
|
8. **Transitive subprocess leaks** — ``CliRunner.invoke(target)`` where
|
|
``target`` transitively calls ``subprocess.run`` without being patched.
|
|
Detected via static call-graph analysis (warning) AND runtime audit
|
|
(authoritative — fails the test if a real subprocess runs).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import subprocess # nosec B404
|
|
import sys
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
import click
|
|
|
|
from devx.i18n import _
|
|
|
|
# ── Configuration ─────────────────────────────────────────────────────────────
|
|
|
|
DEFAULT_MAX_LOOP_ITERATIONS = 100
|
|
|
|
# Heavy modules that are slow to import (>50ms). When imported at module
|
|
# level in a test file, they slow down test collection for ALL tests.
|
|
# Maps module name → approximate import time in milliseconds.
|
|
# NOTE: ``requests`` is excluded because it's a core devx dependency —
|
|
# it's loaded during collection regardless of whether test files import it.
|
|
HEAVY_MODULE_IMPORTS: dict[str, float] = {
|
|
"httpx": 80.0,
|
|
"aiohttp": 120.0,
|
|
"docker": 90.0,
|
|
"kubernetes": 200.0,
|
|
"boto3": 250.0,
|
|
"botocore": 200.0,
|
|
"ansible": 300.0,
|
|
"molecule": 150.0,
|
|
"cv2": 400.0,
|
|
"numpy": 100.0,
|
|
"pandas": 200.0,
|
|
"matplotlib": 300.0,
|
|
"PIL": 80.0,
|
|
"Pillow": 80.0,
|
|
"sqlalchemy": 150.0,
|
|
"django": 200.0,
|
|
"flask": 80.0,
|
|
"fastapi": 100.0,
|
|
"pydantic": 60.0,
|
|
}
|
|
|
|
# Functions known to spawn subprocesses. When a test calls any of these
|
|
# without patching them, the real subprocess runs.
|
|
# Maps function name → human-readable description.
|
|
KNOWN_SUBPROCESS_HELPERS: dict[str, str] = {
|
|
"update_doc_versions": "calls subprocess.run to run check_doc_versions --fix",
|
|
"run_tests": "calls run_cmd to run make lint-ruff and make pytest-cov",
|
|
"run_cmd": "calls subprocess.run for shell commands",
|
|
}
|
|
|
|
# Functions known to do filesystem or network I/O that should be mocked in tests.
|
|
# Maps function name → description of what I/O it does.
|
|
# If a test calls one of these without a corresponding @patch, it's a violation.
|
|
KNOWN_IO_FUNCTIONS: dict[str, str] = { # nosec B105 — descriptions, not passwords
|
|
"get_pat": "reads ZITADEL PAT from filesystem/env (ZitadelAuth._iter_sources)",
|
|
"load_secrets": "reads YAML config file from disk",
|
|
"get_customer_secret": "reads customer-specific config from disk",
|
|
"get_customer_vm_ip": "queries Hetzner Cloud API for VM IP (network I/O)",
|
|
"get_observability_vm_ip": "queries Hetzner Cloud API for observability VM IP (network I/O)",
|
|
"requests.get": "performs HTTP GET to a real server",
|
|
"requests.post": "performs HTTP POST to a real server",
|
|
"requests.put": "performs HTTP PUT to a real server",
|
|
"requests.patch": "performs HTTP PATCH to a real server",
|
|
"requests.delete": "performs HTTP DELETE to a real server",
|
|
"urlopen": "performs HTTP request to a real server",
|
|
"httpx.get": "performs HTTP GET to a real server",
|
|
"httpx.post": "performs HTTP POST to a real server",
|
|
}
|
|
|
|
# Transitive dependencies: if a helper calls another helper that is patched,
|
|
# the call is safe. Maps helper → set of function names it internally calls.
|
|
# If ANY of these are in the test's patches, the helper call is safe.
|
|
HELPER_INTERNAL_CALLS: dict[str, set[str]] = {
|
|
"run_tests": {"run_cmd", "subprocess"},
|
|
"update_doc_versions": {"subprocess"},
|
|
"run_cmd": {"subprocess"},
|
|
}
|
|
|
|
# subprocess functions that the runtime audit wraps.
|
|
_SUBPROCESS_FUNCS = ("run", "call", "check_call", "check_output", "Popen")
|
|
|
|
|
|
# ── Runtime subprocess audit ──────────────────────────────────────────────────
|
|
#
|
|
# The static AST analyzer can only see direct calls in test functions.
|
|
# It cannot trace transitive calls through CliRunner.invoke(main, ...)
|
|
# → main() → update_doc_versions() → subprocess.run().
|
|
#
|
|
# The runtime audit wraps subprocess functions during test execution.
|
|
# If a test does NOT @patch subprocess, the wrapper catches real calls.
|
|
# If a test DOES @patch subprocess, the patch overrides our wrapper
|
|
# (correct — the test is mocking it).
|
|
|
|
|
|
class _SubprocessAudit:
|
|
"""Thread-local audit tracker for real subprocess calls during tests."""
|
|
|
|
def __init__(self) -> None:
|
|
self._local = threading.local()
|
|
self._installed = False
|
|
self._originals: dict[str, object] = {}
|
|
|
|
def _ensure_installed(self) -> None:
|
|
"""Install wrappers on subprocess module (once)."""
|
|
if self._installed:
|
|
return
|
|
for name in _SUBPROCESS_FUNCS:
|
|
original = getattr(subprocess, name, None)
|
|
if original is None:
|
|
continue
|
|
self._originals[name] = original
|
|
setattr(subprocess, name, self._make_wrapper(name, original))
|
|
self._installed = True
|
|
|
|
def _make_wrapper(self, name: str, original: object) -> object:
|
|
"""Create a wrapper that records calls when auditing is active."""
|
|
|
|
def wrapper(*args: object, **kwargs: object) -> object:
|
|
calls = getattr(self._local, "calls", None)
|
|
if calls is not None:
|
|
# Extract command for diagnostics
|
|
cmd = args[0] if args else kwargs.get("args", "?")
|
|
if isinstance(cmd, (list, tuple)) and cmd:
|
|
cmd_str = " ".join(str(c) for c in cmd[:4])
|
|
if len(cmd) > 4:
|
|
cmd_str += " ..."
|
|
else:
|
|
cmd_str = str(cmd)
|
|
calls.append((name, cmd_str))
|
|
return original(*args, **kwargs) # type: ignore[misc]
|
|
|
|
return wrapper
|
|
|
|
def start_test(self) -> None:
|
|
"""Begin auditing subprocess calls for the current test."""
|
|
self._ensure_installed()
|
|
self._local.calls = []
|
|
|
|
def stop_test(self) -> list[tuple[str, str]]:
|
|
"""Stop auditing and return recorded calls."""
|
|
calls = getattr(self._local, "calls", [])
|
|
self._local.calls = None
|
|
return calls
|
|
|
|
|
|
# Singleton instance used by the pytest plugin
|
|
_audit = _SubprocessAudit()
|
|
|
|
|
|
# ── Data structures ───────────────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class Violation:
|
|
"""A single isolation violation found in a test file."""
|
|
|
|
file: Path
|
|
line: int
|
|
col: int
|
|
category: str
|
|
message: str
|
|
|
|
def format(self) -> str:
|
|
try:
|
|
rel = self.file.relative_to(Path.cwd())
|
|
except ValueError:
|
|
rel = self.file
|
|
return f"{rel}:{self.line}:{self.col}: [{self.category}] {self.message}"
|
|
|
|
|
|
@dataclass
|
|
class TestFunctionInfo:
|
|
"""Information about a test function or method."""
|
|
|
|
name: str
|
|
node: ast.FunctionDef | ast.AsyncFunctionDef
|
|
patches: set[str] = field(default_factory=set)
|
|
class_patches: set[str] = field(default_factory=set)
|
|
is_test: bool = False
|
|
|
|
|
|
# ── AST helpers ───────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _extract_patch_targets(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> set[str]:
|
|
"""Extract @patch targets from decorators AND ``with patch(...)`` statements.
|
|
|
|
Detects:
|
|
- ``@patch("module.func")`` decorators
|
|
- ``with patch("module.func")`` context managers
|
|
- ``with patch.object(module, "func")`` context managers
|
|
- ``with patch("a"), patch("b")`` multiple patches
|
|
"""
|
|
targets: set[str] = set()
|
|
|
|
def _process_patch_call(call: ast.Call) -> None:
|
|
"""Extract target from a patch() or patch.object() call."""
|
|
func = call.func
|
|
# patch("module.func") — either bare `patch(...)` or `mock.patch(...)`
|
|
if (isinstance(func, ast.Name) and func.id == "patch") or (
|
|
isinstance(func, ast.Attribute) and func.attr == "patch"
|
|
):
|
|
if call.args and isinstance(call.args[0], ast.Constant) and isinstance(call.args[0].value, str):
|
|
target = call.args[0].value
|
|
targets.add(target)
|
|
targets.add(target.rsplit(".", 1)[-1])
|
|
# patch.object(module, "func") — extract short name from 2nd arg
|
|
elif (
|
|
isinstance(func, ast.Attribute)
|
|
and func.attr == "object"
|
|
and isinstance(func.value, ast.Name)
|
|
and func.value.id == "patch"
|
|
and len(call.args) >= 2
|
|
and isinstance(call.args[1], ast.Constant)
|
|
and isinstance(call.args[1].value, str)
|
|
and call.args[0]
|
|
and isinstance(call.args[0], ast.Name)
|
|
):
|
|
short = call.args[1].value
|
|
targets.add(short)
|
|
# We can't resolve the module alias here, but the short
|
|
# name is enough for patch matching in the call graph.
|
|
|
|
# 1. Extract from decorators
|
|
for decorator in node.decorator_list:
|
|
if isinstance(decorator, ast.Call):
|
|
_process_patch_call(decorator)
|
|
|
|
# 2. Extract from `with patch(...)` context managers in the body
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
for child in ast.walk(node):
|
|
if isinstance(child, ast.With):
|
|
for item in child.items:
|
|
ctx = item.context_expr
|
|
if isinstance(ctx, ast.Call):
|
|
_process_patch_call(ctx)
|
|
|
|
return targets
|
|
|
|
|
|
def _is_test_function(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
|
|
return node.name.startswith("test_")
|
|
|
|
|
|
def _has_integration_marker(node: ast.FunctionDef | ast.AsyncFunctionDef) -> bool:
|
|
"""Check if a test function has @pytest.mark.integration decorator."""
|
|
for decorator in node.decorator_list:
|
|
# @pytest.mark.integration → ast.Attribute(attr='integration')
|
|
if isinstance(decorator, ast.Attribute) and decorator.attr == "integration":
|
|
return True
|
|
# @pytest.mark.integration(...) → ast.Call(func=ast.Attribute(attr='integration'))
|
|
if isinstance(decorator, ast.Call):
|
|
func = decorator.func
|
|
if isinstance(func, ast.Attribute) and func.attr == "integration":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _get_called_name(node: ast.Call) -> str | None:
|
|
func = node.func
|
|
if isinstance(func, ast.Name):
|
|
return func.id
|
|
if isinstance(func, ast.Attribute):
|
|
return func.attr
|
|
return None
|
|
|
|
|
|
def _get_full_called_name(node: ast.Call) -> str | None:
|
|
func = node.func
|
|
parts: list[str] = []
|
|
current = func
|
|
while isinstance(current, ast.Attribute):
|
|
parts.append(current.attr)
|
|
current = current.value
|
|
if isinstance(current, ast.Name):
|
|
parts.append(current.id)
|
|
parts.reverse()
|
|
if not parts:
|
|
return None
|
|
return ".".join(parts)
|
|
|
|
|
|
def _get_range_count(node: ast.Call) -> int | None:
|
|
if not isinstance(node.func, ast.Name) or node.func.id != "range":
|
|
return None
|
|
if not node.args:
|
|
return None
|
|
# range(N) — single argument
|
|
if len(node.args) == 1:
|
|
arg = node.args[0]
|
|
if isinstance(arg, ast.Constant) and isinstance(arg.value, int):
|
|
return arg.value
|
|
return None
|
|
# range(start, stop) — two or more arguments
|
|
if len(node.args) >= 2:
|
|
stop = node.args[1]
|
|
if not isinstance(stop, ast.Constant) or not isinstance(stop.value, int):
|
|
return None
|
|
start = node.args[0]
|
|
if isinstance(start, ast.Constant) and isinstance(start.value, int):
|
|
return stop.value - start.value
|
|
# Non-constant start — assume 0
|
|
return stop.value
|
|
return None # pragma: no cover
|
|
|
|
|
|
# ── Call-graph builder ────────────────────────────────────────────────────────
|
|
#
|
|
# The static AST analyzer can only see direct calls in test functions.
|
|
# It cannot trace transitive calls through CliRunner.invoke(main, ...)
|
|
# → main() → update_doc_versions() → subprocess.run().
|
|
#
|
|
# The call-graph builder parses all source files in the package and builds
|
|
# a map: function_name → set of function_names it calls.
|
|
# When a test calls runner.invoke(target, ...), we trace the call graph
|
|
# from target to find all reachable functions, then check if any of them
|
|
# call subprocess.run (or other dangerous functions) without being patched.
|
|
|
|
|
|
# Dangerous functions that should never run in unit tests.
|
|
# Maps full call name → description.
|
|
_DANGEROUS_CALLS: dict[str, str] = {
|
|
"subprocess.run": "spawns a real subprocess",
|
|
"subprocess.call": "spawns a real subprocess",
|
|
"subprocess.check_call": "spawns a real subprocess",
|
|
"subprocess.check_output": "spawns a real subprocess",
|
|
"subprocess.Popen": "spawns a real subprocess",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class _FunctionNode:
|
|
"""AST node for a function with its called names."""
|
|
|
|
name: str
|
|
module: str
|
|
calls: set[str] # short names of functions called
|
|
subprocess_calls: set[str] # dangerous subprocess calls made directly
|
|
io_calls: set[str] # known I/O function calls made directly
|
|
|
|
|
|
class CallGraph:
|
|
"""Call graph built from source files in a package directory."""
|
|
|
|
def __init__(self, src_dir: Path) -> None:
|
|
self.src_dir = src_dir
|
|
# Maps "module.func" → _FunctionNode
|
|
self._nodes: dict[str, _FunctionNode] = {}
|
|
# Maps short name → list of full names (for resolution)
|
|
self._by_short: dict[str, list[str]] = {}
|
|
self._built = False
|
|
|
|
def _ensure_built(self) -> None:
|
|
if self._built:
|
|
return
|
|
self._build()
|
|
self._built = True
|
|
|
|
def _build(self) -> None:
|
|
"""Parse all .py files under src_dir and build the call graph."""
|
|
for py_file in sorted(self.src_dir.rglob("*.py")):
|
|
try:
|
|
source = py_file.read_text()
|
|
tree = ast.parse(source, filename=str(py_file))
|
|
except (SyntaxError, UnicodeDecodeError):
|
|
continue
|
|
# Derive module name from path relative to src_dir
|
|
rel = py_file.relative_to(self.src_dir)
|
|
module_parts = list(rel.with_suffix("").parts)
|
|
if module_parts and module_parts[-1] == "__init__":
|
|
module_parts = module_parts[:-1]
|
|
module = ".".join(module_parts)
|
|
self._scan_module(tree, module)
|
|
|
|
def _scan_module(self, tree: ast.Module, module: str) -> None:
|
|
"""Scan a module AST and register all top-level functions.
|
|
|
|
Methods defined inside classes are NOT registered — they are called
|
|
via objects (e.g. ``tea.create_issue()``) and resolving them by short
|
|
name alone causes false positives when the class is patched (e.g.
|
|
``@patch("...TeaCLI")`` mocks all methods).
|
|
"""
|
|
for node in tree.body:
|
|
self._scan_node(node, module)
|
|
|
|
def _scan_node(self, node: ast.AST, module: str) -> None:
|
|
"""Recursively scan a node, registering non-method functions."""
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
self._register_function(node, module)
|
|
# Don't recurse into function bodies — nested functions are
|
|
# not callable by name from outside.
|
|
return
|
|
if isinstance(node, ast.ClassDef):
|
|
# Skip class body — methods are not registered.
|
|
return
|
|
# Recurse into other compound statements (if/for/try/with/etc.)
|
|
for child in ast.iter_child_nodes(node):
|
|
self._scan_node(child, module)
|
|
|
|
def _register_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef, module: str) -> None:
|
|
"""Register a function and its direct calls in the call graph."""
|
|
full_name = f"{module}.{node.name}"
|
|
calls: set[str] = set()
|
|
subprocess_calls: set[str] = set()
|
|
io_calls: set[str] = set()
|
|
|
|
for child in ast.walk(node):
|
|
if isinstance(child, ast.Call):
|
|
full = _get_full_called_name(child)
|
|
short = _get_called_name(child)
|
|
if short:
|
|
calls.add(short)
|
|
if full and full in _DANGEROUS_CALLS:
|
|
subprocess_calls.add(full)
|
|
if short and short in KNOWN_IO_FUNCTIONS:
|
|
io_calls.add(short)
|
|
# KNOWN_SUBPROCESS_HELPERS are intermediate functions (e.g.
|
|
# run_tests → run_cmd → subprocess.run). They are already
|
|
# in *calls* so the BFS will traverse into them and find the
|
|
# actual subprocess call. Adding them to *subprocess_calls*
|
|
# here would cause false positives when the helper itself is
|
|
# transitively patched (e.g. run_cmd is patched → run_tests
|
|
# is safe, but would still be reported).
|
|
|
|
fn_node = _FunctionNode(
|
|
name=node.name,
|
|
module=module,
|
|
calls=calls,
|
|
subprocess_calls=subprocess_calls,
|
|
io_calls=io_calls,
|
|
)
|
|
self._nodes[full_name] = fn_node
|
|
self._by_short.setdefault(node.name, []).append(full_name)
|
|
|
|
def find_reachable_dangerous(
|
|
self,
|
|
target_name: str,
|
|
patches: set[str],
|
|
max_depth: int = 10,
|
|
import_map: dict[str, str] | None = None,
|
|
) -> list[tuple[str, str]]:
|
|
"""Find all dangerous calls reachable from target_name that aren't patched.
|
|
|
|
Returns a list of (function_name, description) tuples for each
|
|
unpatched dangerous call found in the transitive closure.
|
|
|
|
If import_map is provided (mapping short names to fully-qualified
|
|
module paths), it's used to resolve the target precisely instead
|
|
of matching by short name alone.
|
|
"""
|
|
self._ensure_built()
|
|
|
|
# Resolve target to full name(s)
|
|
# First try precise resolution via import_map
|
|
candidates: list[str] = []
|
|
if import_map and target_name in import_map:
|
|
full = import_map[target_name]
|
|
candidates = [full] if full in self._nodes else self._by_short.get(target_name, [])
|
|
elif target_name in self._nodes:
|
|
# Already a fully-qualified name (e.g. devx.tools.build_image.main)
|
|
candidates = [target_name]
|
|
else:
|
|
# Fall back to short name resolution
|
|
short = target_name.rsplit(".", 1)[-1]
|
|
candidates = self._by_short.get(short, [])
|
|
|
|
if not candidates:
|
|
return []
|
|
|
|
visited: set[str] = set()
|
|
dangerous: list[tuple[str, str]] = []
|
|
queue: list[tuple[str, int]] = [(c, 0) for c in candidates]
|
|
|
|
while queue:
|
|
full_name, depth = queue.pop(0)
|
|
if full_name in visited or depth > max_depth:
|
|
continue
|
|
visited.add(full_name)
|
|
|
|
node = self._nodes.get(full_name)
|
|
if node is None:
|
|
continue
|
|
|
|
# Check direct subprocess calls
|
|
for sc in node.subprocess_calls:
|
|
short = sc.rsplit(".", 1)[-1]
|
|
if not self._is_patched(sc, short, patches):
|
|
desc = _DANGEROUS_CALLS.get(sc, "")
|
|
dangerous.append((full_name, desc))
|
|
|
|
# Check direct IO calls
|
|
for io in node.io_calls:
|
|
if not self._is_patched(io, io, patches):
|
|
desc = KNOWN_IO_FUNCTIONS.get(io, "")
|
|
if desc:
|
|
dangerous.append((full_name, desc))
|
|
|
|
# Enqueue called functions — skip if the called function is patched
|
|
for called_short in node.calls:
|
|
if self._is_patched(called_short, called_short, patches):
|
|
continue
|
|
# Prefer same-module resolution, then fall back to short name
|
|
# only if there's a single global match (avoids false positives
|
|
# when multiple modules define functions with the same name).
|
|
same_module = f"{node.module}.{called_short}"
|
|
if same_module in self._nodes and same_module not in visited:
|
|
queue.append((same_module, depth + 1))
|
|
else:
|
|
matches = self._by_short.get(called_short, [])
|
|
if len(matches) == 1 and matches[0] not in visited:
|
|
queue.append((matches[0], depth + 1))
|
|
|
|
return dangerous
|
|
|
|
@staticmethod
|
|
def _is_patched(full: str, short: str, patches: set[str]) -> bool:
|
|
"""Check if a function is covered by the test's @patch set."""
|
|
if short in patches or full in patches:
|
|
return True
|
|
# Check if any patch entry ends with ".short" (e.g. "subprocess.run"
|
|
# is patched by "devx.ci.release.subprocess.run"). Use exact
|
|
# endswith, not substring, to avoid "run" matching "run_cmd".
|
|
return any(p.endswith(f".{short}") or p == full for p in patches)
|
|
|
|
|
|
# ── Analyzers ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
class TestIsolationVisitor(ast.NodeVisitor):
|
|
"""AST visitor that detects un-hermetic test patterns."""
|
|
|
|
def __init__(
|
|
self,
|
|
file_path: Path,
|
|
max_loop_iterations: int = DEFAULT_MAX_LOOP_ITERATIONS,
|
|
call_graph: CallGraph | None = None,
|
|
):
|
|
self.file_path = file_path
|
|
self.max_loop_iterations = max_loop_iterations
|
|
self.call_graph = call_graph
|
|
self.violations: list[Violation] = []
|
|
self._current_function: TestFunctionInfo | None = None
|
|
self._current_class_patches: set[str] = set()
|
|
self._in_test_class = False
|
|
self._reload_calls: list[tuple[int, str | None]] = []
|
|
# Import map: short name → fully-qualified module.func
|
|
# e.g. {"main": "devx.ci.release.main"} for `from devx.ci.release import main`
|
|
self._import_map: dict[str, str] = {}
|
|
|
|
def visit_Import(self, node: ast.Import) -> None:
|
|
# Track imports for call-graph resolution
|
|
if self._current_function is None:
|
|
for alias in node.names:
|
|
name = alias.asname or alias.name
|
|
self._import_map[name] = alias.name
|
|
# Check for heavy module imports
|
|
if self._current_function is None:
|
|
for alias in node.names:
|
|
mod = alias.name.split(".")[0]
|
|
if mod in HEAVY_MODULE_IMPORTS:
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="heavy-module-import",
|
|
message=_(
|
|
"Heavy import '{mod}' (~{ms:.0f}ms) at module level — "
|
|
"this slows test collection for all tests. "
|
|
"Move inside test functions or use lazy import.",
|
|
mod=alias.name,
|
|
ms=HEAVY_MODULE_IMPORTS[mod],
|
|
),
|
|
)
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
|
|
# Track imports for call-graph resolution
|
|
if self._current_function is None and node.module:
|
|
for alias in node.names:
|
|
name = alias.asname or alias.name
|
|
self._import_map[name] = f"{node.module}.{alias.name}"
|
|
# Check for heavy module imports
|
|
if self._current_function is None and node.module:
|
|
mod = node.module.split(".")[0]
|
|
if mod in HEAVY_MODULE_IMPORTS:
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="heavy-module-import",
|
|
message=_(
|
|
"Heavy import '{mod}' (~{ms:.0f}ms) at module level — "
|
|
"this slows test collection for all tests. "
|
|
"Move inside test functions or use lazy import.",
|
|
mod=node.module,
|
|
ms=HEAVY_MODULE_IMPORTS[mod],
|
|
),
|
|
)
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
def visit_ClassDef(self, node: ast.ClassDef) -> None:
|
|
old_class_patches = self._current_class_patches
|
|
old_in_test = self._in_test_class
|
|
self._current_class_patches = _extract_patch_targets(node)
|
|
self._in_test_class = node.name.startswith("Test")
|
|
self.generic_visit(node)
|
|
self._current_class_patches = old_class_patches
|
|
self._in_test_class = old_in_test
|
|
|
|
def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
|
|
self._visit_function(node)
|
|
|
|
def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
|
|
self._visit_function(node)
|
|
|
|
def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
|
|
if not _is_test_function(node):
|
|
self.generic_visit(node)
|
|
return
|
|
|
|
# Skip integration tests — they intentionally do real I/O
|
|
if _has_integration_marker(node):
|
|
self.generic_visit(node)
|
|
return
|
|
|
|
patches = _extract_patch_targets(node)
|
|
info = TestFunctionInfo(
|
|
name=node.name,
|
|
node=node,
|
|
patches=patches,
|
|
class_patches=self._current_class_patches,
|
|
is_test=True,
|
|
)
|
|
old_func = self._current_function
|
|
old_reloads = self._reload_calls
|
|
self._current_function = info
|
|
self._reload_calls = []
|
|
self.generic_visit(node)
|
|
# Check 7: importlib.reload without cleanup
|
|
# Each reload mutates global module state. An odd number of
|
|
# reloads means the module is left in a modified state.
|
|
if len(self._reload_calls) % 2 != 0:
|
|
first_line, mod_name = self._reload_calls[0]
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=first_line,
|
|
col=0,
|
|
category="reload-without-cleanup",
|
|
message=_(
|
|
"importlib.reload({mod}) called {n} time(s) in test '{test}' — "
|
|
"odd count leaves module in modified state. "
|
|
"Add a final reload to restore defaults or wrap in try/finally.",
|
|
mod=mod_name or "module",
|
|
n=len(self._reload_calls),
|
|
test=info.name,
|
|
),
|
|
)
|
|
)
|
|
self._current_function = old_func
|
|
self._reload_calls = old_reloads
|
|
|
|
def visit_Call(self, node: ast.Call) -> None:
|
|
if self._current_function is None:
|
|
self.generic_visit(node)
|
|
return
|
|
|
|
full_name = _get_full_called_name(node)
|
|
short_name = _get_called_name(node)
|
|
all_patches = self._current_function.patches | self._current_function.class_patches
|
|
|
|
# Track importlib.reload calls for cleanup check
|
|
if full_name == "importlib.reload" or (short_name == "reload" and "reload" in all_patches):
|
|
mod_arg = node.args[0] if node.args else None
|
|
mod_name = None
|
|
if isinstance(mod_arg, ast.Name):
|
|
mod_name = mod_arg.id
|
|
elif isinstance(mod_arg, ast.Attribute):
|
|
mod_name = mod_arg.attr
|
|
self._reload_calls.append((node.lineno, mod_name))
|
|
|
|
# Check 1: subprocess.run / subprocess.call / subprocess.Popen etc.
|
|
if full_name and full_name.startswith("subprocess."):
|
|
method = full_name.split(".", 1)[1]
|
|
if method in ("run", "call", "Popen", "check_call", "check_output") and not any(
|
|
"subprocess" in p for p in all_patches
|
|
):
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="unpatched-subprocess",
|
|
message=_(
|
|
"{call} called in test '{test}' without @patch — "
|
|
"this spawns a real subprocess. Add "
|
|
'@patch("<module>.subprocess.run") or patch the calling function.',
|
|
call=full_name,
|
|
test=self._current_function.name,
|
|
),
|
|
)
|
|
)
|
|
|
|
# Check 2: time.sleep
|
|
if (
|
|
(full_name == "time.sleep" or (short_name == "sleep" and "sleep" not in all_patches))
|
|
and "sleep" not in all_patches
|
|
and "time.sleep" not in all_patches
|
|
and not any("sleep" in p for p in all_patches)
|
|
):
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="unpatched-sleep",
|
|
message=_(
|
|
"time.sleep called in test '{test}' without @patch — "
|
|
"this causes real wall-clock delays. Add "
|
|
'@patch("<module>.time.sleep").',
|
|
test=self._current_function.name,
|
|
),
|
|
)
|
|
)
|
|
|
|
# Check 3: Known subprocess helpers
|
|
if short_name in KNOWN_SUBPROCESS_HELPERS and not (
|
|
short_name in all_patches
|
|
or any("subprocess" in p for p in all_patches)
|
|
or any(
|
|
dep in all_patches or any(dep in p for p in all_patches)
|
|
for dep in HELPER_INTERNAL_CALLS.get(short_name, set())
|
|
)
|
|
):
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="unpatched-helper",
|
|
message=_(
|
|
"{func} called in test '{test}' without @patch — "
|
|
'this function {desc}. Add @patch("<module>.{func}").',
|
|
func=short_name,
|
|
test=self._current_function.name,
|
|
desc=KNOWN_SUBPROCESS_HELPERS[short_name],
|
|
),
|
|
)
|
|
)
|
|
|
|
# Check 4: Known I/O functions (filesystem/network)
|
|
# Match by short name (e.g. "get_pat") or full name (e.g. "requests.get")
|
|
sn = short_name or ""
|
|
io_key = sn if sn in KNOWN_IO_FUNCTIONS else None
|
|
if io_key is None and full_name and full_name in KNOWN_IO_FUNCTIONS:
|
|
io_key = full_name
|
|
if io_key and not (
|
|
io_key in all_patches
|
|
or sn in all_patches
|
|
or any(io_key in p or sn in p for p in all_patches)
|
|
or any(p.endswith(f".{sn}") for p in all_patches)
|
|
):
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="unpatched-io",
|
|
message=_(
|
|
"{func} called in test '{test}' without @patch — "
|
|
'this function {desc}. Add @patch("<module>.{func}").',
|
|
func=io_key,
|
|
test=self._current_function.name,
|
|
desc=KNOWN_IO_FUNCTIONS[io_key],
|
|
),
|
|
)
|
|
)
|
|
|
|
# Check 8: CliRunner.invoke / runner.invoke — trace call graph
|
|
# Detect runner.invoke(target, ...) or CliRunner().invoke(target, ...)
|
|
if short_name == "invoke" and self.call_graph is not None and node.args:
|
|
target = node.args[0]
|
|
target_name: str | None = None
|
|
if isinstance(target, ast.Name):
|
|
target_name = target.id
|
|
elif isinstance(target, ast.Attribute):
|
|
# Handle module.func pattern (e.g. build_image.main)
|
|
# Resolve module prefix via import_map
|
|
if isinstance(target.value, ast.Name):
|
|
mod_short = target.value.id
|
|
mod_full = self._import_map.get(mod_short)
|
|
target_name = f"{mod_full}.{target.attr}" if mod_full else target.attr
|
|
else:
|
|
target_name = target.attr
|
|
if target_name:
|
|
dangerous = self.call_graph.find_reachable_dangerous(
|
|
target_name, all_patches, import_map=self._import_map
|
|
)
|
|
if dangerous:
|
|
# Deduplicate by function name
|
|
seen: set[str] = set()
|
|
unique: list[tuple[str, str]] = []
|
|
for func, desc in dangerous:
|
|
if func not in seen:
|
|
seen.add(func)
|
|
unique.append((func, desc))
|
|
funcs_desc = "; ".join(f"{f} ({d})" for f, d in unique[:3])
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="transitive-subprocess",
|
|
message=_(
|
|
"CliRunner.invoke({target}) in test '{test}' reaches "
|
|
"unpatched dangerous functions: {funcs}. "
|
|
"Add @patch for each or patch the calling function.",
|
|
target=target_name,
|
|
test=self._current_function.name,
|
|
funcs=funcs_desc,
|
|
),
|
|
)
|
|
)
|
|
|
|
self.generic_visit(node)
|
|
|
|
def visit_For(self, node: ast.For) -> None:
|
|
if self._current_function is not None and isinstance(node.iter, ast.Call):
|
|
count = _get_range_count(node.iter)
|
|
if count is not None and count > self.max_loop_iterations:
|
|
self.violations.append(
|
|
Violation(
|
|
file=self.file_path,
|
|
line=node.lineno,
|
|
col=node.col_offset,
|
|
category="excessive-iterations",
|
|
message=_(
|
|
"Loop with {count} iterations in test '{test}' — "
|
|
"consider property-based testing (hypothesis) or reduce to <= {max} iterations.",
|
|
count=count,
|
|
test=self._current_function.name,
|
|
max=self.max_loop_iterations,
|
|
),
|
|
)
|
|
)
|
|
self.generic_visit(node)
|
|
|
|
|
|
# ── File scanning (shared by CLI and pytest plugin) ──────────────────────────
|
|
|
|
|
|
def find_test_files(test_path: Path) -> list[Path]:
|
|
"""Find all Python test files under the given path."""
|
|
if test_path.is_file():
|
|
return [test_path] if test_path.suffix == ".py" else []
|
|
return sorted(test_path.rglob("test_*.py"))
|
|
|
|
|
|
def analyze_file(
|
|
file_path: Path,
|
|
max_loop_iterations: int = DEFAULT_MAX_LOOP_ITERATIONS,
|
|
call_graph: CallGraph | None = None,
|
|
) -> list[Violation]:
|
|
"""Analyze a single test file for isolation violations.
|
|
|
|
Files in ``integration/`` directories are skipped — integration tests
|
|
intentionally do real I/O (subprocess, network, filesystem).
|
|
"""
|
|
if "integration" in file_path.parts:
|
|
return []
|
|
try:
|
|
source = file_path.read_text()
|
|
tree = ast.parse(source, filename=str(file_path))
|
|
except SyntaxError as exc:
|
|
return [
|
|
Violation(
|
|
file=file_path,
|
|
line=exc.lineno or 0,
|
|
col=exc.offset or 0,
|
|
category="syntax-error",
|
|
message=f"Could not parse file: {exc}",
|
|
)
|
|
]
|
|
|
|
visitor = TestIsolationVisitor(file_path, max_loop_iterations, call_graph)
|
|
visitor.visit(tree)
|
|
return visitor.violations
|
|
|
|
|
|
def analyze_test_files(
|
|
test_path: Path,
|
|
max_loop_iterations: int = DEFAULT_MAX_LOOP_ITERATIONS,
|
|
categories: set[str] | None = None,
|
|
call_graph: CallGraph | None = None,
|
|
) -> list[Violation]:
|
|
"""Analyze all test files under test_path. Returns list of violations."""
|
|
test_files = find_test_files(test_path)
|
|
all_violations: list[Violation] = []
|
|
for file_path in test_files:
|
|
violations = analyze_file(file_path, max_loop_iterations, call_graph)
|
|
if categories:
|
|
violations = [v for v in violations if v.category in categories]
|
|
all_violations.extend(violations)
|
|
return all_violations
|
|
|
|
|
|
# ── Pytest plugin ─────────────────────────────────────────────────────────────
|
|
#
|
|
# When devx is installed, pytest auto-discovers this plugin via the
|
|
# `pytest11` entry point. The plugin runs static analysis on every
|
|
# test file during collection and **fails** on any violation.
|
|
# It also wraps subprocess at runtime to catch transitive leaks.
|
|
|
|
|
|
def pytest_addoption(parser): # type: ignore[no-untyped-def] # pragma: no cover
|
|
"""Register pytest command-line options."""
|
|
parser.addoption(
|
|
"--no-test-isolation",
|
|
action="store_true",
|
|
default=False,
|
|
help="Disable test isolation static analysis and runtime subprocess audit.",
|
|
)
|
|
parser.addoption(
|
|
"--test-isolation-max-loop",
|
|
type=int,
|
|
default=DEFAULT_MAX_LOOP_ITERATIONS,
|
|
help=f"Max iterations allowed in a test loop (default: {DEFAULT_MAX_LOOP_ITERATIONS}).",
|
|
)
|
|
|
|
|
|
def pytest_collection_finish(session): # type: ignore[no-untyped-def] # pragma: no cover
|
|
"""Run static analysis after all test files are collected. Always strict."""
|
|
if session.config.getoption("--no-test-isolation"):
|
|
return
|
|
|
|
max_loop = session.config.getoption("--test-isolation-max-loop")
|
|
|
|
# Build call graph from source directory for transitive analysis
|
|
call_graph: CallGraph | None = None
|
|
for item in session.items:
|
|
fspath = Path(str(item.fspath))
|
|
for parent in fspath.parents:
|
|
src_dir = parent / "src"
|
|
if src_dir.is_dir():
|
|
call_graph = CallGraph(src_dir)
|
|
break
|
|
if call_graph is not None:
|
|
break
|
|
|
|
test_files: set[Path] = set()
|
|
for item in session.items:
|
|
test_files.add(Path(str(item.fspath)))
|
|
|
|
all_violations: list[Violation] = []
|
|
for file_path in sorted(test_files):
|
|
violations = analyze_file(file_path, max_loop, call_graph)
|
|
all_violations.extend(violations)
|
|
|
|
if not all_violations:
|
|
return
|
|
|
|
# transitive-subprocess is advisory (static can't predict early exits).
|
|
# All other categories are hard errors.
|
|
errors = [v for v in all_violations if v.category != "transitive-subprocess"]
|
|
transitive = [v for v in all_violations if v.category == "transitive-subprocess"]
|
|
|
|
if errors:
|
|
count = len(errors)
|
|
files = len({v.file for v in errors})
|
|
click.echo(
|
|
_(
|
|
"\nTest isolation check FAILED: {count} violation(s) in {files} file(s).\n",
|
|
count=count,
|
|
files=files,
|
|
),
|
|
err=True,
|
|
)
|
|
for v in sorted(errors, key=lambda x: (str(x.file), x.line)):
|
|
click.echo(f" {v.format()}", err=True)
|
|
click.echo(
|
|
_(
|
|
"Fix: add @patch decorators or with patch() context managers "
|
|
"for subprocess/time.sleep calls, or patch the calling function.\n"
|
|
),
|
|
err=True,
|
|
)
|
|
import pytest
|
|
|
|
pytest.fail(
|
|
f"Test isolation: {count} violation(s) found. See output above.",
|
|
pytrace=False,
|
|
)
|
|
|
|
# transitive-subprocess warnings are advisory — runtime audit is authoritative
|
|
if transitive:
|
|
import warnings
|
|
|
|
for v in sorted(transitive, key=lambda x: (str(x.file), x.line)):
|
|
msg = f"Test isolation advisory: {v.format()}"
|
|
warnings.warn(msg, UserWarning, stacklevel=2)
|
|
|
|
|
|
# ── Runtime subprocess audit hooks ────────────────────────────────────────────
|
|
|
|
|
|
def _is_integration_test(item: object) -> bool:
|
|
"""Check if a test item is an integration test."""
|
|
markers = getattr(item, "keywords", {})
|
|
if "integration" in markers:
|
|
return True
|
|
fspath = str(getattr(item, "fspath", ""))
|
|
return "integration" in fspath
|
|
|
|
|
|
def pytest_runtest_setup(item: object) -> None: # type: ignore[no-untyped-def] # pragma: no cover
|
|
"""Start subprocess audit for non-integration tests."""
|
|
config = getattr(item, "config", None)
|
|
if config is None:
|
|
return
|
|
if config.getoption("--no-test-isolation"):
|
|
return
|
|
if _is_integration_test(item):
|
|
return
|
|
_audit.start_test()
|
|
|
|
|
|
def pytest_runtest_teardown(item: object, nextitem: object) -> None: # type: ignore[no-untyped-def] # pragma: no cover
|
|
"""Fail test if real subprocess calls were made without @patch."""
|
|
config = getattr(item, "config", None)
|
|
if config is None:
|
|
return
|
|
if config.getoption("--no-test-isolation"):
|
|
return
|
|
if _is_integration_test(item):
|
|
return
|
|
calls = _audit.stop_test()
|
|
if not calls:
|
|
return
|
|
|
|
test_name = getattr(item, "name", str(item))
|
|
lines = [
|
|
_(
|
|
"Real subprocess call(s) detected in test '{test}' without @patch:",
|
|
test=test_name,
|
|
)
|
|
]
|
|
for func_name, cmd in calls:
|
|
lines.append(f" {func_name}({cmd})")
|
|
lines.append(_('Add @patch("subprocess.run") or patch the calling function to fix this.'))
|
|
msg = "\n".join(lines)
|
|
|
|
import pytest
|
|
|
|
pytest.fail(msg, pytrace=False)
|
|
|
|
|
|
# ── Standalone CLI ────────────────────────────────────────────────────────────
|
|
|
|
|
|
@click.command()
|
|
@click.option(
|
|
"--test-path",
|
|
"test_paths",
|
|
type=click.Path(exists=True, path_type=Path),
|
|
multiple=True,
|
|
default=[Path("tests/")],
|
|
show_default=True,
|
|
help="Path to test directory or file to analyze (can be specified multiple times).",
|
|
)
|
|
@click.option(
|
|
"--max-loop-iterations",
|
|
type=int,
|
|
default=DEFAULT_MAX_LOOP_ITERATIONS,
|
|
show_default=True,
|
|
help="Maximum allowed iterations in a single test loop.",
|
|
)
|
|
@click.option(
|
|
"--categories",
|
|
type=str,
|
|
default="",
|
|
help="Comma-separated list of categories to check (default: all). "
|
|
"Available: unpatched-subprocess, unpatched-sleep, unpatched-helper, "
|
|
"excessive-iterations, heavy-module-import, reload-without-cleanup, "
|
|
"transitive-subprocess",
|
|
)
|
|
@click.option(
|
|
"--src-dir",
|
|
type=click.Path(exists=True, file_okay=False, path_type=Path),
|
|
default=None,
|
|
help="Source directory for call-graph analysis (auto-detected if omitted).",
|
|
)
|
|
def cli(
|
|
test_paths: tuple[Path, ...],
|
|
max_loop_iterations: int,
|
|
categories: str,
|
|
src_dir: Path | None,
|
|
) -> None:
|
|
"""Check test files for un-hermetic patterns that cause slow or flaky tests.
|
|
|
|
Always exits non-zero on any hard violation. Transitive-subprocess
|
|
findings are reported as advisories (exit 0) since static analysis
|
|
can't predict early exits — the runtime audit is authoritative.
|
|
"""
|
|
allowed: set[str] | None = None
|
|
if categories:
|
|
allowed = {c.strip() for c in categories.split(",")}
|
|
|
|
# Build call graph for transitive subprocess detection
|
|
call_graph: CallGraph | None = None
|
|
if src_dir is not None:
|
|
call_graph = CallGraph(src_dir)
|
|
else:
|
|
for tp in test_paths:
|
|
for parent in Path(tp).resolve().parents:
|
|
candidate = parent / "src"
|
|
if candidate.is_dir():
|
|
call_graph = CallGraph(candidate)
|
|
break
|
|
if call_graph is not None:
|
|
break
|
|
|
|
all_violations: list[Violation] = []
|
|
total_files = 0
|
|
for test_path in test_paths:
|
|
violations = analyze_test_files(test_path, max_loop_iterations, allowed, call_graph)
|
|
all_violations.extend(violations)
|
|
total_files += len(find_test_files(test_path))
|
|
|
|
errors = [v for v in all_violations if v.category != "transitive-subprocess"]
|
|
advisories = [v for v in all_violations if v.category == "transitive-subprocess"]
|
|
|
|
if not errors and not advisories:
|
|
click.echo(
|
|
_("Test isolation check passed: {count} test files analyzed, no violations found.", count=total_files)
|
|
)
|
|
sys.exit(0)
|
|
|
|
if errors:
|
|
click.echo(
|
|
_(
|
|
"Test isolation check FAILED: {count} violation(s) in {files} file(s).",
|
|
count=len(errors),
|
|
files=len({v.file for v in errors}),
|
|
),
|
|
err=True,
|
|
)
|
|
click.echo("")
|
|
for v in sorted(errors, key=lambda x: (str(x.file), x.line)):
|
|
click.echo(f" {v.format()}", err=True)
|
|
click.echo("")
|
|
click.echo(
|
|
_(
|
|
"Fix: add @patch decorators or with patch() context managers "
|
|
"for subprocess/time.sleep calls, or patch the calling function."
|
|
),
|
|
err=True,
|
|
)
|
|
sys.exit(1)
|
|
|
|
# Advisories only — exit 0 but print them
|
|
click.echo(
|
|
_(
|
|
"Test isolation check passed with {count} advisory warning(s) in {files} file(s).",
|
|
count=len(advisories),
|
|
files=len({v.file for v in advisories}),
|
|
)
|
|
)
|
|
click.echo(_("Transitive-subprocess advisories (runtime audit is authoritative):"))
|
|
for v in sorted(advisories, key=lambda x: (str(x.file), x.line))[:10]:
|
|
click.echo(f" {v.format()}")
|
|
if len(advisories) > 10:
|
|
click.echo(f" ... and {len(advisories) - 10} more")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
cli() # pragma: no cover
|