orchestrata_ultimate.verification.review.checks_structure

Reference › orchestrata_ultimate › verification › review › checks_structure

The structure checks of the deterministic review, the largest of the check modules. It is called by the review engine and the refactor codemod's review checks. The checks are:

• check_function_length (check 10): at most 40 lines per function.
• check_function_args (check 11): at most 5 parameters.
• check_no_placeholder (check 12): no pass-only bodies or raise NotImplementedError; it relies on _is_stub_body, _strip_docstrings, _is_not_implemented_raise, _contains_await and _contains_external_call to tell a real body from a stub.
• check_error_handling (check 13): no bare except.
• check_nesting_depth (check 30): control-flow nesting no deeper than 4 levels.
• check_god_objects (check 31): classes within size and complexity thresholds.
• check_stub_methods (check 32): methods must not be log-only or no-ops, using _is_log_call.

The first four return issue tuples; the last three return (check id, file, description) tuples in the same shape. Helpers are in helpers.

At a glance

Class diagram

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/review/checks_structure.py

# mypy: ignore-errors
"""Structure checks: function length, args, placeholders, error handling.

Covers checklist items 10, 11, 12, 13 from the CODE_REVIEW_PROMPT.
"""

import ast
import os
from typing import TypeGuard

from orchestrata_ultimate.verification.review.helpers import (
    _function_line_count,
)
from orchestrata_ultimate.verification.review.types import Issue


def check_function_length(filepath: str, tree: ast.Module) -> list[Issue]:
    """Check 10: FUNCTION_LENGTH -- max 40 lines per function.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        List of function length violation issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            length = _function_line_count(node)
            if length > 40:
                issues.append(
                    (
                        "FUNC_LENGTH",
                        rel_name,
                        f"L{node.lineno}: `{node.name}()` is {length} lines (max 40).",
                    )
                )
    return issues


def check_function_args(filepath: str, tree: ast.Module) -> list[Issue]:
    """Check 11: FUNCTION_ARGS -- max 5 parameters.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        List of parameter count violation issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            args = node.args
            count = len(args.args) + len(args.kwonlyargs)
            if args.args and args.args[0].arg in ("self", "cls"):
                count -= 1
            if count > 5:
                issues.append(
                    (
                        "FUNC_ARGS",
                        rel_name,
                        f"L{node.lineno}: `{node.name}()` has {count} params (max 5).",
                    )
                )
    return issues


