diff --git a/desloppify/languages/python/detectors/deps_dynamic.py b/desloppify/languages/python/detectors/deps_dynamic.py index a28ca5562..3bb948607 100644 --- a/desloppify/languages/python/detectors/deps_dynamic.py +++ b/desloppify/languages/python/detectors/deps_dynamic.py @@ -6,13 +6,208 @@ import logging from pathlib import Path -from .deps_resolution import resolve_absolute_import +from .deps_resolution import resolve_absolute_import, try_resolve_path logger = logging.getLogger(__name__) +def _string_literal(node: ast.AST) -> str | None: + """Return a string constant without evaluating arbitrary source expressions.""" + + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + return None + + +def _import_module_bindings(tree: ast.AST) -> tuple[set[str], set[str]]: + """Return local names bound to the importlib module and import_module function.""" + + module_bindings: set[str] = set() + function_bindings: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.name == "importlib" or alias.name.startswith("importlib."): + module_bindings.add(alias.asname or alias.name.partition(".")[0]) + elif isinstance(node, ast.ImportFrom) and node.module == "importlib": + for alias in node.names: + if alias.name == "import_module": + function_bindings.add(alias.asname or alias.name) + return module_bindings, function_bindings + + +def _is_import_module_call( + node: ast.Call, + *, + module_bindings: set[str], + function_bindings: set[str], +) -> bool: + """Return whether *node* is a call through a known importlib binding.""" + + func = node.func + return bool( + node.args + and ( + ( + isinstance(func, ast.Attribute) + and func.attr == "import_module" + and isinstance(func.value, ast.Name) + and func.value.id in module_bindings + ) + or (isinstance(func, ast.Name) and func.id in function_bindings) + ) + ) + + +def _mapping_module_specs(value: ast.AST) -> set[str]: + """Extract first tuple/list members from a static lazy-export mapping.""" + + if not isinstance(value, ast.Dict): + return set() + + module_specs: set[str] = set() + for entry in value.values: + if isinstance(entry, ast.Tuple | ast.List) and entry.elts: + spec = _string_literal(entry.elts[0]) + else: + spec = _string_literal(entry) + if spec: + module_specs.add(spec) + return module_specs + + +def _lazy_export_mappings(tree: ast.Module) -> dict[str, set[str]]: + """Collect module-level literal mappings used by lazy package exports.""" + + mappings: dict[str, set[str]] = {} + for node in tree.body: + if isinstance(node, ast.Assign): + targets = node.targets + value = node.value + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets = [node.target] + value = node.value + else: + continue + module_specs = _mapping_module_specs(value) + if not module_specs: + continue + for target in targets: + if isinstance(target, ast.Name): + mappings[target.id] = module_specs + return mappings + + +def _unpacked_mapping_binding( + node: ast.AST, + mappings: dict[str, set[str]], +) -> tuple[str, str] | None: + """Return ``(local_name, mapping_name)`` for ``name, _ = MAPPING[...]``.""" + + if isinstance(node, ast.Assign): + targets = node.targets + value = node.value + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets = [node.target] + value = node.value + else: + return None + if not ( + isinstance(value, ast.Subscript) + and isinstance(value.value, ast.Name) + and value.value.id in mappings + ): + return None + for target in targets: + if isinstance(target, ast.Tuple | ast.List) and target.elts: + first = target.elts[0] + if isinstance(first, ast.Name): + return first.id, value.value.id + return None + + +def _lazy_export_variable(argument: ast.AST) -> str | None: + """Return the selected module variable for ``f\"{__name__}.{module}\"``.""" + + if not isinstance(argument, ast.JoinedStr) or len(argument.values) != 3: + return None + package, separator, module = argument.values + if not ( + isinstance(package, ast.FormattedValue) + and isinstance(package.value, ast.Name) + and package.value.id == "__name__" + and _string_literal(separator) == "." + and isinstance(module, ast.FormattedValue) + and isinstance(module.value, ast.Name) + ): + return None + return module.value.id + + +def _lazy_export_targets( + tree: ast.Module, + py_file: Path, + *, + module_bindings: set[str], + function_bindings: set[str], +) -> set[str]: + """Resolve child modules selected from literal package lazy-export mappings.""" + + mappings = _lazy_export_mappings(tree) + if not mappings: + return set() + + targets: set[str] = set() + for scope in ast.walk(tree): + if not isinstance(scope, ast.FunctionDef | ast.AsyncFunctionDef): + continue + bindings = { + binding[0]: binding[1] + for node in ast.walk(scope) + if (binding := _unpacked_mapping_binding(node, mappings)) is not None + } + if not bindings: + continue + for node in ast.walk(scope): + if not isinstance(node, ast.Call) or not _is_import_module_call( + node, + module_bindings=module_bindings, + function_bindings=function_bindings, + ): + continue + module_variable = _lazy_export_variable(node.args[0]) + mapping_name = bindings.get(module_variable) + if mapping_name is None: + continue + for module_spec in mappings[mapping_name]: + candidate = py_file.parent.joinpath(*module_spec.split(".")) + resolved = try_resolve_path(candidate) + if resolved: + targets.add(resolved) + return targets + + +def _is_explicit_legacy_module_alias(node: ast.Call) -> bool: + """Recognize physical ``install_legacy_module_alias(__name__, target)`` wrappers.""" + + func = node.func + is_alias_installer = ( + isinstance(func, ast.Name) and func.id == "install_legacy_module_alias" + ) or ( + isinstance(func, ast.Attribute) and func.attr == "install_legacy_module_alias" + ) + return bool( + is_alias_installer + and len(node.args) >= 2 + and isinstance(node.args[0], ast.Name) + and node.args[0].id == "__name__" + and _string_literal(node.args[1]) + ) + + def find_python_dynamic_imports(path: Path, extensions: list[str]) -> set[str]: - """Find module specifiers referenced by ``importlib.import_module`` calls.""" + """Find module files entered through dynamic imports or explicit alias wrappers.""" + del extensions targets: set[str] = set() for py_file in path.rglob("*.py"): @@ -25,27 +220,39 @@ def find_python_dynamic_imports(path: Path, extensions: list[str]) -> set[str]: exc, ) continue + module_bindings, function_bindings = _import_module_bindings(tree) + for node in tree.body: + if ( + isinstance(node, ast.Expr) + and isinstance(node.value, ast.Call) + and _is_explicit_legacy_module_alias(node.value) + ): + targets.add(str(py_file.resolve())) for node in ast.walk(tree): if not isinstance(node, ast.Call): continue - func = node.func - if not ( - isinstance(func, ast.Attribute) - and func.attr == "import_module" - and isinstance(func.value, ast.Name) - and func.value.id == "importlib" - and node.args - and isinstance(node.args[0], ast.Constant) - and isinstance(node.args[0].value, str) + if not _is_import_module_call( + node, + module_bindings=module_bindings, + function_bindings=function_bindings, ): continue - - spec = node.args[0].value + spec = _string_literal(node.args[0]) + if spec is None: + continue resolved = resolve_absolute_import(spec, path) if resolved: targets.add(resolved) else: targets.add(spec) + targets.update( + _lazy_export_targets( + tree, + py_file, + module_bindings=module_bindings, + function_bindings=function_bindings, + ) + ) return targets diff --git a/desloppify/languages/python/tests/test_py_deps.py b/desloppify/languages/python/tests/test_py_deps.py index 7ee69466a..09a197f20 100644 --- a/desloppify/languages/python/tests/test_py_deps.py +++ b/desloppify/languages/python/tests/test_py_deps.py @@ -344,11 +344,73 @@ def test_finds_importlib_import_module(self, tmp_path): assert len(targets) >= 1 # The raw specifier should match if resolution fails, # or a resolved path ending in auth.py if it succeeds - found = any( - "auth" in t for t in targets - ) + found = any("auth" in t for t in targets) assert found, f"Expected 'auth' in targets, got {targets}" + def test_finds_from_importlib_import_module(self, tmp_path): + """A directly imported import_module function remains a dynamic import.""" + pkg = _make_pkg( + tmp_path, + { + "__init__.py": "", + "loader.py": textwrap.dedent("""\ + from importlib import import_module + mod = import_module("mypkg.plugins.auth") + """), + "plugins/__init__.py": "", + "plugins/auth.py": "x = 1\n", + }, + ) + + targets = find_python_dynamic_imports(pkg, [".py"]) + + assert any("auth" in target for target in targets) + + def test_finds_literal_lazy_export_mapping_targets(self, tmp_path): + """Lazy package __getattr__ mappings keep their child modules reachable.""" + pkg = _make_pkg( + tmp_path, + { + "__init__.py": textwrap.dedent("""\ + from importlib import import_module + + _LAZY_EXPORTS: dict[str, tuple[str, str]] = { + "Service": ("services.service", "Service"), + } + + def __getattr__(name: str): + module_name, attr_name = _LAZY_EXPORTS[name] + module = import_module(f"{__name__}.{module_name}") + return getattr(module, attr_name) + """), + "services/__init__.py": "", + "services/service.py": "class Service: pass\n", + }, + ) + + targets = find_python_dynamic_imports(pkg, [".py"]) + + assert str((pkg / "services" / "service.py").resolve()) in targets + + def test_finds_explicit_legacy_module_alias_wrapper(self, tmp_path): + """Physical aliases are import entry points even with no first-party callers.""" + pkg = _make_pkg( + tmp_path, + { + "__init__.py": "", + "legacy.py": textwrap.dedent("""\ + from mypkg.compat import install_legacy_module_alias + + install_legacy_module_alias(__name__, "mypkg.canonical") + """), + "canonical.py": "VALUE = 1\n", + }, + ) + + targets = find_python_dynamic_imports(pkg, [".py"]) + + assert str((pkg / "legacy.py").resolve()) in targets + def test_ignores_non_string_args(self, tmp_path): """importlib.import_module(variable) should NOT be found.""" pkg = _make_pkg( diff --git a/desloppify/tests/detectors/test_orphaned.py b/desloppify/tests/detectors/test_orphaned.py index dff923f1b..768f05964 100644 --- a/desloppify/tests/detectors/test_orphaned.py +++ b/desloppify/tests/detectors/test_orphaned.py @@ -13,6 +13,9 @@ _is_nextjs_convention_entry, detect_orphaned_files, ) +from desloppify.languages.python.detectors.deps_dynamic import ( + find_python_dynamic_imports, +) # --------------------------------------------------------------------------- # Helpers @@ -250,6 +253,61 @@ def mock_dynamic_finder(path, extensions): assert len(entries) == 1 assert entries[0]["file"] == str(f2) + def test_explicit_legacy_module_alias_not_orphaned(self, tmp_path): + """A physical compatibility alias remains an import entry point.""" + alias = tmp_path / "legacy.py" + alias.write_text( + "from pkg.compat import install_legacy_module_alias\n" + "install_legacy_module_alias(__name__, 'pkg.canonical')\n" + ) + graph = {str(alias): _graph_entry(importer_count=0)} + + with patch( + "desloppify.engine.detectors.orphaned.rel", + side_effect=lambda p: str(Path(p).relative_to(tmp_path)), + ): + entries, _ = detect_orphaned_files( + tmp_path, + graph, + [".py"], + options=OrphanedDetectionOptions( + dynamic_import_finder=find_python_dynamic_imports + ), + ) + + assert entries == [] + + def test_literal_lazy_package_export_not_orphaned(self, tmp_path): + """A static lazy-export table keeps its selected child module reachable.""" + package = tmp_path / "auth" + package.mkdir() + (package / "__init__.py").write_text( + "from importlib import import_module\n" + "_LAZY_EXPORTS = {'Service': ('service', 'Service')}\n" + "def __getattr__(name):\n" + " module_name, attr_name = _LAZY_EXPORTS[name]\n" + " module = import_module(f'{__name__}.{module_name}')\n" + " return getattr(module, attr_name)\n" + ) + service = package / "service.py" + service.write_text("class Service:\n pass\n") + graph = {str(service): _graph_entry(importer_count=0)} + + with patch( + "desloppify.engine.detectors.orphaned.rel", + side_effect=lambda p: str(Path(p).relative_to(tmp_path)), + ): + entries, _ = detect_orphaned_files( + tmp_path, + graph, + [".py"], + options=OrphanedDetectionOptions( + dynamic_import_finder=find_python_dynamic_imports + ), + ) + + assert entries == [] + def test_results_sorted_by_loc_descending(self, tmp_path): """Results are sorted by LOC descending (largest files first).""" f_small = _write_file(tmp_path / "small.py", lines=20) @@ -414,8 +472,7 @@ def test_dunder_all_file_not_orphaned(self, tmp_path): """Files defining __all__ are public API surfaces and not orphaned.""" api_file = tmp_path / "api.py" api_file.write_text( - "__all__ = ['Foo', 'Bar']\n" - + "\n".join(f"line {i}" for i in range(30)) + "__all__ = ['Foo', 'Bar']\n" + "\n".join(f"line {i}" for i in range(30)) ) orphan_file = _write_file(tmp_path / "orphan.py", lines=30) @@ -438,8 +495,7 @@ def test_dunder_all_with_type_annotation(self, tmp_path): """Files using ``__all__: list[str] = [...]`` syntax are also excluded.""" api_file = tmp_path / "api.py" api_file.write_text( - "__all__: list[str] = ['Foo']\n" - + "\n".join(f"line {i}" for i in range(30)) + "__all__: list[str] = ['Foo']\n" + "\n".join(f"line {i}" for i in range(30)) ) graph = {