orchestrata_tools.test_analyzer.matcher

Reference › orchestrata_tools › test_analyzer › matcher

TestMatcher is the deterministic engine that maps tests to the source methods they exercise, using a multi-strategy, import-aware approach. It is built over the tests' ASTCodebase and queried with a source ASTIndex. It is used by coverage, by the query evaluator for the tested filter and by the combined_generator. Tests are tests/test_orchestrata_tools_matcher.py and tests/test_orchestrata_tools_reviewer_query_indexer.py. Match types are the MatchStrategy values in test_models.

• _build_index collects test entries and the classes each test module imports. is_imported checks a class against a test module's imports.
• find_matches returns all TestMatch records. _compute_matches is the uncached full pass. Strategies run strongest first within each test: an exact class pairing (TestX with test_y to X.y), exact name matches, prefix and suffix variants, compound names split on and, substring and word-level matches, and finally the body call match from body_matcher.
• Strategies only propose candidates. Claims are resolved afterwards by confidence, so the result does not depend on the order tests are visited in. The code records that a first-come claim had made the reported strategy drift as tests were added.
• Substring strategies use precomputed substring indexes in place of scanning every source name.
• find_tested_methods returns the keys of all matched methods and find_untested_methods returns public methods without a match.

Matching depends only on the pair of objects, so results are cached by object identity in _MATCHES_CACHE, a lock-guarded, size-bounded dictionary, and each hit returns a fresh shallow copy. The comment records a measured 15.2 seconds for a first call and 9.4 seconds per repeat before the cache.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

TestMatcher

Line 46

Maps test methods to source methods using deterministic strategies.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_tools/test_analyzer/matcher.py

"""Deterministic test-to-source-method matching engine.

Uses a multi-strategy approach with import-aware resolution to
accurately map test methods to the source methods they test.
Strategies are ordered by confidence and short-circuit on first match.
Within a strategy, multiple source methods may be matched (the original
ast_explorer.py does not break early in most strategy loops).
"""

from __future__ import annotations

import re
import threading
from collections import OrderedDict, defaultdict
from typing import TYPE_CHECKING

from orchestrata_tools.models.test_models import MatchStrategy, TestEntry, TestMatch

if TYPE_CHECKING:
    from orchestrata_tools.models.ast_models import ASTCodebase, MethodInfo
    from orchestrata_tools.parser.indexer import ASTIndex
    from orchestrata_tools.test_analyzer.body_matcher import BodyMatcher


def _camel_to_snake(name: str) -> str:
    s1 = re.sub(r"(.)([A-Z][a-z]+)", r"\1_\2", name)
    return re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", s1).lower()


# Matching is a pure function of the (test_ast, source_index) object pair,
# but every API entry point (CoverageAnalyzer.analyze, find_untested_methods,
# QueryEvaluator's tested predicate, CombinedTestGenerator.generate_for_gaps)
# historically re-ran the full pass — measured 15.2s first call / 9.4s per
# repeat over this repo's ASTs, five times per consumer session. The cache is
# keyed on object IDENTITY (the value holds strong refs to both keys, so an
# id() can never be reused while its entry lives) and bounded to the last few
# pairs; entries store the canonical list and every hit returns a fresh
# shallow copy so no caller can mutate shared state. Lock-guarded for 3.14t.
_MATCHES_CACHE: OrderedDict[tuple[int, int], tuple[object, object, list[TestMatch]]] = (
    OrderedDict()
)
_MATCHES_CACHE_LOCK = threading.Lock()
_MATCHES_CACHE_MAX = 4


