Extract helpers

extract_helpers

Post-pipeline polish: collapse duplicated test helpers.

After SPLIT and MERGE leave anvil's shared_helpers="duplicate" copies of helpers (gold_project, _make_result, …) in every post-move file, this module collapses them into a single tests/<tier>/_helpers.py (for pure helpers) or the tier's conftest.py (for @pytest.fixture).

Three layers:

• extract_shared_helpers — top-level entry; iterates _once until fixed-point, deduplicating ambiguous-fixture warnings across iterations.
• extract_shared_helpers_once / _in_tier — single-pass extraction across all canonical tiers.
• load_or_create_* / strip_def_* — file-level mutation helpers, libcst-based.

extract_shared_helpers(project_path)

Iterate extract_shared_helpers_once until fixed-point.

A single pass cannot catch every duplicate: promoting helper A can expose helper B as duplicate (e.g. A's body referenced B locally, so B looked non-shared until A moved out). Loop until no further extraction happens. Capped at _EXTRACT_MAX_ITERS to fail loud on a buggy fixed-point.

ambiguous fixture messages are re-emitted on every iteration (the same fixtures stay ambiguous forever) — collapse them so the operator sees each one exactly once.

Source code in packages/axm-audit/src/axm_audit/core/fix/extract_helpers.py

def extract_shared_helpers(project_path: Path) -> list[str]:
    """Iterate ``extract_shared_helpers_once`` until fixed-point.

    A single pass cannot catch every duplicate: promoting helper A can
    expose helper B as duplicate (e.g. A's body referenced B locally,
    so B looked non-shared until A moved out). Loop until no further
    extraction happens. Capped at ``_EXTRACT_MAX_ITERS`` to fail loud
    on a buggy fixed-point.

    ``ambiguous fixture`` messages are re-emitted on every iteration
    (the same fixtures stay ambiguous forever) — collapse them so the
    operator sees each one exactly once.
    """
    all_msgs: list[str] = []
    seen_ambiguous: set[str] = set()
    for _ in range(_EXTRACT_MAX_ITERS):
        msgs = extract_shared_helpers_once(project_path)
        progress = [m for m in msgs if "ambiguous fixture" not in m]
        deduped: list[str] = list(progress)
        for m in msgs:
            if "ambiguous fixture" in m and m not in seen_ambiguous:
                seen_ambiguous.add(m)
                deduped.append(m)
        all_msgs.extend(deduped)
        if not progress:
            break
    return all_msgs

extract_shared_helpers_in_tier(project_path, tier_dir)

Process a single tier. Splitting per-tier keeps imports local.

Source code in packages/axm-audit/src/axm_audit/core/fix/extract_helpers.py

def extract_shared_helpers_in_tier(project_path: Path, tier_dir: Path) -> list[str]:
    """Process a single tier. Splitting per-tier keeps imports local."""
    scan = _scan_tier(tier_dir)
    index = _index_tier_scan(scan)
    part = _partition_duplicates(index, project_path)
    part.skipped_names |= scan.location_skipped
    _resolve_cascading_skips(part, index)
    if not part.duplicates and not part.skip_msgs:
        return []
    if not part.duplicates:
        return part.skip_msgs
    targets = _build_emit_targets(project_path, tier_dir)
    if targets is None:
        return []
    msgs = _emit_duplicates(targets, part.duplicates, scan.per_file, project_path)
    msgs.extend(part.skip_msgs)
    return msgs

extract_shared_helpers_once(project_path)

Promote helpers duplicated across a tier into tests/<tier>/_helpers.py.

Source code in packages/axm-audit/src/axm_audit/core/fix/extract_helpers.py

def extract_shared_helpers_once(project_path: Path) -> list[str]:
    """Promote helpers duplicated across a tier into ``tests/<tier>/_helpers.py``."""
    msgs: list[str] = []
    tests_root = project_path / "tests"
    if not tests_root.is_dir():
        return msgs
    for tier in ("integration", "e2e", "unit"):
        tier_dir = tests_root / tier
        if not tier_dir.is_dir():
            continue
        msgs.extend(extract_shared_helpers_in_tier(project_path, tier_dir))
    return msgs

strip_def_and_inject_import(file, name, helpers_module, project_path)

Remove the top-level def of name from file and import it instead.

Source code in packages/axm-audit/src/axm_audit/core/fix/extract_helpers.py

def strip_def_and_inject_import(
    file: Path, name: str, helpers_module: str, project_path: Path
) -> None:
    """Remove the top-level def of ``name`` from *file* and import it instead."""
    module = cst_load(file)
    if module is None:
        return
    new_body: list[cst.BaseStatement] = []
    stripped = False
    for stmt in module.body:
        if isinstance(stmt, cst.FunctionDef | cst.ClassDef) and stmt.name.value == name:
            stripped = True
            continue
        if _is_top_level_assign_to(stmt, name):
            stripped = True
            continue
        new_body.append(stmt)
    if not stripped:
        return
    import_stmt = cst.parse_statement(f"from {helpers_module} import {name}")
    assert isinstance(import_stmt, cst.SimpleStatementLine)
    new_body.insert(_compute_import_insert_index(new_body), import_stmt)
    new_module = module.with_changes(body=new_body)
    new_module = _dedupe_imports_cst(new_module)
    cst_save(file, new_module)

strip_def_only(file, name)

Remove the top-level def of name from file without injecting an import.

Used for fixtures whose new home (conftest.py) is auto-discovered by pytest — no import would be valid syntactically (it would shadow the injected fixture parameter) and none is needed.

Source code in packages/axm-audit/src/axm_audit/core/fix/extract_helpers.py

def strip_def_only(file: Path, name: str) -> None:
    """Remove the top-level def of *name* from *file* without injecting an import.

    Used for fixtures whose new home (``conftest.py``) is auto-discovered
    by pytest — no import would be valid syntactically (it would shadow
    the injected fixture parameter) and none is needed.
    """
    module = cst_load(file)
    if module is None:
        return
    new_body: list[cst.BaseStatement] = []
    stripped = False
    for stmt in module.body:
        if isinstance(stmt, cst.FunctionDef | cst.ClassDef) and stmt.name.value == name:
            stripped = True
            continue
        new_body.append(stmt)
    if stripped:
        cst_save(file, module.with_changes(body=new_body))