shared

_shared

Shared AST primitives for test_quality rules.

Registry-neutral helpers ported from the detect_pyramid_level_v6 and triage_tautologies_v4 prototypes. Rules in this subpackage import from here; no @register_rule lives here.

analyze_imports(tree, pkg_prefixes, init_all, pkg_root)

Classify imports + collect IO module names.

Returns (public, internal, modules, has_private, io_module_names, io_signals).

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def analyze_imports(
    tree: ast.Module,
    pkg_prefixes: set[str],
    init_all: set[str] | None,
    pkg_root: Path,
) -> tuple[list[str], list[str], list[str], bool, set[str], list[str]]:
    """Classify imports + collect IO module names.

    Returns ``(public, internal, modules, has_private, io_module_names,
    io_signals)``.
    """
    scan = _ImportScan()
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom) and node.module:
            _process_import_from(node, pkg_prefixes, init_all, pkg_root, scan)
        elif isinstance(node, ast.Import):
            _process_import(node, scan)
    return (
        scan.public,
        scan.internal,
        scan.import_modules,
        scan.has_private,
        scan.io_module_names,
        scan.io_signals,
    )

collect_pkg_contract_classes(pkg_root)

Contract classes (Protocol/ABC/TypedDict) in pkg_root and sibling packages.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def collect_pkg_contract_classes(pkg_root: Path) -> set[str]:
    """Contract classes (Protocol/ABC/TypedDict) in *pkg_root* and sibling packages."""
    out: set[str] = _scan_pkg_for_contracts(pkg_root)

    parent = pkg_root.parent
    if parent.name == "packages" and parent.exists():
        for sibling in parent.iterdir():
            if not sibling.is_dir() or sibling == pkg_root:
                continue
            out |= _scan_pkg_for_contracts(sibling)

    axm_core = (
        pkg_root.parent.parent.parent / "axm-nexus" / "packages" / "axm"
        if parent.name == "packages"
        else None
    )
    if axm_core is not None and axm_core.exists():
        out |= _scan_pkg_for_contracts(axm_core)

    return out

collect_pkg_public_symbols(pkg_root)

Collect top-level public function, class, and constant names across src/.

Walks every *.py file under {pkg_root}/src and returns the union of non-underscore names defined at module top level — functions, async functions, classes, and simple / annotated assignments to Name targets.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def collect_pkg_public_symbols(pkg_root: Path) -> set[str]:
    """Collect top-level public function, class, and constant names across ``src/``.

    Walks every ``*.py`` file under ``{pkg_root}/src`` and returns the union of
    non-underscore names defined at module top level — functions, async
    functions, classes, and simple / annotated assignments to ``Name`` targets.
    """
    out: set[str] = set()
    for path in _iter_src_py(pkg_root):
        tree = _parse_cached(path)
        if tree is None:
            continue
        for node in tree.body:
            out.update(_public_names_from_node(node))
    return out

current_level_from_path(test_file, tests_dir)

Map test_file to its pyramid level based on its location.

Parameters:

Name	Type	Description	Default
test_file	Path	Test file path to classify.	required
tests_dir	Path	Root tests directory used to compute the relative path.	required

Returns:

Type	Description
str	One of "unit", "integration", "e2e", or "root" when
str	the level cannot be inferred. functional folders are mapped to
str	"integration" for backward compatibility.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def current_level_from_path(test_file: Path, tests_dir: Path) -> str:
    """Map ``test_file`` to its pyramid level based on its location.

    Args:
        test_file: Test file path to classify.
        tests_dir: Root tests directory used to compute the relative path.

    Returns:
        One of ``"unit"``, ``"integration"``, ``"e2e"``, or ``"root"`` when
        the level cannot be inferred. ``functional`` folders are mapped to
        ``"integration"`` for backward compatibility.
    """
    try:
        rel = test_file.relative_to(tests_dir)
    except ValueError:
        return "root"
    parts = rel.parts
    if len(parts) > 1:
        first = parts[0]
        if first in ("unit", "integration", "e2e"):
            return first
        if first == "functional":
            return "integration"
    return "root"

