Duplicate tests

duplicate_tests

Duplicate-tests rule — cluster likely-duplicate test functions.

Three clustering signals (S1/S2/S3) + six rescue anti-signals (P1-P6) ported from the detect_duplicates.py prototype.

DuplicateTestsCheckResult

Bases: CheckResult

:class:CheckResult with cluster metadata.

metadata keys:

• clusters — list of cluster payloads, each with cluster_hash, a members list of test entries, and an optional acknowledged flag (True when the hash is present in [[tool.axm-audit.duplicate_tests.acknowledged]]). The cluster shape exposes members only; the legacy tests alias was removed in axm-1728 — consumers reading cluster["tests"] get a KeyError by design.
• bucket_counts — CLUSTERED / AMBIGUOUS / UNIQUE test counts.
• stale_acknowledged — acknowledgement entries ({hash, reason}) whose hash no longer matches any vivant cluster. Informational only: does not affect passed / score / severity.
• config_error — present when pyproject.toml is malformed or the [tool.axm-audit.duplicate_tests] schema is invalid. The audit falls back to "no acknowledgements" rather than crashing.

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

class DuplicateTestsCheckResult(CheckResult):  # type: ignore[explicit-any]  # pydantic synthesizes __init__(**data: Any)
    """:class:`CheckResult` with cluster metadata.

    ``metadata`` keys:

    * ``clusters`` — list of cluster payloads, each with ``cluster_hash``,
      a ``members`` list of test entries, and an optional ``acknowledged``
      flag (``True`` when the hash is present in
      ``[[tool.axm-audit.duplicate_tests.acknowledged]]``). The cluster
      shape exposes ``members`` only; the legacy ``tests`` alias was
      removed in axm-1728 — consumers reading ``cluster["tests"]`` get
      a ``KeyError`` by design.
    * ``bucket_counts`` — CLUSTERED / AMBIGUOUS / UNIQUE test counts.
    * ``stale_acknowledged`` — acknowledgement entries (``{hash, reason}``)
      whose hash no longer matches any vivant cluster. Informational only:
      does not affect ``passed`` / ``score`` / ``severity``.
    * ``config_error`` — present when ``pyproject.toml`` is malformed or
      the ``[tool.axm-audit.duplicate_tests]`` schema is invalid. The audit
      falls back to "no acknowledgements" rather than crashing.
    """

    metadata: dict[str, object] = Field(default_factory=dict)

    model_config = {"extra": "forbid"}

DuplicateTestsRule dataclass

Bases: ProjectRule

Cluster likely-duplicate test functions via structural signals.

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

@register_rule("test_quality")
@dataclass
class DuplicateTestsRule(ProjectRule):
    """Cluster likely-duplicate test functions via structural signals."""

    ast_similarity_threshold: float = 0.8

    @property
    def rule_id(self) -> str:
        """Stable identifier for this rule."""
        return "TEST_QUALITY_DUPLICATE_TESTS"

    def check(self, project_path: Path) -> DuplicateTestsCheckResult:
        """Cluster duplicate tests in ``project_path`` and return verdicts."""
        tests = _collect_tests(project_path)
        if not tests:
            return self._empty_result()

        config = _load_duplicate_tests_config(project_path)
        clusters = _cluster(tests, self.ast_similarity_threshold)
        bucket_counts = _bucket_counts(tests, clusters)
        slim = _slim_clusters(clusters)
        _mark_acknowledged(slim, config.acknowledged)
        stale_acknowledged = _stale_acknowledged(slim, config.acknowledged)

        n_clustered_pairs = _count_clustered_pairs(slim)
        passed = n_clustered_pairs == 0
        score = max(0, 100 - n_clustered_pairs * _SCORE_PENALTY)
        message = _build_message(clusters, n_clustered_pairs)
        text = _build_text(slim, stale_acknowledged, passed)
        fix_hint = None if passed else _FIX_HINT
        metadata = _build_metadata(
            slim, bucket_counts, stale_acknowledged, config.error
        )
        return DuplicateTestsCheckResult(
            rule_id=self.rule_id,
            passed=passed,
            message=message,
            severity=Severity.WARNING,
            score=score,
            metadata=metadata,
            text=text,
            fix_hint=fix_hint,
        )

    def _empty_result(self) -> DuplicateTestsCheckResult:
        return DuplicateTestsCheckResult(
            rule_id=self.rule_id,
            passed=True,
            message="no tests found",
            severity=Severity.INFO,
            score=100,
            metadata={
                "clusters": [],
                "bucket_counts": {"CLUSTERED": 0, "AMBIGUOUS": 0, "UNIQUE": 0},
            },
        )

rule_id property

Stable identifier for this rule.

check(project_path)

Cluster duplicate tests in project_path and return verdicts.

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

