orchestrata_tools.test_generator.testgen.adapter

Reference › orchestrata_tools › test_generator › testgen › adapter

This module bridges the current AST layer into the shapes the rest of testgen consumes, and parses nothing itself. The generator was restored in August 2026 after an earlier private parser tree was deleted as a duplicate, and was re-pointed at orchestrata_ast: the flat facts in orchestrata_ast.models and the ast.Module trees that CodebaseAnalyzer.trees keeps from its own parse (see analyzer). Its importer is generator, and the tests are tests/test_orchestrata_tools_testgen_adapter_coverage.py, ..._structural_coverage.py and ..._units.py.

A few facts the flat model lacks, such as dataclass field defaults, enum member values and full function bodies, are read off the already-parsed trees by small helpers: find_class_node and find_function_node locate a node by name and line, args_info_from_ast converts an argument list to a dictionary, looks_abstract_via_mixin checks whether a base class makes a class abstract, and class_structural_extra builds the "extra" dictionary that the generators in structural take.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""adapter.py — Bridges the CURRENT AST layer into the shapes the rest of
``testgen`` consumes.

Before 2026-07-09 this generator ran on ``ast_serializer``'s own
``StatementRecord``/``ModuleRecord``/``ProjectRecord`` tree — a second,
competing Python-source-to-tree layer (``scripts/ast_serializer/{parser,
models}.py``), deleted that day as a correctly-identified duplicate of
``orchestrata_ast`` + ``scripts/serialize_codebase_ast.py``. Restoring
``testgen`` 2026-08-19 means re-pointing it at that survivor instead of
resurrecting its own parser.

Two ingredients, both already produced upstream — this module parses
NOTHING itself:

  * ``orchestrata_ast.models.CodebaseAST`` (``ClassInfo``/``FunctionInfo``/
    ``ModuleInfo``) — the flat, already-computed structural facts
    (is_dataclass/is_enum/is_protocol/is_abstract, decorators, bases,
    methods, params, docstrings).
  * ``orchestrata_ast.analyzer.CodebaseAnalyzer.trees`` — the ``ast.Module``
    for each file, already parsed once by the analyzer while it built the
    above. A handful of facts the flat model doesn't carry (dataclass
    field defaults, enum member values/``auto()``-ness, full function
    bodies for branch analysis) are read directly off these ALREADY-PARSED
    trees with small, targeted ``ast.walk``/``iter_child_nodes`` helpers —
    the same idiom ``scripts/serialize_codebase_ast.py`` itself uses
    throughout (``_params_of``, ``_class_node``, ``_branches_of``, ...).
    That is "today's AST snapshot", not a second parser.