detect_real_io(tree)

File-scope I/O detection from a parsed test module.

Scans the module for three kinds of evidence that a test performs real I/O: fixtures listed in _IO_FIXTURES consumed by test_* functions, dotted calls listed in _IO_CALLS (e.g. subprocess.run, pathlib.Path.write_text), and CLI runner invocations such as CliRunner().invoke(...). Attribute-IO inside helpers is intentionally out of scope — see :func:func_attr_io_transitive.

Returns:

Type	Description
bool	(has_io, has_subprocess, signals). has_io is True when any
bool	signal was found. has_subprocess is True when a
list[str]	subprocess.* call or a CLI runner was seen. signals is the
tuple[bool, bool, list[str]]	list of evidence strings (fixture:<name>, call:<dotted>,
tuple[bool, bool, list[str]]	cli:<runner>) preserving discovery order.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def detect_real_io(tree: ast.Module) -> tuple[bool, bool, list[str]]:
    """File-scope I/O detection from a parsed test module.

    Scans the module for three kinds of evidence that a test performs real I/O:
    fixtures listed in ``_IO_FIXTURES`` consumed by ``test_*`` functions,
    dotted calls listed in ``_IO_CALLS`` (e.g. ``subprocess.run``,
    ``pathlib.Path.write_text``), and CLI runner invocations such as
    ``CliRunner().invoke(...)``. Attribute-IO inside helpers is intentionally
    out of scope — see :func:`func_attr_io_transitive`.

    Returns:
        ``(has_io, has_subprocess, signals)``. ``has_io`` is ``True`` when any
        signal was found. ``has_subprocess`` is ``True`` when a
        ``subprocess.*`` call or a CLI runner was seen. ``signals`` is the
        list of evidence strings (``fixture:<name>``, ``call:<dotted>``,
        ``cli:<runner>``) preserving discovery order.
    """
    fixture_signals = _fixture_arg_signals(tree)
    sub_from_calls, call_signals = _io_call_signals(tree)
    sub_from_cli, cli_signals = _cli_runner_signals(tree)

    signals = fixture_signals + call_signals + cli_signals
    has_io = bool(signals)
    has_subprocess = sub_from_calls or sub_from_cli
    return has_io, has_subprocess, signals

extract_mock_targets(func)

Collect dotted mock/patch targets + mock-factory:Mock/... markers.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def extract_mock_targets(func: ast.FunctionDef) -> list[str]:
    """Collect dotted mock/patch targets + ``mock-factory:Mock/...`` markers."""
    out: list[str] = []
    for node in ast.walk(func):
        if not isinstance(node, ast.Call):
            continue
        _patch_call_targets(node, out)
        qual = _dotted_of(node.func) or ""
        leaf = qual.rsplit(".", 1)[-1] if qual else ""
        if leaf in _MOCK_FACTORIES:
            out.append(f"mock-factory:{leaf}")
    return out

fixture_does_io(fix_name, fixtures, visited, depth)

True if fixture body does real I/O.

Walks the fixture body for attr-IO, _IO_CALLS matches and transitive tmp_path deps; depth-bounded by _FIXTURE_DEPTH_LIMIT.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def fixture_does_io(
    fix_name: str,
    fixtures: dict[str, ast.FunctionDef],
    visited: set[str],
    depth: int,
) -> bool:
    """True if fixture body does real I/O.

    Walks the fixture body for attr-IO, ``_IO_CALLS`` matches and
    transitive ``tmp_path`` deps; depth-bounded by ``_FIXTURE_DEPTH_LIMIT``.
    """
    if fix_name in visited or depth >= _FIXTURE_DEPTH_LIMIT:
        return False
    visited.add(fix_name)
    fdef = fixtures.get(fix_name)
    if fdef is None:
        return False
    return _fixture_has_direct_io(fdef) or _fixture_calls_io_fixture(
        fdef, fix_name, fixtures, visited, depth
    )

