diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e11c331f2..c3a1d42da6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ Full release notes with details on each version: [GitHub Releases](https://githu ## 0.9.42 (unreleased) +- Fix: Python classes now retain an `inherits [EXTRACTED]` edge to an imported base class instead of being downgraded to `uses [INFERRED]`; aliases, qualified/generic bases, package re-exports, duplicate class names, and incremental rebuilds resolve through exact import evidence without ghost nodes (#2736, thanks @NithishKumar04). - Fix: a JS/TS `for...of` / `for...in` loop binding is now shadowed, so passing it as a call argument no longer fabricates an `indirect_call` edge to an unrelated same-named callable (#2685, thanks @ousamabenyounes); completes the loop/closure/catch shadow family (#2568/#2569/#2517). - Fix: graph provenance (`built_at_commit`) is stamped from the analysed repository rather than the shell's working directory, so `graphify extract` run from elsewhere records the target's commit, not the caller's (#2534 family; #2699, thanks @C0KERNEL). - Fix: `affected` resolves a seed passed as a `./`-relative path (or an absolute path when run from the repo root) instead of silently returning nothing (#2707, thanks @phudayyy). Note: an absolute-path seed still requires the working directory to be the analysed repo root. diff --git a/graphify/extract.py b/graphify/extract.py index 6e1fca8541..0d892a5b60 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -126,6 +126,7 @@ _resolve_lua_import_target, _probe_python_module_candidate, _resolve_python_module_path, + _resolve_python_inheritance_references, _resolve_tsconfig_alias, _resolve_workspace_import, _source_key, @@ -6013,15 +6014,54 @@ def _learn(e: dict) -> None: logging.getLogger(__name__).warning( "Go type-reference resolution failed, skipping: %s", exc ) + # Resolve imported Python base classes by exact module + symbol before the + # generic unique-label rewire. The import disambiguates same-named classes, + # while resolution context keeps changed-child -> unchanged-base edges on + # incremental rebuilds (#2736). + py_paths = [p for p in paths if p.suffix == ".py"] + suppressed_python_inferred_uses: set[tuple[str, str, str, str]] = set() + if py_paths: + try: + suppressed_python_inferred_uses = _resolve_python_inheritance_references( + py_paths, + all_nodes, + all_edges, + root, + resolution_context_nodes, + resolution_context_edges, + ) + except Exception as exc: + import logging + logging.getLogger(__name__).warning( + "Python inheritance resolution failed, skipping: %s", exc + ) _rewire_unique_stub_nodes(all_nodes, all_edges) # Add cross-file class-level edges (Python only - uses Python parser internally) - py_paths = [p for p in paths if p.suffix == ".py"] if py_paths: py_results = [r for r, p in zip(per_file, paths) if p.suffix == ".py"] try: cross_file_edges = _resolve_cross_file_imports(py_results, py_paths) - all_edges.extend(cross_file_edges) + # Suppress only legacy `uses` edges emitted for import statements + # that bind an inherited base. Including the import line preserves + # legitimate uses of a same-named class imported from another module + # while still removing ambiguous or first-writer misresolution. + lookup_nodes = all_nodes + list(resolution_context_nodes or []) + labels_by_id = { + str(node.get("id")): str(node.get("label") or "") + for node in lookup_nodes + if node.get("id") + } + all_edges.extend( + edge + for edge in cross_file_edges + if ( + str(edge.get("source")), + str(edge.get("source_file") or ""), + str(edge.get("source_location") or ""), + labels_by_id.get(str(edge.get("target")), ""), + ) not in suppressed_python_inferred_uses + ) except Exception as exc: import logging logging.getLogger(__name__).warning("Cross-file import resolution failed, skipping: %s", exc) diff --git a/graphify/extractors/engine.py b/graphify/extractors/engine.py index c3b13bb737..7f5601b120 100644 --- a/graphify/extractors/engine.py +++ b/graphify/extractors/engine.py @@ -1189,6 +1189,23 @@ def walk(n) -> None: walk(root) return bound + +def _python_base_reference(node, source: bytes) -> str | None: + """Return the resolvable head of one Python class-base expression. + + Handles bare names, module-qualified names, and generic subscriptions while + deliberately ignoring keyword bases such as ``metaclass=Meta``. Keeping a + qualified name intact lets the corpus-level resolver bind its module prefix + through the exact ``import ... [as ...]`` statement (#2736). + """ + if node.type in ("identifier", "attribute"): + return _read_text(node, source) + if node.type == "subscript": + value = node.child_by_field_name("value") + if value is not None: + return _python_base_reference(value, source) + return None + _JS_SCOPE_BOUNDARY = frozenset({ "function_declaration", "function_expression", "function", "arrow_function", "method_definition", "class_declaration", "class", "generator_function", @@ -2859,8 +2876,8 @@ def walk(node, parent_class_nid: str | None = None) -> None: args = node.child_by_field_name("superclasses") if args: for arg in args.children: - if arg.type == "identifier": - base = _read_text(arg, source) + base = _python_base_reference(arg, source) + if base: base_nid = ensure_named_node(base, line) add_edge(class_nid, base_nid, "inherits", line) diff --git a/graphify/extractors/resolution.py b/graphify/extractors/resolution.py index aa57e351ec..cad7f9d0c6 100644 --- a/graphify/extractors/resolution.py +++ b/graphify/extractors/resolution.py @@ -1877,6 +1877,286 @@ def _augment_symbol_resolution_edges( _collect_python_symbol_resolution_facts(paths, root, facts) _apply_symbol_resolution_facts(paths, nodes, edges, root, facts) + +def _resolve_python_inheritance_references( + paths: list[Path], + all_nodes: list[dict], + all_edges: list[dict], + root: Path, + resolution_context_nodes: list[dict] | None = None, + resolution_context_edges: list[dict] | None = None, +) -> set[tuple[str, str, str, str]]: + """Resolve imported Python base classes to their exact definition nodes. + + The per-file extractor emits ``inherits`` to a sourceless bare-name stub. + ``_rewire_unique_stub_nodes`` can repair that only when the base name is + globally unique; aliases stay ghosted, and duplicate names are deliberately + left unresolved. A ``from module import Base [as Alias]`` statement carries + exact file + symbol evidence, so use it before the generic rewire (#2736). + + Import bindings in re-export modules are followed recursively (for example, + ``pkg.__init__`` re-exporting ``Base`` from ``pkg.base``). Unresolved and + external bases remain on their existing stubs, preserving the conservative + fallback. Resolution-context nodes/edges are lookup-only so an incremental + rebuild of a child can still target an unchanged base without copying the + base into the fresh extraction result. + + Return exact legacy import-edge keys to suppress so the later inferred-uses + pass cannot overwrite or guess an inheritance relationship. + """ + py_paths = [path for path in paths if path.suffix == ".py"] + if not py_paths: + return set() + + definition_nodes = all_nodes + list(resolution_context_nodes or []) + definition_edges = all_edges + list(resolution_context_edges or []) + contained = { + edge.get("target") + for edge in definition_edges + if edge.get("relation") == "contains" + } + + def source_path(raw: object) -> Path | None: + if not raw: + return None + path = Path(str(raw)) + if not path.is_absolute(): + path = root / path + try: + return path.resolve() + except OSError: + return path.absolute() + + # Exact (defining file, class name) -> candidate ids. Requiring one + # candidate avoids guessing when malformed/recovered syntax produced two + # definitions with the same label in one file. + definitions: dict[tuple[Path, str], set[str]] = {} + for node in definition_nodes: + nid = node.get("id") + label = str(node.get("label") or "") + path = source_path(node.get("source_file")) + if not nid or not label or path is None or path.suffix != ".py": + continue + if not node.get("_callable_class") and nid not in contained: + continue + if not _is_type_like_definition(node): + continue + definitions.setdefault((path, label), set()).add(str(nid)) + + if not definitions: + return set() + + # file -> local name -> [(import line, target file, target symbol)]. Keep + # every binding so a later import cannot retroactively resolve a class that + # appeared before it. Multiple distinct origins at/before the class are left + # unresolved rather than guessing across conditional/fallback imports. + binding_cache: dict[Path, dict[str, list[tuple[int, Path, str]]]] = {} + module_binding_cache: dict[Path, dict[str, list[tuple[int, Path]]]] = {} + + def import_bindings(path: Path) -> dict[str, list[tuple[int, Path, str]]]: + try: + resolved_path = path.resolve() + except OSError: + resolved_path = path.absolute() + cached = binding_cache.get(resolved_path) + if cached is not None: + return cached + + bindings: dict[str, list[tuple[int, Path, str]]] = {} + binding_cache[resolved_path] = bindings + module_bindings: dict[str, list[tuple[int, Path]]] = {} + module_binding_cache[resolved_path] = module_bindings + parsed = _parse_python_tree(resolved_path) + if parsed is None: + return bindings + source, root_node = parsed + + # Imports nested inside functions/classes do not bind a module-level + # class base. Module-level try/if blocks are traversed because their + # imports do bind the module namespace when that branch executes. + def walk(node) -> None: + if node is not root_node and node.type in ( + "function_definition", "class_definition", "lambda" + ): + return + if node.type == "import_from_statement": + module = _python_import_from_module(node, source) + if module is None: + return + level, module_name = module + target = _resolve_python_module_path( + module_name, resolved_path, root, level + ) + if target is None: + return + try: + target = target.resolve() + except OSError: + target = target.absolute() + line = node.start_point[0] + 1 + for imported_name, local_name in _python_imported_names(node, source): + bindings.setdefault(local_name, []).append( + (line, target, imported_name) + ) + return + if node.type == "import_statement": + line = node.start_point[0] + 1 + for child in node.children: + module_name = "" + qualifier = "" + if child.type == "dotted_name": + module_name = _read_text(child, source) + qualifier = module_name + elif child.type == "aliased_import": + name_node = child.child_by_field_name("name") + alias_node = child.child_by_field_name("alias") + if name_node is not None and alias_node is not None: + module_name = _read_text(name_node, source) + qualifier = _read_text(alias_node, source) + if not module_name or not qualifier: + continue + target = _resolve_python_module_path( + module_name, resolved_path, root, 0 + ) + if target is None: + continue + try: + target = target.resolve() + except OSError: + target = target.absolute() + module_bindings.setdefault(qualifier, []).append((line, target)) + return + for child in node.children: + walk(child) + + walk(root_node) + return bindings + + def binding_at( + path: Path, local_name: str, use_line: int | None + ) -> tuple[Path, str] | None: + candidates = import_bindings(path).get(local_name, []) + if use_line is not None: + candidates = [entry for entry in candidates if entry[0] <= use_line] + if not candidates: + return None + origins = {(entry[1], entry[2]) for entry in candidates} + if len(origins) != 1: + return None + return next(iter(origins)) + + def qualified_binding_at( + path: Path, qualified_name: str, use_line: int | None + ) -> tuple[Path, str] | None: + # The final segment is the class; everything before it is the written + # module qualifier (`mod.Base`, `pkg.mod.Base`, or an import alias). + qualifier, separator, symbol = qualified_name.rpartition(".") + if not separator or not qualifier or not symbol: + return None + import_bindings(path) # populates module_binding_cache as a side effect + try: + resolved_path = path.resolve() + except OSError: + resolved_path = path.absolute() + candidates = module_binding_cache.get(resolved_path, {}).get(qualifier, []) + if use_line is not None: + candidates = [entry for entry in candidates if entry[0] <= use_line] + targets = {entry[1] for entry in candidates} + if len(targets) != 1: + return None + return next(iter(targets)), symbol + + def resolve_origin( + path: Path, + name: str, + seen: set[tuple[Path, str]] | None = None, + ) -> str | None: + try: + path = path.resolve() + except OSError: + path = path.absolute() + key = (path, name) + candidates = definitions.get(key, set()) + if len(candidates) == 1: + return next(iter(candidates)) + if len(candidates) > 1: + return None + visited = set() if seen is None else seen + if key in visited: + return None + visited.add(key) + reexport = binding_at(path, name, None) + if reexport is None: + return None + return resolve_origin(reexport[0], reexport[1], visited) + + stub_labels = { + str(node["id"]): str(node.get("label") or "") + for node in all_nodes + if node.get("id") and not node.get("source_file") + } + if not stub_labels: + return set() + + def edge_line(edge: dict) -> int | None: + location = str(edge.get("source_location") or "") + if location.startswith("L") and location[1:].isdigit(): + return int(location[1:]) + return None + + repointed_from: set[str] = set() + suppressed_inferred_uses: set[tuple[str, str, str, str]] = set() + for edge in all_edges: + if edge.get("relation") != "inherits": + continue + target = str(edge.get("target") or "") + local_name = stub_labels.get(target) + referencing_path = source_path(edge.get("source_file")) + if not local_name or referencing_path is None: + continue + use_line = edge_line(edge) + for import_line, _, target_symbol in import_bindings(referencing_path).get( + local_name, [] + ): + if use_line is not None and import_line > use_line: + continue + suppressed_inferred_uses.add( + ( + str(edge.get("source") or ""), + str(edge.get("source_file") or ""), + f"L{import_line}", + target_symbol, + ) + ) + binding = binding_at(referencing_path, local_name, use_line) + if binding is None: + binding = qualified_binding_at( + referencing_path, local_name, use_line + ) + if binding is None: + continue + resolved = resolve_origin(binding[0], binding[1]) + if resolved is None or resolved == target: + continue + edge["target"] = resolved + repointed_from.add(target) + + if not repointed_from: + return suppressed_inferred_uses + + referenced = { + endpoint + for edge in all_edges + for endpoint in (edge.get("source"), edge.get("target")) + } + all_nodes[:] = [ + node + for node in all_nodes + if node.get("id") not in repointed_from or node.get("id") in referenced + ] + return suppressed_inferred_uses + + def _resolve_cross_file_imports( per_file: list[dict], paths: list[Path], diff --git a/tests/test_python_inheritance_resolution.py b/tests/test_python_inheritance_resolution.py new file mode 100644 index 0000000000..3fdf4bc297 --- /dev/null +++ b/tests/test_python_inheritance_resolution.py @@ -0,0 +1,302 @@ +"""Regression coverage for import-aware Python inheritance resolution (#2736).""" + +from __future__ import annotations + +from pathlib import Path + +from graphify.build import build_from_json +from graphify.extract import extract + + +def _write(path: Path, text: str) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return path + + +def _real_node(result: dict, label: str, source_suffix: str | None = None) -> dict: + matches = [ + node + for node in result["nodes"] + if node.get("label") == label + and node.get("source_file") + and ( + source_suffix is None + or str(node.get("source_file", "")).endswith(source_suffix) + ) + ] + assert len(matches) == 1, [ + (node.get("id"), node.get("source_file")) + for node in result["nodes"] + if node.get("label") == label + ] + return matches[0] + + +def _edge(result: dict, source: str, target: str, relation: str) -> list[dict]: + return [ + edge + for edge in result["edges"] + if edge.get("source") == source + and edge.get("target") == target + and edge.get("relation") == relation + ] + + +def test_cross_file_inheritance_survives_final_graph_build(tmp_path: Path) -> None: + base = _write(tmp_path / "module_a.py", "class BaseDriver:\n pass\n") + child = _write( + tmp_path / "module_b.py", + "from module_a import BaseDriver\n\n\n" + "class ChildDriver(BaseDriver):\n pass\n", + ) + + result = extract( + [base, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + base_node = _real_node(result, "BaseDriver", "module_a.py") + child_node = _real_node(result, "ChildDriver", "module_b.py") + + inheritance = _edge(result, child_node["id"], base_node["id"], "inherits") + assert len(inheritance) == 1 + assert inheritance[0]["confidence"] == "EXTRACTED" + assert inheritance[0]["source_location"] == "L4" + assert not _edge(result, child_node["id"], base_node["id"], "uses") + assert not [ + node + for node in result["nodes"] + if node.get("label") == "BaseDriver" and not node.get("source_file") + ] + + graph = build_from_json(result, root=tmp_path) + built = graph.get_edge_data(child_node["id"], base_node["id"]) + assert built is not None + assert built["relation"] == "inherits" + assert built["confidence"] == "EXTRACTED" + assert built["source_location"] == "L4" + + +def test_aliased_import_resolves_to_canonical_base_without_ghost(tmp_path: Path) -> None: + base = _write(tmp_path / "module_a.py", "class BaseDriver:\n pass\n") + child = _write( + tmp_path / "module_b.py", + "from module_a import BaseDriver as ImportedBase\n\n\n" + "class ChildDriver(ImportedBase):\n pass\n", + ) + + result = extract( + [base, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + base_node = _real_node(result, "BaseDriver", "module_a.py") + child_node = _real_node(result, "ChildDriver", "module_b.py") + + inheritance = _edge(result, child_node["id"], base_node["id"], "inherits") + assert len(inheritance) == 1 + assert inheritance[0]["source_location"] == "L4" + assert not [ + node for node in result["nodes"] if node.get("label") == "ImportedBase" + ] + assert not [ + edge + for edge in result["edges"] + if edge.get("source") == child_node["id"] + and edge.get("relation") == "uses" + and edge.get("target") == base_node["id"] + ] + + +def test_duplicate_base_names_follow_the_exact_import(tmp_path: Path) -> None: + wrong = _write( + tmp_path / "pkg_a" / "base.py", "class SharedBase:\n pass\n" + ) + right = _write( + tmp_path / "pkg_b" / "base.py", "class SharedBase:\n pass\n" + ) + child = _write( + tmp_path / "consumer.py", + "from pkg_b.base import SharedBase\n\n\n" + "class Concrete(SharedBase):\n pass\n", + ) + + result = extract( + [wrong, right, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + wrong_node = _real_node(result, "SharedBase", "pkg_a/base.py") + right_node = _real_node(result, "SharedBase", "pkg_b/base.py") + child_node = _real_node(result, "Concrete", "consumer.py") + + assert len(_edge(result, child_node["id"], right_node["id"], "inherits")) == 1 + assert not _edge(result, child_node["id"], wrong_node["id"], "inherits") + assert not _edge(result, child_node["id"], wrong_node["id"], "uses") + assert not [ + node + for node in result["nodes"] + if node.get("label") == "SharedBase" and not node.get("source_file") + ] + + +def test_same_named_non_base_import_keeps_inferred_use(tmp_path: Path) -> None: + inherited = _write(tmp_path / "inherited.py", "class Base:\n pass\n") + used = _write(tmp_path / "used.py", "class Base:\n pass\n") + child = _write( + tmp_path / "child.py", + "from inherited import Base\n" + "from used import Base as OtherBase\n\n" + "class Child(Base):\n" + " dependency: OtherBase\n", + ) + + result = extract( + [inherited, used, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + inherited_node = _real_node(result, "Base", "inherited.py") + used_node = _real_node(result, "Base", "used.py") + child_node = _real_node(result, "Child", "child.py") + + assert len(_edge(result, child_node["id"], inherited_node["id"], "inherits")) == 1 + assert not _edge(result, child_node["id"], inherited_node["id"], "uses") + uses = _edge(result, child_node["id"], used_node["id"], "uses") + assert len(uses) == 1 + assert uses[0]["source_location"] == "L2" + + +def test_inheritance_follows_relative_alias_through_package_reexport( + tmp_path: Path, +) -> None: + package = tmp_path / "pkg" + init = _write( + package / "__init__.py", + "from .base import BaseDriver as PublicBase\n", + ) + base = _write(package / "base.py", "class BaseDriver:\n pass\n") + child = _write( + tmp_path / "consumer.py", + "from pkg import PublicBase as ImportedBase\n\n\n" + "class ChildDriver(ImportedBase):\n pass\n", + ) + + result = extract( + [init, base, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + base_node = _real_node(result, "BaseDriver", "pkg/base.py") + child_node = _real_node(result, "ChildDriver", "consumer.py") + + assert len(_edge(result, child_node["id"], base_node["id"], "inherits")) == 1 + assert not [ + node + for node in result["nodes"] + if node.get("label") == "ImportedBase" and not node.get("source_file") + ] + + +def test_module_qualified_generic_base_resolves_through_import_alias( + tmp_path: Path, +) -> None: + base = _write( + tmp_path / "pkg" / "base.py", "class SharedBase:\n pass\n" + ) + child = _write( + tmp_path / "consumer.py", + "import pkg.base as model\n\n\n" + "class Concrete(model.SharedBase[int]):\n pass\n", + ) + + result = extract( + [base, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + base_node = _real_node(result, "SharedBase", "pkg/base.py") + child_node = _real_node(result, "Concrete", "consumer.py") + + inheritance = _edge(result, child_node["id"], base_node["id"], "inherits") + assert len(inheritance) == 1 + assert inheritance[0]["source_location"] == "L4" + assert not [ + node + for node in result["nodes"] + if node.get("label") == "model.SharedBase" and not node.get("source_file") + ] + + +def test_incremental_child_resolves_base_from_unchanged_context(tmp_path: Path) -> None: + base = _write(tmp_path / "base.py", "class Base:\n pass\n") + child = _write( + tmp_path / "child.py", + "from base import Base\n\n\nclass Child(Base):\n pass\n", + ) + full = extract( + [base, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + base_node = _real_node(full, "Base", "base.py") + + changed = extract( + [child], + cache_root=tmp_path, + root=tmp_path, + parallel=False, + resolution_context_nodes=full["nodes"], + resolution_context_edges=full["edges"], + ) + child_node = _real_node(changed, "Child", "child.py") + + inheritance = _edge(changed, child_node["id"], base_node["id"], "inherits") + assert len(inheritance) == 1 + assert inheritance[0]["confidence"] == "EXTRACTED" + assert not any(node.get("id") == base_node["id"] for node in changed["nodes"]) + + +def test_ambiguous_unimported_base_is_not_guessed(tmp_path: Path) -> None: + first = _write(tmp_path / "a.py", "class Base:\n pass\n") + second = _write(tmp_path / "b.py", "class Base:\n pass\n") + child = _write(tmp_path / "child.py", "class Child(Base):\n pass\n") + + result = extract( + [first, second, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + child_node = _real_node(result, "Child", "child.py") + real_base_ids = { + node["id"] + for node in result["nodes"] + if node.get("label") == "Base" and node.get("source_file") + } + + assert len(real_base_ids) == 2 + assert not [ + edge + for edge in result["edges"] + if edge.get("source") == child_node["id"] + and edge.get("relation") == "inherits" + and edge.get("target") in real_base_ids + ] + + +def test_conditional_imports_with_distinct_origins_are_not_guessed( + tmp_path: Path, +) -> None: + first = _write(tmp_path / "a.py", "class Base:\n pass\n") + second = _write(tmp_path / "b.py", "class Base:\n pass\n") + child = _write( + tmp_path / "child.py", + "try:\n" + " from a import Base\n" + "except ImportError:\n" + " from b import Base\n\n\n" + "class Child(Base):\n pass\n", + ) + + result = extract( + [first, second, child], cache_root=tmp_path, root=tmp_path, parallel=False + ) + child_node = _real_node(result, "Child", "child.py") + real_base_ids = { + node["id"] + for node in result["nodes"] + if node.get("label") == "Base" and node.get("source_file") + } + + assert not [ + edge + for edge in result["edges"] + if edge.get("source") == child_node["id"] + and edge.get("target") in real_base_ids + and edge.get("relation") in {"inherits", "uses"} + ]