def check(self, project_path: Path) -> DuplicateTestsCheckResult:
    """Cluster duplicate tests in ``project_path`` and return verdicts."""
    tests = _collect_tests(project_path)
    if not tests:
        return self._empty_result()

    config = _load_duplicate_tests_config(project_path)
    clusters = _cluster(tests, self.ast_similarity_threshold)
    bucket_counts = _bucket_counts(tests, clusters)
    slim = _slim_clusters(clusters)
    _mark_acknowledged(slim, config.acknowledged)
    stale_acknowledged = _stale_acknowledged(slim, config.acknowledged)

    n_clustered_pairs = _count_clustered_pairs(slim)
    passed = n_clustered_pairs == 0
    score = max(0, 100 - n_clustered_pairs * _SCORE_PENALTY)
    message = _build_message(clusters, n_clustered_pairs)
    text = _build_text(slim, stale_acknowledged, passed)
    fix_hint = None if passed else _FIX_HINT
    metadata = _build_metadata(
        slim, bucket_counts, stale_acknowledged, config.error
    )
    return DuplicateTestsCheckResult(
        rule_id=self.rule_id,
        passed=passed,
        message=message,
        severity=Severity.WARNING,
        score=score,
        metadata=metadata,
        text=text,
        fix_hint=fix_hint,
    )

collect_assert_call_sigs(node)

Return _call_sig set for every :class:ast.Call inside an assert.

Tests written as assert helper(args) == expected would otherwise have an empty call_sig (the taint-propagation pass only follows assigns). Including these calls lets S1/S2 group such tests by SUT.

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

def collect_assert_call_sigs(node: ast.FunctionDef) -> set[str]:
    """Return ``_call_sig`` set for every :class:`ast.Call` inside an assert.

    Tests written as ``assert helper(args) == expected`` would otherwise have
    an empty ``call_sig`` (the taint-propagation pass only follows assigns).
    Including these calls lets S1/S2 group such tests by SUT.
    """
    out: set[str] = set()
    for parent in ast.walk(node):
        if not isinstance(parent, ast.Assert):
            continue
        for child in ast.walk(parent):
            if isinstance(child, ast.Call):
                sig = _call_sig(child)
                if sig is not None:
                    out.add(sig)
    return out

merge_clusters(clusters)

Union-find merge; ambiguous sub-clusters dominate the merged signal.

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

def merge_clusters(clusters: list[_Cluster]) -> list[_Cluster]:
    """Union-find merge; ambiguous sub-clusters dominate the merged signal."""
    if not clusters:
        return []

    groups = _build_union_find(clusters)
    merged: list[_Cluster] = []
    for indices in groups.values():
        tests_by_key, reasons, signals, max_sim = _aggregate_group(clusters, indices)
        if len(tests_by_key) < _MIN_PAIR:
            continue
        unique_reasons = list(dict.fromkeys(reasons))
        merged.append(
            {
                "signal": _pick_merged_signal(signals),
                "reason": " + ".join(unique_reasons[:3]),
                "similarity": max_sim,
                "members": list(tests_by_key.values()),
            }
        )
    return sorted(merged, key=lambda c: -len(c["members"]))

p10_rescues(tests)

Locality rescue: demote when all pairs are far apart with no strong signal.

Returns True iff every pair in tests has line distance greater than :data:_P10_LINE_DIST AND no pair carries a very strong direct signal (:func:_pair_is_very_strong). Tests that match this pattern are almost always coincidental matches the user does not want flagged firmly.

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

def p10_rescues(tests: list[_TestFunc]) -> bool:
    """Locality rescue: demote when all pairs are far apart with no strong signal.

    Returns True iff every pair in ``tests`` has line distance greater than
    :data:`_P10_LINE_DIST` AND no pair carries a very strong direct signal
    (:func:`_pair_is_very_strong`). Tests that match this pattern are almost
    always coincidental matches the user does not want flagged firmly.
    """
    if len(tests) < _MIN_PAIR:
        return False
    for a, b in combinations(tests, 2):
        if abs(a.line - b.line) <= _P10_LINE_DIST:
            return False
        if _pair_is_very_strong(a, b):
            return False
    return True

render_clusters_text(clusters, stale_acknowledged=None)

Render top-N clusters (signal + ambiguous) as a compact bullet list.

When stale_acknowledged is provided, append one warning line per stale entry of the form ⚠ stale acknowledged cluster: <hash> (<reason>). Stale warnings are suffix-only and never affect passed / score / severity.

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

def render_clusters_text(
    clusters: list[_Cluster],
    stale_acknowledged: list[dict[str, str]] | None = None,
) -> str:
    """Render top-N clusters (signal + ambiguous) as a compact bullet list.

    When ``stale_acknowledged`` is provided, append one warning line per
    stale entry of the form
    ``⚠ stale acknowledged cluster: <hash> (<reason>)``. Stale warnings
    are suffix-only and never affect ``passed`` / ``score`` / ``severity``.
    """
    ordered = sorted(
        clusters,
        key=lambda c: (-len(c.get("members", [])), c.get("signal", "")),
    )
    lines: list[str] = []
    for cluster in ordered[:_MAX_TEXT_CLUSTERS]:
        members = cluster.get("members", [])
        suffix = "…" if len(members) > _MAX_MEMBERS_INLINE else ""
        label = _render_cluster_members(members)
        lines.append(
            f"• cluster[{cluster.get('signal', '')}] "
            f"{len(members)} tests: {label}{suffix}"
        )
    if len(ordered) > _MAX_TEXT_CLUSTERS:
        lines.append(f"(+{len(ordered) - _MAX_TEXT_CLUSTERS} more clusters)")
    for entry in stale_acknowledged or []:
        lines.append(
            f"⚠ stale acknowledged cluster: {entry.get('hash', '')} "
            f"({entry.get('reason', '')})"
        )
    return "\n".join(lines)