diff --git a/desloppify/languages/python/extractors.py b/desloppify/languages/python/extractors.py index 5435dcf1..3c79a1bd 100644 --- a/desloppify/languages/python/extractors.py +++ b/desloppify/languages/python/extractors.py @@ -1,5 +1,6 @@ """Python extraction: function bodies, class structure, param patterns.""" +import ast import hashlib import re from pathlib import Path @@ -125,42 +126,131 @@ def py_passthrough_pattern(name: str) -> str: return rf"\b{escaped}\s*=\s*{escaped}\b" -_PY_DEF_RE = re.compile(r"^def\s+(\w+)\s*\(", re.MULTILINE) +def _function_param_names(node: ast.FunctionDef) -> list[str]: + """Return the explicit parameter names that count toward wrapper forwarding.""" + + arguments = [ + *node.args.posonlyargs, + *node.args.args, + *node.args.kwonlyargs, + ] + if node.args.vararg is not None: + arguments.append(node.args.vararg) + if node.args.kwarg is not None: + arguments.append(node.args.kwarg) + return [ + argument.arg for argument in arguments if argument.arg not in {"self", "cls"} + ] + + +def _is_docstring_expr(statement: ast.stmt) -> bool: + """Return whether a statement is a function docstring expression.""" + + return ( + isinstance(statement, ast.Expr) + and isinstance(statement.value, ast.Constant) + and isinstance(statement.value.value, str) + ) + + +def _is_direct_parameter_value(value: ast.expr, parameter_names: set[str]) -> bool: + """Return whether a call argument directly forwards one function parameter.""" + + if isinstance(value, ast.Name): + return value.id in parameter_names + return ( + isinstance(value, ast.Starred) + and isinstance(value.value, ast.Name) + and value.value.id in parameter_names + ) + + +def _is_direct_call_target(target: ast.expr) -> bool: + """Return whether a call target is a direct non-constructor name chain. + + A forwarding call to a lower-case function can be redundant, while a + Capitalized target conventionally constructs a value. Constructor + factories are meaningful transformations, even when every field is + supplied from a same-named parameter. + """ + + names: list[str] = [] + while isinstance(target, ast.Attribute): + names.append(target.attr) + target = target.value + if not isinstance(target, ast.Name): + return False + names.append(target.id) + return not any(name[:1].isupper() for name in names) + + +def _passthrough_return_statement( + node: ast.FunctionDef, + parameter_names: list[str], +) -> ast.Return | None: + """Return the sole direct-forwarding return statement for a pure wrapper. + + Decorated entrypoints and multi-step functions can forward many arguments + while still constructing requests, making decisions, or retaining state. + They are not the redundant wrappers this detector is meant to surface. + """ + + if node.decorator_list: + return None + statements = list(node.body) + if statements and _is_docstring_expr(statements[0]): + statements = statements[1:] + if len(statements) != 1 or not isinstance(statements[0], ast.Return): + return None + statement = statements[0] + if not isinstance(statement.value, ast.Call): + return None + if not _is_direct_call_target(statement.value.func): + return None + + values = [ + *statement.value.args, + *(keyword.value for keyword in statement.value.keywords), + ] + if not values: + return None + parameter_name_set = set(parameter_names) + if not all( + _is_direct_parameter_value(value, parameter_name_set) for value in values + ): + return None + return statement def detect_passthrough_functions(path: Path) -> list[dict]: - """Detect Python functions where most params are same-name forwarded.""" + """Detect pure Python wrappers where most params are same-name forwarded.""" entries = [] for filepath in find_py_files(path): content = read_file(filepath) if content is None: continue - for m in _PY_DEF_RE.finditer(content): - name = m.group(1) - depth = 1 - i = m.end() - while i < len(content) and depth > 0: - ch = content[i] - if ch == "(": - depth += 1 - elif ch == ")": - depth -= 1 - i += 1 - if depth != 0: + try: + tree = ast.parse(content, filename=filepath) + except SyntaxError: + continue + for node in tree.body: + if not isinstance(node, ast.FunctionDef): continue - param_str = content[m.end() : i - 1] - params = extract_py_params(param_str) + name = node.name + params = _function_param_names(node) if len(params) < 4: continue - rest_after_paren = content[i:] - colon_m = re.search(r":", rest_after_paren) - if not colon_m: + return_statement = _passthrough_return_statement(node, params) + if return_statement is None: + continue + body = ast.get_source_segment(content, return_statement) + if body is None: + continue + call = return_statement.value + if not isinstance(call, ast.Call): continue - rest = rest_after_paren[colon_m.end() :] - bm = re.search(r"\n(?=[^\s\n#])", rest) - body = rest[: bm.start()] if bm else rest - has_kwargs_spread = bool(re.search(r"\*\*kwargs\b", body)) + has_kwargs_spread = any(keyword.arg is None for keyword in call.keywords) pt, direct = classify_params( params, body, py_passthrough_pattern, occurrences_per_match=2 ) @@ -183,7 +273,7 @@ def detect_passthrough_functions(path: Path) -> list[dict]: "passthrough": len(pt), "direct": len(direct), "ratio": round(ratio, 2), - "line": content[: m.start()].count("\n") + 1, + "line": node.lineno, "tier": tier, "confidence": confidence, "passthrough_params": sorted(pt), diff --git a/desloppify/languages/python/tests/test_py_extractors.py b/desloppify/languages/python/tests/test_py_extractors.py index f44d15bb..ac34d8f1 100644 --- a/desloppify/languages/python/tests/test_py_extractors.py +++ b/desloppify/languages/python/tests/test_py_extractors.py @@ -350,7 +350,7 @@ def test_base_classes_extracted(self, tmp_path): class TestDetectPassthrough: - def test_passthrough_detected(self, tmp_path): + def test_lowercase_function_wrapper_detected(self, tmp_path): fp = tmp_path / "pt.py" fp.write_text( textwrap.dedent("""\ @@ -359,9 +359,186 @@ def wrapper(a, b, c, d, e): """) ) entries = detect_passthrough_functions(tmp_path) - if entries: - assert entries[0]["function"] == "wrapper" - assert entries[0]["passthrough"] >= 4 + wrapper = next(entry for entry in entries if entry["function"] == "wrapper") + assert wrapper["passthrough"] >= 4 + + def test_documented_passthrough_detected(self, tmp_path): + fp = tmp_path / "documented.py" + fp.write_text( + textwrap.dedent('''\ + def wrapper(a, b, c, d, e): + """Backward-compatible forwarding alias.""" + return inner(a=a, b=b, c=c, d=d, e=e) + ''') + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "wrapper" in names + + def test_decorated_entrypoint_is_not_flagged(self, tmp_path): + fp = tmp_path / "entrypoint.py" + fp.write_text( + textwrap.dedent("""\ + @register_command + @with_application_context + def seed(a, b, c, d, e): + request = SeedRequest(a=a, b=b, c=c, d=d, e=e) + run_seed(request) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "seed" not in names + + def test_decorated_pure_wrapper_is_not_flagged(self, tmp_path): + fp = tmp_path / "decorated_wrapper.py" + fp.write_text( + textwrap.dedent("""\ + @register_command + def callback(a, b, c, d, e): + return handle(a=a, b=b, c=c, d=d, e=e) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "callback" not in names + + def test_multi_step_decision_router_is_not_flagged(self, tmp_path): + fp = tmp_path / "decision.py" + fp.write_text( + textwrap.dedent("""\ + def decide(model, *, receiver_ip3, threshold, evidence, profile_required, unmodeled_orders): + power_informed = power_decision( + model, + threshold=threshold, + evidence=evidence, + profile_required=profile_required, + ) + if power_informed is not None: + return power_informed + return fallback_decision( + model, + receiver_ip3=receiver_ip3, + threshold=threshold, + evidence=evidence, + profile_required=profile_required, + unmodeled_orders=unmodeled_orders, + ) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "decide" not in names + + def test_looping_validator_is_not_flagged(self, tmp_path): + fp = tmp_path / "validator.py" + fp.write_text( + textwrap.dedent("""\ + def validate_regions(shaped, *, errors, point_to_xy_fn, polygon_points_fn, point_in_polygon_fn): + region_keys = set() + for region in shaped.regions: + polygon = polygon_points_fn(region) + point = point_to_xy_fn(region.get("representative_point")) + if point is not None and not point_in_polygon_fn(point, polygon): + errors.append("representative point outside polygon") + validate_region( + region, + errors=errors, + point_to_xy_fn=point_to_xy_fn, + polygon_points_fn=polygon_points_fn, + point_in_polygon_fn=point_in_polygon_fn, + ) + region_keys.add(region["region_key"]) + return region_keys + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "validate_regions" not in names + + def test_request_construction_is_not_flagged(self, tmp_path): + fp = tmp_path / "request_builder.py" + fp.write_text( + textwrap.dedent("""\ + def build_request(a, b, c, d, e): + return submit(Request(a=a, b=b, c=c, d=d, e=e)) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "build_request" not in names + + def test_typed_dataclass_fixture_factory_is_not_flagged(self, tmp_path): + fp = tmp_path / "runtime_hooks.py" + fp.write_text( + textwrap.dedent("""\ + from dataclasses import dataclass + + + @dataclass(frozen=True) + class AutoCoordinationRunRuntimeHooks: + command_bus: object + coordinator: object + event_store: object + logger: object + metrics: object + run_id: object + + + def make_runtime_hooks( + command_bus: object, + coordinator: object, + event_store: object, + logger: object, + metrics: object, + run_id: object, + ) -> AutoCoordinationRunRuntimeHooks: + return AutoCoordinationRunRuntimeHooks( + command_bus=command_bus, + coordinator=coordinator, + event_store=event_store, + logger=logger, + metrics=metrics, + run_id=run_id, + ) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "make_runtime_hooks" not in names + + def test_constructed_call_target_is_not_flagged(self, tmp_path): + fp = tmp_path / "request_submitter.py" + fp.write_text( + textwrap.dedent("""\ + def submit_via_request(a, b, c, d, e): + return Request(a=a, b=b, c=c, d=d, e=e).submit( + a=a, b=b, c=c, d=d, e=e, + ) + """) + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert "submit_via_request" not in names + + def test_syntax_error_file_is_skipped(self, tmp_path): + (tmp_path / "broken.py").write_text("def broken(\n") + (tmp_path / "wrapper.py").write_text( + "def wrapper(a, b, c, d, e):\n" + " return inner(a=a, b=b, c=c, d=d, e=e)\n" + ) + + names = [entry["function"] for entry in detect_passthrough_functions(tmp_path)] + + assert names == ["wrapper"] def test_non_passthrough_not_flagged(self, tmp_path): fp = tmp_path / "real.py"