refactor: reduce SDK complexity and improve CRAP analysis

.github/scripts/check_crap_threshold.py

@@ -1,26 +1,48 @@
 #!/usr/bin/env python3
-"""Fail when pytest-crap reports a CRAP score at or above a threshold."""
+"""Enforce CRAP scores using executable-statement coverage."""
 
 from __future__ import annotations
 
 import argparse
+from dataclasses import replace
 import sys
 from pathlib import Path
 
-from coverage import CoverageData
-from pytest_crap.calculator import calculate_crap
+from coverage import Coverage
+from pytest_crap.calculator import FunctionScore, calculate_crap
 
 
-def covered_lines_for_file(coverage_file: Path, source_file: Path) -> set[int]:
-    data = CoverageData(basename=str(coverage_file))
-    data.read()
+def scores_for_file(coverage: Coverage, source_file: Path) -> list[FunctionScore]:
+    measured_files = {
+        Path(filename).resolve() for filename in coverage.get_data().measured_files()
+    }
+    if source_file.resolve() not in measured_files:
+        raise SystemExit(f"No coverage data found for {source_file.resolve()}")
 
-    source_file = source_file.resolve()
-    for measured_file in data.measured_files():
-        if Path(measured_file).resolve() == source_file:
-            return set(data.lines(measured_file) or [])
-
-    raise SystemExit(f"No coverage data found for {source_file}")
+    _, statements, excluded, missing, _ = coverage.analysis2(str(source_file))
+    executable_lines = set(statements) - set(excluded)
+    missing_lines = set(missing)
+    scores = []
+    for score in calculate_crap(str(source_file), executable_lines - missing_lines):
+        function_lines = {
+            line
+            for line in executable_lines
+            if score.start_line <= line <= score.end_line
+        }
+        fraction = (
+            len(function_lines - missing_lines) / len(function_lines)
+            if function_lines
+            else 1.0
+        )
+        # pytest-crap's physical-line denominator penalizes comments and docstrings.
+        scores.append(
+            replace(
+                score,
+                coverage_percent=fraction * 100,
+                crap=score.cc**2 * (1 - fraction) ** 3 + score.cc,
+            )
+        )
+    return scores
 
 
 def main() -> int:
@@ -30,9 +52,31 @@ def main() -> int:
     parser.add_argument("--max-crap", type=float, required=True)
     args = parser.parse_args()
 
-    covered_lines = covered_lines_for_file(args.coverage_file, args.source_file)
-    scores = calculate_crap(str(args.source_file), covered_lines)
+    if args.source_file.is_dir():
+        source_files = sorted(
+            path
+            for path in args.source_file.rglob("*.py")
+            if not {"test", "tests"}.intersection(
+                path.relative_to(args.source_file).parts
+            )
+        )
+    else:
+        source_files = [args.source_file]
+    if not source_files:
+        raise SystemExit(f"No production Python files found in {args.source_file}")
+
+    coverage = Coverage(data_file=str(args.coverage_file))
+    coverage.load()
+    scores = [
+        score
+        for source_file in source_files
+        for score in scores_for_file(coverage, source_file)
+    ]
     offenders = [score for score in scores if score.crap >= args.max_crap]
+    print(
+        f"Analyzed {len(scores)} functions in {len(source_files)} production files; "
+        f"{len(offenders)} functions scored at or above {args.max_crap:g}"
+    )
 
     if not offenders:
         print(f"All CRAP scores are below {args.max_crap:g} for {args.source_file}")
@@ -41,7 +85,7 @@ def main() -> int:
     print(f"CRAP scores must be below {args.max_crap:g} for {args.source_file}")
     for score in sorted(offenders, key=lambda item: item.crap, reverse=True):
         print(
-            f"{score.name}: CRAP={score.crap:.2f}, "
+            f"{score.file_path}:{score.start_line} {score.name}: CRAP={score.crap:.2f}, "
             f"CC={score.cc}, coverage={score.coverage_percent:.1f}%"
         )
     return 1

.github/scripts/check_public_api.py

@@ -233,11 +233,14 @@ def _load_griffe():
 
 def generate_snapshot() -> str:
     griffe = _load_griffe()
+    from constructor_helper_attributes import ConstructorHelperAttributes
+
     package = griffe.load(
         "posthog",
         allow_inspection=False,
         search_paths=[ROOT],
         submodules=True,
+        extensions=griffe.load_extensions(ConstructorHelperAttributes()),
     )
     records = sorted(_record(obj) for obj in _iter_module_members(package))
     return HEADER + "\n".join(records) + "\n"

