orchestrata_tools.test_generator.testgen.ranking

Reference › orchestrata_tools › test_generator › testgen › ranking

This module ranks functions by testing priority, so the generators know what to cover first. It is imported by generator and structural, and covered by tests/test_orchestrata_tools_testgen_ranking_coverage.py.

rank_functions takes a CodebaseAST and returns RankedEntry records, each with the qualified name, module, FunctionInfo, fan-in, reachable count, score, and method and class information. Fan-in counts distinct callers; reachable counts other functions transitively called, bounded by max_reach_depth (default 3). Both are computed from each function's unresolved call strings matched by bare callee name, which the module describes as a best-effort proxy and not a resolved call graph. Public-only (default) and dunder exclusion, a minimum fan-in and a top-N cut are options. The docstring lists fan-in, reachability, public visibility and signature availability as the ranking inputs. summarise renders a priority table as text.

At a glance

Class diagram

Dependency diagram

API

Classes

RankedEntry (dataclass)

Decorators: dataclass · Line 36

One function/method with its testability score.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_tools/test_generator/testgen/ranking.py

"""ranking.py — Identify the highest-priority functions to test first.

Strategy:
  1. fan_in     — how many distinct callers reach this scope
  2. reachable  — how many other scopes this function transitively calls
                  (proxy for "how much of the system it exercises")
  3. public     — private symbols are deprioritised
  4. has_signature — functions with typed params/return are easier to test

The result is a ranked list of RankedEntry objects that the test
generators can consume.

Restored 2026-08-19 from ``scripts/ast_serializer/testgen`` (deleted
2026-07-09, commit 6a186c9d) and re-pointed at the current AST layer.
The original ranked over ``ast_serializer``'s own ``ProjectRecord``/
``Symbol`` plus a dedicated ``CallGraphAnalyzer`` (both deleted alongside
their parser, with no replacement) — this version computes fan_in/
reachable directly from ``orchestrata_ast.models.FunctionInfo.calls``
(unresolved call-expression strings the current AST layer already
extracts), matched by bare callee name. That's a best-effort proxy, same
as the original ("proxy for how much of the system it exercises") — good
enough to order candidates for branch-test generation, not a claim of a
fully resolved call graph.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from orchestrata_ast.models import CodebaseAST, FunctionInfo


@dataclass
class RankedEntry:
    """One function/method with its testability score."""

    qualified_name: str  # e.g. "myapp.utils.MyClass.process"
    module: str  # e.g. "myapp.utils"
    symbol: FunctionInfo
    fan_in: int  # distinct callers (best-effort, name-matched)
    reachable: int  # transitively reachable scopes (bounded BFS)
    score: float  # composite priority score
    is_method: bool
    class_name: str | None  # set when is_method


def _bare_target(call_expr: str) -> str:
    """Last dotted segment of a call expression, e.g. "self.foo" → "foo"."""
    return call_expr.rsplit(".", 1)[-1]


def rank_functions(
    codebase: CodebaseAST,
    min_fan_in: int = 0,
    public_only: bool = True,
    exclude_dunder: bool = True,
    top_n: int | None = None,
    max_reach_depth: int = 3,
) -> list[RankedEntry]:
    """
    Analyse *codebase* and return functions ranked by testing priority.

    Parameters
    ----------
    codebase        : a serialized CodebaseAST (orchestrata_ast.models)
    min_fan_in      : exclude functions with fewer callers than this
    public_only     : skip names starting with _
    exclude_dunder  : skip __init__, __repr__, etc.
    top_n           : return at most this many results
    max_reach_depth : bound on the transitive-reachability BFS
    """
    all_functions: list[FunctionInfo] = []
    class_of: dict[str, str] = {}
    module_of: dict[str, str] = {}
    for mod in codebase.modules:
        all_functions.extend(mod.functions)
        for f in mod.functions:
            module_of[f.qualified_name] = mod.path
        for cls in mod.classes:
            all_functions.extend(cls.methods)
            for m in cls.methods:
                class_of[m.qualified_name] = cls.name
                module_of[m.qualified_name] = mod.path

    by_bare_name: dict[str, list[FunctionInfo]] = {}
    for f in all_functions:
        by_bare_name.setdefault(f.name, []).append(f)

    fan_in_map: dict[str, int] = {f.qualified_name: 0 for f in all_functions}
    for caller in all_functions:
        seen_targets: set[str] = set()
        for call in caller.calls:
            for callee in by_bare_name.get(_bare_target(call), ()):
                if callee.qualified_name == caller.qualified_name:
                    continue
                if callee.qualified_name in seen_targets:
                    continue
                seen_targets.add(callee.qualified_name)
                fan_in_map[callee.qualified_name] = (
                    fan_in_map.get(callee.qualified_name, 0) + 1
                )

    def _reachable(start: FunctionInfo) -> int:
        seen = {start.qualified_name}
        frontier = [start]
        for _ in range(max_reach_depth):
            nxt: list[FunctionInfo] = []
            for f in frontier:
                for call in f.calls:
                    for callee in by_bare_name.get(_bare_target(call), ()):
                        if callee.qualified_name not in seen:
                            seen.add(callee.qualified_name)
                            nxt.append(callee)
            frontier = nxt
            if not frontier:
                break
        return len(seen) - 1

    results: list[RankedEntry] = []
    for f in all_functions:
        is_public = not f.name.startswith("_")
        if public_only and not is_public:
            continue
        if exclude_dunder and f.name.startswith("__") and f.name.endswith("__"):
            continue

        fi = fan_in_map.get(f.qualified_name, 0)
        if fi < min_fan_in:
            continue

        reach = _reachable(f)
        type_bonus = _type_coverage(f)
        score = (fi * 10) + (reach * 0.5) + (type_bonus * 2)
        if not is_public:
            score *= 0.3

        results.append(
            RankedEntry(
                qualified_name=f.qualified_name,
                module=module_of.get(f.qualified_name, f.module_path),
                symbol=f,
                fan_in=fi,
                reachable=reach,
                score=score,
                is_method=f.qualified_name in class_of,
                class_name=class_of.get(f.qualified_name),
            )
        )

    results.sort(key=lambda e: -e.score)
    return results[:top_n] if top_n else results


def _type_coverage(sym: FunctionInfo) -> float:
    """0–1 fraction of typed-signature info available."""
    if sym.return_type or any(p.type_annotation for p in sym.parameters):
        return 1.0
    if sym.decorators:
        return 0.5
    return 0.0


def summarise(entries: list[RankedEntry], top: int = 20) -> str:
    """Return a human-readable priority table."""
    lines = [
        f"{'#':<4} {'qualified_name':<60} {'fan_in':>6} {'reach':>5} {'score':>6}",
        "-" * 82,
    ]
    for i, e in enumerate(entries[:top], 1):
        lines.append(
            f"{i:<4} {e.qualified_name:<60} {e.fan_in:>6} "
            f"{e.reachable:>5} {e.score:>6.1f}"
        )
    return "\n".join(lines)