def check_no_placeholder(filepath: str, tree: ast.Module) -> list[Issue]:
    """Check 12: NO_PLACEHOLDER -- no pass/NotImplementedError.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        List of placeholder code issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)

    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        for child in ast.walk(node):
            if isinstance(child, ast.Pass):
                non_doc = [
                    n
                    for n in node.body
                    if not (
                        isinstance(n, ast.Expr) and isinstance(n.value, ast.Constant)
                    )
                ]
                if len(non_doc) == 1 and isinstance(non_doc[0], ast.Pass):
                    issues.append(
                        (
                            "PLACEHOLDER",
                            rel_name,
                            f"L{node.lineno}: `{node.name}()` "
                            f"has empty body (only pass).",
                        )
                    )
                    break

            if _is_not_implemented_raise(child, node):
                issues.append(
                    (
                        "PLACEHOLDER",
                        rel_name,
                        f"L{child.lineno}: NotImplementedError in `{node.name}()`.",
                    )
                )

    return issues


def _is_not_implemented_raise(
    child: ast.AST,
    node: ast.FunctionDef | ast.AsyncFunctionDef,
) -> TypeGuard[ast.Raise]:
    """Check if a node is a raise NotImplementedError.

    A ``TypeGuard`` (not a plain ``bool``) so the caller's ``child.lineno``
    access narrows from ``ast.AST`` (no ``lineno``) to ``ast.Raise`` (has
    ``lineno``) instead of needing a redundant ``isinstance`` re-check.

    Args:
        child: AST node to check.
        node: Parent function node (unused, for context).

    Returns:
        True if the node raises NotImplementedError.
    """
    return (
        isinstance(child, ast.Raise)
        and child.exc is not None
        and isinstance(child.exc, ast.Call)
        and isinstance(child.exc.func, ast.Name)
        and child.exc.func.id == "NotImplementedError"
    )


# ── Stub-method detection predicates (CHECK 32) ───────────────────────


def _is_log_call(node: ast.expr) -> bool:
    """Check if expression is a logging call."""
    if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
        return node.func.attr in ("info", "debug", "warning", "error", "exception")
    return False


def _contains_await(stmts: list[ast.stmt]) -> bool:
    """Check if body contains any await statement (at any depth)."""
    for stmt in stmts:
        for node in ast.walk(stmt):
            if isinstance(node, ast.Await):
                return True
    return False


def _contains_external_call(stmts: list[ast.stmt]) -> bool:
    """Check if body contains function/method calls (not just constructors)."""
    for stmt in stmts:
        for node in ast.walk(stmt):
            if isinstance(node, ast.Call):
                func = node.func
                if isinstance(func, ast.Name):
                    name = func.id
                    if name[0].islower():
                        return True
                if isinstance(func, ast.Attribute):
                    return True
    return False


def _is_stub_body(body: list[ast.stmt]) -> bool:
    """Return True if the function body is a stub.

    A stub is a method that only contains:
    - docstrings
    - pass statements
    - logging calls
    - local variable assignments
    - return statements with constructor calls (no external function calls)

    If there's an await or external function/method call, it's doing real work.
    """
    if _contains_await(body):
        return False
    if _contains_external_call(body):
        return False
    meaningful = [
        stmt
        for stmt in body
        if not isinstance(stmt, (ast.Expr, ast.Pass))
        or (isinstance(stmt, ast.Expr) and not _is_log_call(stmt.value))
    ]
    if not meaningful:
        return True
    non_assign = [
        s
        for s in meaningful
        if not isinstance(s, (ast.Assign, ast.AnnAssign, ast.AugAssign, ast.Return))
    ]
    if meaningful and not non_assign:
        return True
    return False


def _strip_docstrings(node: ast.FunctionDef | ast.AsyncFunctionDef) -> list[ast.stmt]:
    """Return the function body with leading/standalone docstring exprs removed."""
    return [
        s
        for s in node.body
        if not (
            isinstance(s, ast.Expr)
            and isinstance(s.value, ast.Constant)
            and isinstance(s.value.value, str)
        )
    ]


def check_error_handling(filepath: str, tree: ast.Module) -> list[Issue]:
    """Check 13: ERROR_HANDLING -- no bare except.

    Args:
        filepath: Absolute path to the file.
        tree: Parsed AST module.

    Returns:
        List of bare except issues.
    """
    issues: list[Issue] = []
    rel_name = os.path.basename(filepath)

    for node in ast.walk(tree):
        if isinstance(node, ast.ExceptHandler):
            if node.type is None:
                issues.append(
                    (
                        "ERROR_HANDLING",
                        rel_name,
                        f"L{node.lineno}: Bare `except:` -- "
                        f"use specific exception types.",
                    )
                )

    return issues


def check_nesting_depth(
    tree: ast.Module,
    lines: list[str],
    filepath: str,
) -> list[tuple[str, str, str]]:
    """CHECK 30: Control-flow nesting must not exceed 4 levels.

    Counts nested if/for/while/with/try blocks. Functions and
    classes reset the nesting counter.

    Args:
        tree: Parsed AST module.
        lines: Source lines.
        filepath: File path for reporting.

    Returns:
        List of (check_id, filepath, description) issues.
    """
    issues: list[tuple[str, str, str]] = []
    max_depth = 4

    class NestingVisitor(ast.NodeVisitor):
        """Visitor that tracks control-flow nesting depth."""

        def __init__(self) -> None:
            self.depth = 0
            self.max_seen = 0
            self.worst_line = 0

        def _enter_nesting(self, node: ast.AST) -> None:
            """Enter a nesting level, visit children, exit."""
            self.depth += 1
            if self.depth > self.max_seen:
                self.max_seen = self.depth
                self.worst_line = getattr(node, "lineno", 0)
            for child in ast.iter_child_nodes(node):
                self.visit(child)
            self.depth -= 1

        def visit(self, node: ast.AST) -> None:
            """Dispatch to the snake_case handler for nesting-inducing nodes."""
            if isinstance(
                node,
                (
                    ast.If,
                    ast.For,
                    ast.While,
                    ast.With,
                    ast.Try,
                    ast.AsyncWith,
                    ast.AsyncFor,
                ),
            ):
                self._enter_nesting(node)
                return
            super().visit(node)

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            visitor = NestingVisitor()
            visitor.visit(node)
            if visitor.max_seen > max_depth:
                issues.append(
                    (
                        "NESTING_DEPTH",
                        filepath,
                        f"Function '{node.name}' (line {node.lineno}) "
                        f"has nesting depth {visitor.max_seen} "
                        f"(max {max_depth}), "
                        f"deepest at line {visitor.worst_line}",
                    )
                )

    return issues


def check_god_objects(
    tree: ast.Module,
    lines: list[str],
    filepath: str,
) -> list[tuple[str, str, str]]:
    """CHECK 31: Classes must not exceed size/complexity thresholds.

    God object thresholds:
    - Max 300 lines per class
    - Max 15 public methods per class
    - Max 500 lines per file

    Args:
        tree: Parsed AST module.
        lines: Source lines.
        filepath: File path for reporting.

    Returns:
        List of (check_id, filepath, description) issues.
    """
    issues: list[tuple[str, str, str]] = []
    max_class_lines = 300
    max_public_methods = 15
    max_file_lines = 500

    if len(lines) > max_file_lines:
        issues.append(
            (
                "GOD_OBJECT",
                filepath,
                f"File has {len(lines)} lines (max {max_file_lines})",
            )
        )

    for node in ast.walk(tree):
        if not isinstance(node, ast.ClassDef):
            continue

        end_line = getattr(node, "end_lineno", node.lineno)
        class_lines = end_line - node.lineno + 1
        if class_lines > max_class_lines:
            issues.append(
                (
                    "GOD_OBJECT",
                    filepath,
                    f"Class '{node.name}' (line {node.lineno}) has "
                    f"{class_lines} lines (max {max_class_lines})",
                )
            )

        public_methods = [
            n
            for n in node.body
            if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))
            and not n.name.startswith("_")
        ]
        if len(public_methods) > max_public_methods:
            issues.append(
                (
                    "GOD_OBJECT",
                    filepath,
                    f"Class '{node.name}' (line {node.lineno}) has "
                    f"{len(public_methods)} public methods "
                    f"(max {max_public_methods})",
                )
            )

    return issues


def check_stub_methods(
    tree: ast.Module,
    lines: list[str],
    filepath: str,
) -> list[tuple[str, str, str]]:
    """CHECK 32: Methods must not be stubs (log-only, no-op).

    Detects methods that only contain logging calls and local
    variable assignments but perform no actual work (no return
    of computed values, no calls to external services, no state
    mutation beyond local tracking).

    Test files are excluded from this check.

    Args:
        tree: Parsed AST module.
        lines: Source lines.
        filepath: File path for reporting.

    Returns:
        List of (check_id, filepath, description) issues.
    """
    issues: list[tuple[str, str, str]] = []

    if "/tests/" in filepath or filepath.startswith("tests/"):
        return issues

    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        body = _strip_docstrings(node)
        if not body:
            continue
        if _is_stub_body(body):
            issues.append(
                (
                    "STUB_METHOD",
                    filepath,
                    f"Method '{node.name}' (line {node.lineno}) "
                    f"appears to be a stub — only contains "
                    f"logging/local tracking with no actual "
                    f"side effects",
                )
            )

    return issues