.github/scripts/constructor_helper_attributes.py

@@ -0,0 +1,110 @@
+"""Recover public instance attributes initialized by private constructor helpers."""
+
+from __future__ import annotations
+
+import ast
+from typing import Iterator
+
+from griffe import Class, Extension, Function, Visitor, ast_children
+
+
+_DEFERRED_SCOPES = (
+    ast.FunctionDef,
+    ast.AsyncFunctionDef,
+    ast.ClassDef,
+    ast.Lambda,
+    ast.GeneratorExp,
+)
+
+
+def _eager_nodes(node: ast.AST) -> Iterator[ast.AST]:
+    if isinstance(node, _DEFERRED_SCOPES):
+        return
+    yield node
+    for child in ast_children(node):
+        yield from _eager_nodes(child)
+
+
+def _public_instance_target(node: ast.AST) -> ast.Attribute | None:
+    if isinstance(node, ast.Assign):
+        targets = node.targets
+    elif isinstance(node, ast.AnnAssign):
+        targets = [node.target]
+    else:
+        return None
+    if (
+        len(targets) == 1
+        and isinstance(target := targets[0], ast.Attribute)
+        and isinstance(target.value, ast.Name)
+        and target.value.id == "self"
+        and not target.attr.startswith("_")
+    ):
+        return target
+    return None
+
+
+class ConstructorHelperAttributes(Extension):
+    """Add missing attributes from direct, same-class private helper calls.
+
+    Griffe ignores method-body assignments outside ``__init__``. Replay only
+    simple public instance assignments from reachable synchronous, undecorated helpers,
+    keeping existing constructor and class records unchanged.
+    """
+
+    def on_class_members(self, *, node, cls: Class, agent, **kwargs) -> None:
+        if not isinstance(node, ast.ClassDef) or not isinstance(agent, Visitor):
+            return
+        methods = {
+            method.name: method
+            for method in node.body
+            if isinstance(method, ast.FunctionDef)
+            and not method.decorator_list
+            and method.args.args
+            and method.args.args[0].arg == "self"
+        }
+        constructor = cls.members.get("__init__")
+        if not isinstance(constructor, Function) or "__init__" not in methods:
+            return
+        existing_names = set(cls.members)
+        previous = agent.current
+        try:
+            # Griffe's constructor visitor also stores Function despite this annotation.
+            agent.current = constructor  # type: ignore[assignment]
+            self._follow(
+                methods["__init__"], methods, agent, existing_names, set(), False
+            )
+        finally:
+            agent.current = previous
+
+    def _follow(self, method, methods, agent, existing_names, active, in_helper):
+        if method.name in active:
+            return
+        active.add(method.name)
+        try:
+            # ast_children supplies parent links required by Griffe's assignment visitor.
+            for statement in ast_children(method):
+                if statement not in method.body:
+                    continue
+                for node in _eager_nodes(statement):
+                    target = _public_instance_target(node) if in_helper else None
+                    if target is not None and target.attr not in existing_names:
+                        agent.visit(node)
+                    if (
+                        isinstance(node, ast.Call)
+                        and isinstance(node.func, ast.Attribute)
+                        and isinstance(node.func.value, ast.Name)
+                        and node.func.value.id == "self"
+                        and node.func.attr.startswith("_")
+                        and not node.func.attr.startswith("__")
+                        and node.func.attr in methods
+                    ):
+                        self._follow(
+                            methods[node.func.attr],
+                            methods,
+                            agent,
+                            existing_names,
+                            active,
+                            True,
+                        )
+        finally:
+            active.remove(method.name)

.github/scripts/test_check_crap_threshold.py