func_attr_io_transitive(func, helpers, max_depth=2)

Attr-IO signals over the function subtree + transitively reachable helpers.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def func_attr_io_transitive(
    func: ast.FunctionDef,
    helpers: dict[str, ast.FunctionDef],
    max_depth: int = 2,
) -> list[str]:
    """Attr-IO signals over the function subtree + transitively reachable helpers."""
    visited: set[str] = set()
    sigs: list[str] = list(_attr_signals_in_node(func))

    frontier = _names_called_in(func) & set(helpers.keys())
    depth = 1
    while frontier and depth <= max_depth:
        next_frontier: set[str] = set()
        for name in frontier:
            if name in visited:
                continue
            visited.add(name)
            sigs.extend(_attr_signals_in_node(helpers[name]))
            next_frontier |= _names_called_in(helpers[name]) & set(helpers.keys())
        frontier = next_frontier - visited
        depth += 1
    return sigs

get_init_all(pkg_root)

Read __all__ from the first src/<pkg>/__init__.py that defines it.

Parameters:

Name	Type	Description	Default
pkg_root	Path	Repository root expected to follow the src/ layout.	required

Returns:

Type	Description
set[str] | None	Set of exported names, or None when no top-level package
set[str] | None	declares __all__ (or src is missing).

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def get_init_all(pkg_root: Path) -> set[str] | None:
    """Read ``__all__`` from the first ``src/<pkg>/__init__.py`` that defines it.

    Args:
        pkg_root: Repository root expected to follow the ``src/`` layout.

    Returns:
        Set of exported names, or ``None`` when no top-level package
        declares ``__all__`` (or ``src`` is missing).
    """
    src_dir = pkg_root / "src"
    if not src_dir.exists():
        return None
    for pkg_dir in src_dir.iterdir():
        if not pkg_dir.is_dir() or pkg_dir.name.startswith("."):
            continue
        init = pkg_dir / "__init__.py"
        if not init.exists():
            continue
        tree = _parse_cached(init)
        if tree is None:
            continue
        found = _extract_all_from_tree(tree)
        if found is not None:
            return found
    return None

get_module_all(pkg_root, dotted)

Resolve __all__ for dotted module within the src tree.

Parameters:

Name	Type	Description	Default
pkg_root	Path	Repository root expected to follow the src/ layout.	required
dotted	str	Dotted module path relative to src (e.g. pkg.sub).	required

Returns:

Type	Description
set[str] | None	Set of exported names, or None if the module is missing or does
set[str] | None	not declare __all__.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def get_module_all(pkg_root: Path, dotted: str) -> set[str] | None:
    """Resolve ``__all__`` for ``dotted`` module within the ``src`` tree.

    Args:
        pkg_root: Repository root expected to follow the ``src/`` layout.
        dotted: Dotted module path relative to ``src`` (e.g. ``pkg.sub``).

    Returns:
        Set of exported names, or ``None`` if the module is missing or does
        not declare ``__all__``.
    """
    src_dir = pkg_root / "src"
    rel = dotted.replace(".", "/")
    candidates = [src_dir / (rel + ".py"), src_dir / rel / "__init__.py"]
    for cand in candidates:
        if not cand.exists():
            continue
        tree = _parse_cached(cand)
        if tree is None:
            continue
        found = _extract_all_from_tree(tree)
        if found is not None:
            return found
    return None

get_pkg_prefixes(pkg_root)

Return top-level package directory names under <pkg_root>/src.

Parameters:

Name	Type	Description	Default
pkg_root	Path	Repository root expected to follow the src/ layout.	required

Returns:

Type	Description
set[str]	Set of directory names directly under src, excluding hidden
set[str]	entries. Empty when src is missing.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def get_pkg_prefixes(pkg_root: Path) -> set[str]:
    """Return top-level package directory names under ``<pkg_root>/src``.

    Args:
        pkg_root: Repository root expected to follow the ``src/`` layout.

    Returns:
        Set of directory names directly under ``src``, excluding hidden
        entries. Empty when ``src`` is missing.
    """
    src_dir = pkg_root / "src"
    if not src_dir.exists():
        return set()
    return {
        d.name for d in src_dir.iterdir() if d.is_dir() and not d.name.startswith(".")
    }

is_import_smoke_test(func)

Detect an import + weak-assert smoke test.

Returns True when the function body (docstrings excluded, ≤ 4 stmts) contains at least one import/from … import … statement and at least one weak assertion (assert name, assert x is not None, assert isinstance(...), assert callable(...), or self.assertIsNotNone(...)), with no statement stronger than these.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def is_import_smoke_test(func: ast.FunctionDef) -> bool:
    """Detect an import + weak-assert smoke test.

    Returns True when the function body (docstrings excluded, ≤ 4 stmts)
    contains at least one ``import``/``from … import …`` statement and at
    least one weak assertion (``assert name``, ``assert x is not None``,
    ``assert isinstance(...)``, ``assert callable(...)``, or
    ``self.assertIsNotNone(...)``), with no statement stronger than these.
    """
    body = [s for s in func.body if not _is_docstring_stmt(s)]
    if len(body) > _SMOKE_BODY_BUDGET:
        return False
    kinds = {_classify_smoke_stmt(stmt) for stmt in body}
    return "import" in kinds and "weak" in kinds and not (kinds & {"strong", "other"})

iter_test_files(pkg_root)

Yield (path, ast) for every tests/**/test_*.py.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def iter_test_files(pkg_root: Path) -> Iterator[tuple[Path, ast.Module | None]]:
    """Yield ``(path, ast)`` for every ``tests/**/test_*.py``."""
    tests_dir = pkg_root / "tests"
    if not tests_dir.exists():
        return
    for test_file in sorted(tests_dir.rglob("test_*.py")):
        yield test_file, _parse_cached(test_file)

target_matches_io(target)

Return True when target references a known I/O call or module.

Parameters:

Name	Type	Description	Default
target	str	Dotted call expression captured from the AST (e.g. pathlib.Path.read_text).	required

Returns:

Type	Description
bool	True when the leaf call name, the full dotted target, or any of
bool	its dotted segments matches the package's I/O catalog; False
bool	otherwise (including for empty input).

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def target_matches_io(target: str) -> bool:
    """Return ``True`` when ``target`` references a known I/O call or module.

    Args:
        target: Dotted call expression captured from the AST (e.g.
            ``pathlib.Path.read_text``).

    Returns:
        ``True`` when the leaf call name, the full dotted target, or any of
        its dotted segments matches the package's I/O catalog; ``False``
        otherwise (including for empty input).
    """
    if not target:
        return False
    if target in _IO_CALLS:
        return True
    for call in _IO_CALLS:
        if target.endswith("." + call):
            return True
    leaf_io_mods = {m.split(".")[-1] for m in _IO_MODULES}
    tokens = target.split(".")
    return any(tok in leaf_io_mods for tok in tokens)

test_is_in_lazy_import_context(func, tree_module, test_file)

Detect when the import itself is the system-under-test.

Returns True when the surrounding test file, class, or module docstring signals that the test exists to verify lazy/optional imports — in which case bare import statements inside the test body must not be flagged as smoke tests.

Source code in packages/axm-audit/src/axm_audit/core/rules/test_quality/_shared.py

def test_is_in_lazy_import_context(
    func: ast.FunctionDef,
    tree_module: ast.Module,
    test_file: Path,
) -> bool:
    """Detect when the import itself is the system-under-test.

    Returns True when the surrounding test file, class, or module
    docstring signals that the test exists to verify lazy/optional
    imports — in which case bare ``import`` statements inside the
    test body must not be flagged as smoke tests.
    """
    return (
        _has_lazy_filename(test_file)
        or _has_lazy_classname(func, tree_module)
        or _has_lazy_module_docstring(tree_module)
    )