class TestMatcher:
    """Maps test methods to source methods using deterministic strategies.

    Strategies are tried in order of confidence. For each test, the first
    strategy that yields any matches is used and subsequent strategies are
    skipped. Within a strategy, multiple source methods may be matched.
    """

    __test__ = False  # Not a pytest test class

    def __init__(self, test_ast: ASTCodebase):
        self.test_ast = test_ast
        self.test_entries: list[TestEntry] = []
        self.imported_names_by_module: dict[str, set[str]] = defaultdict(set)
        self._body_matcher: BodyMatcher | None = None
        self._build_index()

    def _build_index(self) -> None:
        # test_calls_by_module[method_name] = list of test methods that call it
        self.test_calls: dict[str, list[tuple[str, str | None, list[str]]]] = (
            defaultdict(list)
        )

        for mod in self.test_ast.modules:
            mod_path = mod.path
            for imp in mod.imports:
                src_module = imp.module
                if src_module.startswith("orchestrata_ultimate"):
                    if imp.is_from:
                        for name in imp.names:
                            self.imported_names_by_module[mod_path].add(name)
                    else:
                        parts = src_module.split(".")
                        if parts:
                            self.imported_names_by_module[mod_path].add(parts[-1])
                        for part in parts:
                            if part and part[0].isupper():
                                self.imported_names_by_module[mod_path].add(part)

            for func in mod.functions:
                if func.name.startswith("test_"):
                    self.test_entries.append(
                        TestEntry(name=func.name, module=mod_path, class_name=None)
                    )
                    calls = getattr(func, "calls", [])
                    for called in calls:
                        self.test_calls[called].append((func.name, None, calls))

            for cls in mod.classes:
                for meth in cls.methods:
                    if meth.name.startswith("test_"):
                        self.test_entries.append(
                            TestEntry(
                                name=meth.name, module=mod_path, class_name=cls.name
                            )
                        )
                        for called in meth.calls:
                            self.test_calls[called].append(
                                (meth.name, cls.name, meth.calls)
                            )

    def is_imported(self, test_module: str, class_name: str) -> bool:
        if class_name in self.imported_names_by_module.get(test_module, set()):
            return True
        # Fallback to BodyMatcher's direct source parsing (catches imports
        # the serialized AST may have missed).
        if self._body_matcher is not None:
            return self._body_matcher.class_used_by_test(test_module, class_name)
        return False

    def find_matches(self, source_index: ASTIndex) -> list[TestMatch]:
        """Find all test-to-method matches with strategy details.

        Per test, only the highest-confidence strategy that yields matches
        is used. Within that strategy, all matching source methods are added.

        Results are memoized per (test_ast, source_index) identity pair —
        see the module cache above. Callers receive a fresh shallow copy.
        """
        key = (id(self.test_ast), id(source_index))
        with _MATCHES_CACHE_LOCK:
            entry = _MATCHES_CACHE.get(key)
            if (
                entry is not None
                and entry[0] is self.test_ast
                and entry[1] is source_index
            ):
                _MATCHES_CACHE.move_to_end(key)
                return list(entry[2])
        matches = self._compute_matches(source_index)
        with _MATCHES_CACHE_LOCK:
            _MATCHES_CACHE[key] = (self.test_ast, source_index, matches)
            _MATCHES_CACHE.move_to_end(key)
            while len(_MATCHES_CACHE) > _MATCHES_CACHE_MAX:
                _MATCHES_CACHE.popitem(last=False)
        return list(matches)

    def _compute_matches(self, source_index: ASTIndex) -> list[TestMatch]:
        """The uncached full matching pass (see ``find_matches``)."""
        matches: list[TestMatch] = []
        tested_keys: set[str] = set()
        # Every (test, method, strategy) the strategy loop PROPOSES, before any
        # of them claims anything. Claiming is resolved after the loop, by
        # confidence — see the resolution pass below for why.
        candidates: list[tuple[int, int, TestMatch]] = []

        source_by_name: dict[str, list[MethodInfo]] = defaultdict(list)
        source_by_class: dict[str, list[MethodInfo]] = defaultdict(list)
        for m in source_index.codebase.all_methods:
            source_by_name[m.name].append(m)
            source_by_class[m.class_name].append(m)

        # ---- Substring indexes for Strategies 7/8 (order-preserving) ----
        # Both strategies scanned every source name per unmatched test
        # (~12.4K tests x 3,471 names = ~96% of strategy-loop time). These
        # indexes serve the identical lookups; `name_order` preserves the
        # source_by_name iteration order so results are byte-identical.
        long_names = [n for n in source_by_name if len(n) > 3]
        long_name_set = frozenset(long_names)
        name_order = {n: i for i, n in enumerate(long_names)}
        max_name_len = max(map(len, long_names), default=0)
        # substring -> long names containing it (insertion order preserved)
        contains_index: dict[str, list[str]] = {}
        for n in long_names:
            ln = len(n)
            seen_subs: set[str] = set()
            for i in range(ln):
                for j in range(i + 1, ln + 1):
                    sub = n[i:j]
                    if sub not in seen_subs:
                        seen_subs.add(sub)
                        contains_index.setdefault(sub, []).append(n)

        def _resolve_target_class(
            test_class_name: str | None, test_module: str
        ) -> tuple[str | None, bool]:
            if not test_class_name or not test_class_name.startswith("Test"):
                return None, False
            candidate = test_class_name[4:]
            if candidate in source_by_class:
                return candidate, True
            imported = self.imported_names_by_module.get(test_module, set())
            best_match = None
            best_len = 0
            for cls_name in imported:
                if cls_name in source_by_class and candidate.startswith(cls_name):
                    if len(cls_name) > best_len:
                        best_len = len(cls_name)
                        best_match = cls_name
            if best_match:
                return best_match, True
            return None, False

        def _add(
            methods: list[MethodInfo], strategy: MatchStrategy, confidence: int
        ) -> None:
            """PROPOSE a match; it does not claim the method.

            This used to claim on the spot, first-come. The strategy loop is
            ordered strongest-first WITHIN one test, but the outer loop is over
            TESTS, so across tests the winner was whichever test happened to be
            iterated first — regardless of how weak its strategy was. Measured
            2026-08-21: of 599 pairs following the ``TestX::test_y -> X.y``
            convention, 355 had been claimed by a DIFFERENT test, e.g.
            ``BaseAgent.set_postman`` (an exact 1:1 name match, confidence 10)
            lost to a module-level ``test_base_agent_set_postman_stores_
            reference`` matching only by IMPORT_WORD, confidence 4.

            That made the reported strategy a function of iteration order, so
            ``class_exact`` drifted downward whenever tests were ADDED and had
            to be re-baselined three times (300 -> 240 -> 225) without any
            coverage ever being lost.
            """
            for m in methods:
                candidates.append(
                    (
                        confidence,
                        len(candidates),
                        TestMatch(
                            test_entry=test,
                            method=m,
                            strategy=strategy,
                            confidence=confidence,
                        ),
                    )
                )

        for test in self.test_entries:
            target_name = test.target_name
            imported = self.imported_names_by_module.get(test.module, set())
            target_class, class_resolved = _resolve_target_class(
                test.class_name, test.module
            )
            target_methods = (
                source_by_class.get(target_class, []) if target_class else []
            )
            matched = False

            # Strategy 1: Class-qualified exact match (highest confidence)
            if target_class:
                found = [m for m in target_methods if m.name == target_name]
                if found:
                    _add(found, MatchStrategy.CLASS_EXACT, 10)
                    matched = True

            # Strategy 2: Import-filtered global exact match
            if not matched and target_name in source_by_name:
                filtered = [
                    m for m in source_by_name[target_name] if m.class_name in imported
                ]
                if filtered:
                    _add(filtered, MatchStrategy.GLOBAL_EXACT, 9)
                    matched = True
                elif not test.class_name:
                    _add(source_by_name[target_name], MatchStrategy.GLOBAL_EXACT, 7)
                    matched = True

            # Strategy 3: Suffix/prefix stripping within class
            if not matched and target_class:
                found = []
                for m in target_methods:
                    if (
                        target_name.endswith(f"_{m.name}")
                        or target_name.startswith(f"{m.name}_")
                        or m.name.endswith(f"_{target_name}")
                        or m.name.startswith(f"{target_name}_")
                    ):
                        found.append(m)
                if found:
                    _add(found, MatchStrategy.SUFFIX_PREFIX, 8)
                    matched = True

            # Strategy 4: Compound exact match
            if not matched:
                for separator in ("_and_", "_or_", "_then_", "_with_"):
                    if separator in target_name:
                        parts = target_name.split(separator)
                        found = []
                        for part in parts:
                            if part not in source_by_name:
                                continue
                            for m in source_by_name[part]:
                                if target_class and m.class_name != target_class:
                                    continue
                                if not target_class and m.class_name not in imported:
                                    continue
                                found.append(m)
                        if found:
                            _add(found, MatchStrategy.COMPOUND_EXACT, 8)
                            matched = True
                            break

            # Strategy 4b: Compound substring within class
            if not matched and target_class:
                for separator in ("_and_", "_or_", "_then_", "_with_"):
                    if separator in target_name:
                        parts = target_name.split(separator)
                        found = []
                        for part in parts:
                            if len(part) <= 2:
                                continue
                            for m in target_methods:
                                if part in m.name or m.name in part:
                                    found.append(m)
                        if found:
                            _add(found, MatchStrategy.COMPOUND_SUBSTRING, 6)
                            matched = True
                            break

            # Strategy 5: Class-context substring containment
            if not matched and target_class:
                found = []
                for m in target_methods:
                    if target_name in m.name or m.name in target_name:
                        found.append(m)
                if found:
                    _add(found, MatchStrategy.SUBSTRING, 5)
                    matched = True

            # Strategy 5b: Class-context word-level matching
            if not matched and target_class:
                test_words = set(target_name.split("_"))
                found = []
                for m in target_methods:
                    if test_words & set(m.name.split("_")):
                        found.append(m)
                if found:
                    _add(found, MatchStrategy.WORD_LEVEL, 5)
                    matched = True

            # Strategy 5c: Import-filtered word-level for unresolved/top-level
            if not matched and (not test.class_name or not class_resolved):
                test_words = set(target_name.split("_"))
                found = []
                for cls_name in imported:
                    if cls_name not in source_by_class:
                        continue
                    for m in source_by_class[cls_name]:
                        if test_words & set(m.name.split("_")):
                            found.append(m)
                if found:
                    _add(found, MatchStrategy.IMPORT_WORD, 4)
                    matched = True

            # Strategy 6: Top-level prefix-stripped match
            if not matched and not test.class_name:
                for cls_name in imported:
                    prefix = _camel_to_snake(cls_name)
                    if prefix in target_name:
                        remainder = target_name.replace(prefix, "").strip("_")
                        if remainder in source_by_name:
                            found = [
                                m
                                for m in source_by_name[remainder]
                                if m.class_name == cls_name
                            ]
                            if found:
                                _add(found, MatchStrategy.PREFIX_STRIPPED, 6)
                                matched = True
                                break

            # Strategy 7: Import-filtered substring match
            if not matched:
                tl = len(target_name)
                cap = min(max_name_len, tl)
                cand = {
                    target_name[i : i + k]
                    for k in range(4, cap + 1)
                    for i in range(tl - k + 1)
                }
                found = []
                for src_name in sorted(
                    cand & long_name_set, key=name_order.__getitem__
                ):
                    for m in source_by_name[src_name]:
                        if m.class_name in imported:
                            found.append(m)
                if found:
                    _add(found, MatchStrategy.IMPORT_SUBSTRING, 3)
                    matched = True

            # Strategy 8: Weak global substring match (last resort)
            if not matched:
                if target_name:
                    names8 = contains_index.get(target_name, [])
                else:  # "" is a substring of every name (original behavior)
                    names8 = long_names
                found = [m for src_name in names8 for m in source_by_name[src_name]]
                if found:
                    _add(found, MatchStrategy.GLOBAL_SUBSTRING, 1)

        # ------------------------------------------------------------------
        # Claim resolution: the strongest proposal wins, across ALL tests
        # ------------------------------------------------------------------
        # Each method goes to its HIGHEST-confidence proposal; ties keep the
        # earliest one, which is exactly the previous behaviour within a
        # confidence tier. The claimed SET is identical either way — it is the
        # union of every proposal, and that union does not depend on ordering —
        # so find_untested_methods/coverage is INVARIANT under this pass. Only
        # which test claims a method, and under which strategy label, changes.
        # Emission is re-sorted back into proposal order so downstream
        # consumers see the same sequence shape they always have.
        winners: dict[str, tuple[int, int, TestMatch]] = {}
        for confidence, seq, match in candidates:
            key = match.method.fully_qualified_name
            current = winners.get(key)
            if current is None or confidence > current[0]:
                winners[key] = (confidence, seq, match)
        for winner in sorted(winners.values(), key=lambda w: w[1]):
            match = winner[2]
            tested_keys.add(match.method.fully_qualified_name)
            matches.append(match)

        # ------------------------------------------------------------------
        # Body-based matching pass: parse test source files directly
        # ------------------------------------------------------------------
        if self._body_matcher is None:
            from orchestrata_tools.test_analyzer.body_matcher import BodyMatcher

            self._body_matcher = BodyMatcher(self.test_ast, source_index.codebase)

        for source_method in source_index.codebase.all_methods:
            key = source_method.fully_qualified_name
            if key in tested_keys:
                continue
            tests = self._body_matcher.find_tests_for_method(source_method)
            for test_module, test_name, test_cls in tests:
                if self.is_imported(test_module, source_method.class_name):
                    tested_keys.add(key)
                    matches.append(
                        TestMatch(
                            test_entry=TestEntry(
                                name=test_name, module=test_module, class_name=test_cls
                            ),
                            method=source_method,
                            strategy=MatchStrategy.BODY_CALL,
                            confidence=7,
                        )
                    )
                    break

        return matches

    def _method_key(self, method: MethodInfo) -> str:
        return method.fully_qualified_name

    def find_tested_methods(self, source_index: ASTIndex) -> set[str]:
        """Return set of ALL matched method keys (public and private).

        Uses the same matching as the original ast_explorer.py: all matched
        methods are tracked, and filtering happens at reporting time.
        """
        matches = self.find_matches(source_index)
        return {m.method.fully_qualified_name for m in matches}

    def find_untested_methods(self, source_index: ASTIndex) -> list[MethodInfo]:
        """Return public methods without corresponding tests."""
        tested = self.find_tested_methods(source_index)
        return [
            m
            for m in source_index.public_methods()
            if m.fully_qualified_name not in tested
        ]