@@ -0,0 +1,147 @@
+"""Regression tests for executable-line CRAP scoring."""
+
+from pathlib import Path
+import subprocess
+import sys
+
+from coverage import CoverageData
+import pytest
+
+
+SCRIPT_PATH = Path(__file__).with_name("check_crap_threshold.py")
+FUNCTION = """def choose(value):
+    if value == 1:
+        return 'one'
+    if value == 2:
+        return 'two'
+    if value == 3:
+        return 'three'
+    if value == 4:
+        return 'four'
+    return 'other'
+
+for value in range(5):
+    choose(value)
+"""
+
+
+def collect_coverage(tmp_path, source):
+    source_file = (tmp_path / "example.py").resolve()
+    source_file.write_text(source)
+    data_file = tmp_path / ".coverage"
+    config_file = tmp_path / ".coveragerc"
+    config_file.write_text("[run]\n")
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "coverage",
+            "run",
+            f"--rcfile={config_file}",
+            f"--data-file={data_file}",
+            str(source_file),
+        ],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return source_file, data_file
+
+
+def check(source_file, data_file, threshold):
+    return subprocess.run(
+        [
+            sys.executable,
+            str(SCRIPT_PATH),
+            str(source_file),
+            "--coverage-file",
+            str(data_file),
+            "--max-crap",
+            str(threshold),
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+
+@pytest.mark.parametrize(
+    "source",
+    [
+        FUNCTION,
+        FUNCTION.replace(
+            "    if value == 1:", "    # Explanation\n\n" * 30 + "    if value == 1:"
+        ),
+        FUNCTION.replace(
+            "    if value == 1:",
+            '    """Description.\n'
+            + "    Details.\n" * 30
+            + '    """\n    if value == 1:',
+        ),
+        FUNCTION.replace("def choose(value):", "def choose(\n    value,\n):"),
+    ],
+    ids=["plain", "comments-and-blanks", "docstring", "multiline-signature"],
+)
+def test_fully_covered_function_is_unaffected_by_non_executable_lines(tmp_path, source):
+    source_file, data_file = collect_coverage(tmp_path, source)
+    result = check(source_file, data_file, 10)
+    assert result.returncode == 0, result.stdout + result.stderr
+    at_complexity_floor = check(source_file, data_file, 5)
+    assert at_complexity_floor.returncode == 1
+    assert "CRAP=5.00, CC=5, coverage=100.0%" in at_complexity_floor.stdout
+
+
+def test_uncovered_executable_branch_still_increases_score(tmp_path):
+    source_file, data_file = collect_coverage(
+        tmp_path,
+        "def choose(value):\n    if value:\n        return 1\n    return 0\n\nchoose(True)\n",
+    )
+    result = check(source_file, data_file, 2.0625)
+    assert result.returncode == 1
+    assert "CRAP=2.06, CC=2, coverage=75.0%" in result.stdout
+    assert check(source_file, data_file, 2.0626).returncode == 0
+
+
+def test_coverage_exclusions_are_not_missing_statements(tmp_path):
+    source_file, data_file = collect_coverage(
+        tmp_path,
+        "def choose(value):\n    if value:  # pragma: no cover\n        return 1\n    return 0\n\nchoose(False)\n",
+    )
+    result = check(source_file, data_file, 2)
+    assert result.returncode == 1
+    assert "CRAP=2.00, CC=2, coverage=100.0%" in result.stdout
+
+
+def test_directory_includes_unexecuted_production_and_excludes_test_helpers(tmp_path):
+    package = tmp_path / "posthog"
+    source_file = package / "ai" / "converter.py"
+    source_file.parent.mkdir(parents=True)
+    source_file.write_text(
+        "def choose(value):\n    if value:\n        return 1\n    return 0\n"
+    )
+    for directory in ("test", "tests"):
+        helper = package / directory / "ai" / "utils.py"
+        helper.parent.mkdir(parents=True)
+        helper.write_text("not valid Python")
+    data_file = tmp_path / ".coverage"
+    data = CoverageData(basename=str(data_file))
+    data.add_lines({str(source_file.resolve()): set()})
+    data.write()
+    result = check(package, data_file, 6)
+    assert result.returncode == 1
+    assert "1 functions in 1 production files" in result.stdout
+    assert "CRAP=6.00, CC=2, coverage=0.0%" in result.stdout
+    assert str(source_file) in result.stdout
+
+
+def test_missing_coverage_does_not_pass(tmp_path):
+    source_file = tmp_path / "example.py"
+    source_file.write_text("def choose():\n    return 1\n")
+    result = check(source_file, tmp_path / ".coverage", 10)
+    assert result.returncode != 0
+    assert "No coverage data found" in result.stderr
+
+
+def test_empty_source_directory_does_not_pass(tmp_path):
+    result = check(tmp_path, tmp_path / ".coverage", 10)
+    assert result.returncode != 0
+    assert "No production Python files found" in result.stderr