"""

from __future__ import annotations

import ast
import re
from pathlib import Path
from typing import TYPE_CHECKING, Any

if TYPE_CHECKING:
    from orchestrata_ast.models import ClassInfo, CodebaseAST

# ---------------------------------------------------------------------------
# Locating nodes in an already-parsed tree
# ---------------------------------------------------------------------------


def find_class_node(
    class_name: str, line: int, tree: ast.Module
) -> ast.ClassDef | None:
    """The ``ast.ClassDef`` matching *class_name* declared at *line*."""
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.ClassDef)
            and node.name == class_name
            and node.lineno == line
        ):
            return node
    return None


def find_function_node(
    func_name: str, line: int, tree: ast.Module
) -> ast.FunctionDef | ast.AsyncFunctionDef | None:
    """The ``ast.FunctionDef``/``ast.AsyncFunctionDef`` matching *func_name*
    declared at *line*."""
    for node in ast.walk(tree):
        if (
            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
            and node.name == func_name
            and node.lineno == line
        ):
            return node
    return None


# ---------------------------------------------------------------------------
# Function signatures → the args dict conditions.py/branch.py expect
# ---------------------------------------------------------------------------


def args_info_from_ast(args: ast.arguments) -> dict[str, Any]:
    """``ast.arguments`` → the ``{"args": [...], "kwonlyargs": [...],
    "posonlyargs": [...], "defaults_count": N, "kw_defaults_count": N}``
    shape ``conditions.analyse``/``branch._build_constructor`` expect.

    ``args`` intentionally includes ``self``/``cls`` at index 0 (when
    present) — callers filter it by name, matching how
    ``ast.arguments.defaults`` counts trailing positional args including
    it.
    """

    def _param(a: ast.arg) -> dict[str, str | None]:
        return {
            "name": a.arg,
            "annotation": ast.unparse(a.annotation) if a.annotation else None,
        }

    return {
        "posonlyargs": [_param(a) for a in args.posonlyargs],
        "args": [_param(a) for a in args.args],
        "kwonlyargs": [_param(a) for a in args.kwonlyargs],
        "defaults_count": len(args.defaults),
        "kw_defaults_count": sum(1 for d in args.kw_defaults if d is not None),
    }


# ---------------------------------------------------------------------------
# Template-defect fix (c): abstract-looking mixin detection
# ---------------------------------------------------------------------------

#: The "mixin self-type" trick this project uses to give pyright visibility
#: across mixin compositions (see e.g. agents/volks/qa/_host.py): a
#: package-local ``_XHost`` Protocol, inherited only under TYPE_CHECKING via
#: ``_MixinBase = _XHost if TYPE_CHECKING else object``. pyright's
#: abstractness check follows the TYPE_CHECKING branch and flags the
#: composed host class as abstract even though it instantiates fine at
#: runtime (mixins inherit plain ``object`` outside TYPE_CHECKING).
_MIXIN_HOST_RE = re.compile(r"_MixinBase\s*=\s*_\w*Host\b")


def looks_abstract_via_mixin(cls_info: ClassInfo, codebase: CodebaseAST) -> bool:
    """True when any of *cls_info*'s base classes uses the
    ``_MixinBase = _XHost if TYPE_CHECKING else object`` pattern in its own
    source — i.e. pyright will flag instantiating *cls_info* as abstract
    even though it is not, at runtime.
    """
    base_names = set(cls_info.bases)
    if not base_names:
        return False
    checked_files: set[str] = set()
    for mod in codebase.modules:
        for c in mod.classes:
            if c.name not in base_names or c.file_path in checked_files:
                continue
            checked_files.add(c.file_path)
            try:
                src = Path(c.file_path).read_text(encoding="utf-8")
            except OSError:
                continue
            if _MIXIN_HOST_RE.search(src):
                return True
    return False


# ---------------------------------------------------------------------------
# Structural "extra" dict — dataclass / enum / abstract / protocol
# ---------------------------------------------------------------------------


def _callee_name(node: ast.expr) -> str:
    if isinstance(node, ast.Name):
        return node.id
    if isinstance(node, ast.Attribute):
        return node.attr
    return ""


def _safe_unparse(node: ast.expr) -> str:
    try:
        return ast.unparse(node)
    except Exception:
        return "..."


def _dataclass_fields_from_node(class_node: ast.ClassDef) -> list[dict[str, Any]]:
    """Per-field metadata for a ``@dataclass`` body, read directly off the
    already-parsed ``ast.ClassDef`` — ``ClassInfo.class_vars`` only carries
    name/annotation/default-as-string, not enough to tell a bare default
    apart from a ``field(default_factory=..., metadata=...)`` call.
    """
    fields: list[dict[str, Any]] = []
    for item in class_node.body:
        if not (isinstance(item, ast.AnnAssign) and isinstance(item.target, ast.Name)):
            continue
        name = item.target.id
        if name.startswith("_"):
            continue
        annotation = _safe_unparse(item.annotation)
        value = item.value

        default_repr: str | None = None
        default_factory_repr: str | None = None
        metadata_repr: str | None = None
        is_field_call = False
        required = value is None

        if isinstance(value, ast.Call) and _callee_name(value.func) == "field":
            is_field_call = True
            for kw in value.keywords:
                if kw.arg == "default":
                    default_repr = _safe_unparse(kw.value)
                elif kw.arg == "default_factory":
                    default_factory_repr = _safe_unparse(kw.value)
                elif kw.arg == "metadata":
                    metadata_repr = _safe_unparse(kw.value)
            required = default_repr is None and default_factory_repr is None
        elif value is not None:
            default_repr = _safe_unparse(value)
            required = False

        fields.append(
            {
                "name": name,
                "type": annotation,
                "required": required,
                "default": default_repr,
                "default_factory": default_factory_repr,
                "is_field_call": is_field_call,
                "metadata": metadata_repr,
            }
        )
    return fields


def _dataclass_params(cls_info: ClassInfo) -> dict[str, str]:
    dec = next((d for d in cls_info.decorators if "dataclass" in d), "")
    params: dict[str, str] = {}
    for key in ("frozen", "slots", "eq"):
        m = re.search(rf"{key}\s*=\s*(True|False)", dec)
        if m:
            params[key] = m.group(1)
    return params


def _enum_members_from_node(class_node: ast.ClassDef) -> list[dict[str, Any]]:
    """Enum members, read directly off the ``ast.ClassDef`` body —
    ``ClassInfo.class_vars`` only captures ANNOTATED assignments
    (``ast.AnnAssign``), and enum members are conventionally plain
    ``NAME = value``/``NAME = auto()`` (``ast.Assign``).
    """
    members: list[dict[str, Any]] = []
    for item in class_node.body:
        target: ast.expr | None = None
        value: ast.expr | None = None
        if isinstance(item, ast.Assign) and len(item.targets) == 1:
            target = item.targets[0]
            value = item.value
        elif isinstance(item, ast.AnnAssign) and item.value is not None:
            target = item.target
            value = item.value
        if not (isinstance(target, ast.Name) and value is not None):
            continue
        name = target.id
        if name.startswith("_"):
            continue
        is_auto = isinstance(value, ast.Call) and _callee_name(value.func) == "auto"
        value_repr = "None" if is_auto else _safe_unparse(value)
        members.append({"name": name, "is_auto": is_auto, "value_repr": value_repr})
    return members


def class_structural_extra(
    cls_info: ClassInfo, class_node: ast.ClassDef
) -> dict[str, Any]:
    """Build the ``extra`` dict ``structural.py``'s ``gen_*_tests`` functions
    expect, from the current AST layer's ``ClassInfo`` plus a narrow direct
    read of the class's own already-parsed ``ast.ClassDef`` body.
    """
    extra: dict[str, Any] = {
        "name": cls_info.name,
        "is_dataclass": cls_info.is_dataclass,
        "is_enum": cls_info.is_enum,
        "is_protocol": cls_info.is_protocol,
        "is_abstract": cls_info.is_abstract,
    }

    if cls_info.is_dataclass:
        extra["dataclass_fields"] = _dataclass_fields_from_node(class_node)
        extra["dataclass_params"] = _dataclass_params(cls_info)
        extra["has_post_init"] = any(
            m.name == "__post_init__" for m in cls_info.methods
        )

    if cls_info.is_enum:
        extra["enum_members"] = _enum_members_from_node(class_node)
        base = next(
            (b for b in cls_info.bases if b in ("IntEnum", "StrEnum")),
            "Enum" if any("Enum" in b for b in cls_info.bases) else "Enum",
        )
        extra["enum_base"] = base

    abstract_methods = [m for m in cls_info.methods if m.is_abstract]
    extra["abstract_methods"] = [
        {
            "name": m.name,
            "signature": f"-> {m.return_type}" if m.return_type else "",
            "is_property": m.is_property,
        }
        for m in abstract_methods
    ]
    extra["abc_base"] = next(
        (b for b in cls_info.bases if "ABC" in b or "Abstract" in b), "ABC"
    )

    if cls_info.is_protocol:
        members: list[dict[str, Any]] = []
        for cv in cls_info.class_vars:
            if cv.name.startswith("_"):
                continue
            members.append(
                {"kind": "attribute", "name": cv.name, "type": cv.type_annotation}
            )
        for m in cls_info.methods:
            if m.name.startswith("_") and not m.is_property:
                continue
            members.append(
                {
                    "kind": "async_method" if m.is_async else "method",
                    "name": m.name,
                    "returns": m.return_type,
                    "is_property": m.is_property,
                    "is_classmethod": m.is_classmethod,
                    "is_staticmethod": m.is_staticmethod,
                }
            )
        extra["protocol_members"] = members
        extra["is_runtime_checkable"] = any(
            "runtime_checkable" in d for d in cls_info.decorators
        )

    return extra