feat(python): resolve self/cls/super() calls to a base-class method in another file

graphify/cache.py

@@ -36,7 +36,7 @@
     _EXTRACTOR_VERSION = "unknown"
 
 # Bump when AST cache-key semantics change independently of the package version.
-_AST_CACHE_SCHEMA = 6  # Python typed-receiver facts (receiver_type) in persisted raw calls.
+_AST_CACHE_SCHEMA = 7  # Python opaque-base marker and super(args) flag on raw calls.
 
 # Version dirs already swept this process — cleanup runs once per (base, version).
 _cleaned_ast_dirs: set[str] = set()

graphify/cli.py

@@ -4018,6 +4018,7 @@ def _ctx_identity(source_file) -> str | None:
                                     "ruby_resolution_schema",
                                     "ruby_method_kind",
                                     "ruby_lookup_unsafe",
+                                    "python_opaque_bases",
                                     "ruby_reopened",
                                     "ruby_external_method_owners",
                                     # Erlang remote-call resolution keys (#3714):

graphify/extract.py

@@ -4277,6 +4277,58 @@ def _emit_call(
             "weight": 1.0,
         })
 
+    # A changed-files rebuild resolves only its own batch, so a base class that lives
+    # in an unchanged file is still a name stub here, while a full build has already
+    # rewired it to the unique definition. Run that same rewire on COPIES of the
+    # inherits edges so both builds walk the same chain; the real edges are untouched.
+    inherits_edges = [dict(e) for e in all_edges if e.get("relation") == "inherits"]
+    if any(not node_by_id.get(str(e.get("target")), {}).get("source_file") for e in inherits_edges):
+        _rewire_unique_stub_nodes(list(all_nodes), inherits_edges)
+    bases_of: dict[str, list[str]] = {}
+    for e in inherits_edges:
+        src, tgt = e.get("source"), e.get("target")
+        if isinstance(src, str) and isinstance(tgt, str):
+            bases_of.setdefault(src, [])
+            if tgt not in bases_of[src]:
+                bases_of[src].append(tgt)
+
+    def _inherited_method(caller: str, callee: str, skip_own: bool) -> "str | None":
+        # `self.m()` / `cls.m()` / `super().m()` whose `m` lives on a base class in
+        # another file: the in-file pass only walks bases declared in the same file.
+        # Walk a single-inheritance chain only; stop (no edge) at a class with two
+        # or more bases (MRO order across files is not modelled), a base the edge
+        # list cannot show, or a base outside the corpus. The caller's own base must
+        # also be defined in or imported by the caller's file. Further hops trust the
+        # `inherits` edges: a changed-files rebuild carries an unchanged class's
+        # inherits edges but not its imports, and both builds must walk alike.
+        # The first class owning `m` wins, as in the MRO.
+        cls = method_class.get(caller)
+        seen: set[str] = set()
+        first_hop = True
+        while cls and cls not in seen:
+            seen.add(cls)
+            if not skip_own:
+                hit = method_index.get((cls, _key(callee)))
+                if hit:
+                    return hit
+            skip_own = False
+            if (node_by_id.get(cls, {}).get("metadata") or {}).get("python_opaque_bases"):
+                return None
+            bases = bases_of.get(cls, [])
+            if len(bases) != 1 or bases[0] == cls:
+                return None
+            base = bases[0]
+            cls_file, base_file = _file_node(cls), _file_node(base)
+            if cls_file is None or base_file is None or not node_by_id.get(base, {}).get("source_file"):
+                return None
+            if first_hop:
+                imported = imported_by_filenode.get(cls_file, set())
+                if not (base_file == cls_file or base in imported or base_file in imported):
+                    return None
+            first_hop = False
+            cls = base
+        return None
+
     for rc in all_raw_calls:
         if rc.get("_ambiguous_python_import"):
             continue
@@ -4285,6 +4337,12 @@ def _emit_call(
         receiver = rc.get("receiver")
         callee = rc.get("callee")
         caller = rc.get("caller_nid")
+        if receiver in ("self", "cls", "super") and callee and caller and not rc.get("_python_super_args"):
+            target = _inherited_method(caller, callee, skip_own=receiver == "super")
+            if target:
+                # Same rule as the in-file base walk (_self_call_target), across files.
+                _emit_call(caller, target, rc)
+                continue
         attr_type = rc.get("_python_self_attr_type")
         if attr_type and callee and caller:
             # `self.X.m()` with X bound to one class (#2860): only a unique class

graphify/extractors/engine.py

@@ -4811,6 +4811,18 @@ def walk(node, parent_class_nid: str | None = None) -> None:
                         ruby_body, source
                     ):
                         metadata["ruby_lookup_unsafe"] = True
+            if config.ts_module == "tree_sitter_python":
+                # Only a bare-name base becomes an `inherits` edge (below). A base
+                # the edge list cannot show (`mod.Base`, `Generic[T]`, a call) still
+                # sits in the MRO, so the cross-file inherited-method walk must not
+                # read this class's edges as its complete base list.
+                _py_bases = node.child_by_field_name("superclasses")
+                if _py_bases is not None and any(
+                    c.is_named and c.type not in ("identifier", "keyword_argument", "comment")
+                    for c in _py_bases.children
+                ):
+                    metadata = dict(metadata or {})
+                    metadata["python_opaque_bases"] = True
             add_node(class_nid, class_name, line, metadata=metadata)
             if config.ts_module == "tree_sitter_ruby" and metadata:
                 # Reopened declarations collapse onto the same file-local id;
@@ -7857,6 +7869,14 @@ def walk_calls(
                             _py_type = python_var_types.get(caller_nid, {}).get(member_receiver)
                             if _py_type:
                                 rc_entry["receiver_type"] = _py_type
+                        # `super(Cls, obj).m()` starts the MRO after Cls, not after the
+                        # caller's class; only bare `super()` may walk the caller's bases.
+                        if member_receiver == "super" and config.ts_module == "tree_sitter_python":
+                            _sfn = node.child_by_field_name("function")
+                            _sobj = _sfn.child_by_field_name("object") if _sfn is not None else None
+                            _sargs = _sobj.child_by_field_name("arguments") if _sobj is not None else None
+                            if _sargs is not None and any(c.is_named for c in _sargs.children):
+                                rc_entry["_python_super_args"] = True
                         # Tag the C++ raw_call's language so the cross-file C++ resolver
                         # claims it unambiguously: a `.h` file routes to extract_cpp or
                         # extract_objc by content, and both resolvers see `.h` in their

graphify/watch.py

@@ -1923,6 +1923,7 @@ def _add_deleted_source(path: Path) -> None:
                                 "ruby_resolution_schema",
                                 "ruby_method_kind",
                                 "ruby_lookup_unsafe",
+                                "python_opaque_bases",
                                 "ruby_reopened",
                                 "ruby_external_method_owners",
                                 # Erlang remote-call resolution keys (#3714): an

tests/test_python_inherited_method_calls.py

@@ -0,0 +1,211 @@
+"""Python `self.m()` / `cls.m()` / `super().m()` where `m` is defined on a base class in
+another file.
+
+The in-file pass (`_self_call_target`) already walks base classes declared in the same
+file. A base imported from another module was invisible to it, so `Flask.__init__ ->
+App.__init__` (flask/app.py -> flask/sansio/app.py) and every other inherited call
+across modules had no `calls` edge.
+
+`_resolve_python_member_calls` now walks the caller's class up a single-inheritance chain
+across files and binds the first class that owns `m`. It stops, with no edge, at a class
+with two or more bases (MRO order across files is not modelled), a base the edge list
+cannot show (`mod.Base`, `Generic[T]`), a base outside the corpus, a base the class's
+file neither defines nor imports, and `super(Cls, obj)` with arguments.
+"""
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+from graphify.extract import extract
+
+_BASE = "class Base:\n    def m(self):\n        return 1\n"
+
+
+def _write(root: Path, files: dict[str, str]) -> list[Path]:
+    files = {"pkg/__init__.py": "", **{f"pkg/{n}": b for n, b in files.items()}}
+    paths = []
+    for name, body in files.items():
+        p = root / name
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(body, encoding="utf-8")
+        paths.append(p)
+    return paths
+
+
+def _calls(tmp_path, files: dict[str, str]):
+    r = extract(_write(tmp_path, files), cache_root=tmp_path / "graphify-out", parallel=False)
+    nodes = {n["id"]: n for n in r["nodes"]}
+    edges = [e for e in r["edges"] if e["relation"] == "calls"]
+    return nodes, edges
+
+
+def _hits(nodes, edges, caller: str, callee: str = ".m()") -> list[str]:
+    """Stems of the files whose `callee` the method `caller` calls."""
+    return sorted(
+        Path(nodes[e["target"]]["source_file"]).stem
+        for e in edges
+        if nodes[e["source"]]["label"] == f".{caller}()" and nodes[e["target"]]["label"] == callee
+    )
+
+
+def test_self_call_reaches_a_base_method_in_another_file(tmp_path):
+    nodes, edges = _calls(tmp_path, {
+        "base.py": _BASE,
+        "use.py": "from .base import Base\n\nclass C(Base):\n    def run(self):\n        return self.m()\n",
+    })
+    assert _hits(nodes, edges, "run") == ["base"]
+    edge = next(e for e in edges if nodes[e["source"]]["label"] == ".run()")
+    assert edge["confidence"] == "EXTRACTED"  # same rule as the in-file base walk
+
+
+def test_super_and_cls_calls_reach_the_base(tmp_path):
+    nodes, edges = _calls(tmp_path, {
+        "base.py": _BASE + "    @classmethod\n    def build(cls):\n        return cls()\n",
+        "use.py": (
+            "from .base import Base\n\nclass C(Base):\n"
+            "    def m(self):\n        return super().m()\n"
+            "    @classmethod\n    def make(cls):\n        return cls.build()\n"
+        ),
+    })
+    assert _hits(nodes, edges, "m") == ["base"]
+    assert _hits(nodes, edges, "make", ".build()") == ["base"]
+
+
+def test_the_nearest_definition_wins_across_two_levels(tmp_path):
+    nodes, edges = _calls(tmp_path, {
+        "a.py": _BASE + "    def n(self):\n        return 2\n",
+        "b.py": "from .a import Base\n\nclass Mid(Base):\n    def m(self):\n        return 3\n",
+        "use.py": (
+            "from .b import Mid\n\nclass C(Mid):\n"
+            "    def run(self):\n        return self.m()\n"
+            "    def other(self):\n        return self.n()\n"
+        ),
+    })
+    assert _hits(nodes, edges, "run") == ["b"]  # Mid.m overrides Base.m
+    assert _hits(nodes, edges, "other", ".n()") == ["a"]  # through Mid, which has no n
+
+
+def test_chains_the_walk_cannot_order_emit_no_edge(tmp_path):
+    nodes, edges = _calls(tmp_path, {
+        "base.py": _BASE,
+        "other.py": "class Other:\n    def m(self):\n        return 2\n",
+        "use.py": (
+            "import lib\nfrom typing import Generic, TypeVar\n"
+            "from .base import Base\nfrom .other import Other\nT = TypeVar('T')\n\n"
+            "class TwoBases(Base, Other):\n    def run(self):\n        return self.m()\n\n"
+            "class Opaque(lib.Thing, Base):\n    def run(self):\n        return self.m()\n\n"
+            "class Generic_(Base, Generic[T]):\n    def run(self):\n        return self.m()\n\n"
+            "class SuperArgs(Base):\n    def m(self):\n        return super(Base, self).m()\n\n"
+            "class Missing(Base):\n    def run(self):\n        return self.zzz()\n\n"
+            "class Nested(Base):\n    def run(self):\n        def inner():\n            return self.m()\n"
+            "        return inner\n"
+        ),
+    })
+    calls_into = {(nodes[e["source"]]["label"], Path(nodes[e["target"]]["source_file"]).stem)
+                  for e in edges if nodes[e["target"]]["label"] in (".m()", ".zzz()")}
+    assert not {c for c in calls_into if c[1] in ("base", "other")}, calls_into
+
+
+def test_a_base_from_outside_or_not_imported_emits_no_edge(tmp_path):
+    nodes, edges = _calls(tmp_path, {
+        "base.py": _BASE,
+        "external.py": "from lib import Base\n\nclass C(Base):\n    def run(self):\n        return self.m()\n",
+        "unimported.py": "class D(Base):\n    def run(self):\n        return self.m()\n",
+    })
+    assert _hits(nodes, edges, "run") == []
+
+
+def _rebuild_calls(root: Path) -> set[tuple[str, str]]:
+    g = json.loads((root / "graphify-out" / "graph.json").read_text(encoding="utf-8"))
+    return {(e["source"], e["target"]) for e in g.get("links", g.get("edges", [])) if e["relation"] == "calls"}
+
+
+_CHAIN = {
+    "base.py": _BASE,
+    "mid.py": "import lib\nfrom .base import Base\n\nclass Mid(lib.Thing, Base):\n    pass\n",
+    "child.py": "from .base import Base\n\nclass Child(Base):\n    def run(self):\n        return self.m()\n",
+    "grand.py": "from .mid import Mid\n\nclass Grand(Mid):\n    def run(self):\n        return self.m()\n",
+    # Two hops through a class in an unchanged file: Deep -> Plain -> Base.
+    "plain.py": "from .base import Base\n\nclass Plain(Base):\n    pass\n",
+    "deep.py": "from .plain import Plain\n\nclass Deep(Plain):\n    def run(self):\n        return self.m()\n",
+}
+_CHANGED = ("child.py", "grand.py", "deep.py")
+
+
+def test_changed_files_rebuild_matches_a_full_build(tmp_path):
+    """The changed file's base sits in an unchanged file (a name stub in that batch),
+    and an unchanged class's opaque base must still stop the walk."""
+    from graphify.watch import _rebuild_code
+
+    inc, full = tmp_path / "inc", tmp_path / "full"
+    for root in (inc, full):
+        _write(root, _CHAIN)
+    assert _rebuild_code(inc, no_cluster=True, acquire_lock=False, force=True)
+    for root in (inc, full):
+        for name in _CHANGED:
+            p = root / "pkg" / name
+            p.write_text(p.read_text(encoding="utf-8") + "\n# touched\n", encoding="utf-8")
+    assert _rebuild_code(
+        inc, changed_paths=[inc / "pkg" / name for name in _CHANGED],
+        no_cluster=True, acquire_lock=False,
+    )
+    assert _rebuild_code(full, no_cluster=True, acquire_lock=False, force=True)
+    assert _rebuild_calls(inc) == _rebuild_calls(full)
+    assert ("pkg_child_child_run", "pkg_base_base_m") in _rebuild_calls(full)
+    assert ("pkg_deep_deep_run", "pkg_base_base_m") in _rebuild_calls(full)
+    assert not any(t == "pkg_base_base_m" and s.startswith("pkg_grand") for s, t in _rebuild_calls(full))
+
+
+def test_incremental_extract_matches_a_full_extract(tmp_path):
+    env = dict(os.environ, PYTHONHASHSEED="0")
+    inc, full = tmp_path / "inc", tmp_path / "full"
+    for root in (inc, full):
+        _write(root, _CHAIN)
+
+    def run(root: Path) -> None:
+        r = subprocess.run([sys.executable, "-m", "graphify", "extract", str(root), "--code-only", "--no-cluster"],
+                           cwd=root, env=env, capture_output=True, text=True)
+        assert r.returncode == 0, r.stderr[-2000:]
+
+    run(inc)
+    for root in (inc, full):
+        for name in _CHANGED:
+            p = root / "pkg" / name
+            p.write_text(p.read_text(encoding="utf-8") + "\n# touched\n", encoding="utf-8")
+    run(inc)
+    run(full)
+    assert _rebuild_calls(inc) == _rebuild_calls(full)
+    assert ("pkg_child_child_run", "pkg_base_base_m") in _rebuild_calls(full)
+    assert ("pkg_deep_deep_run", "pkg_base_base_m") in _rebuild_calls(full)
+    assert not any(t == "pkg_base_base_m" and s.startswith("pkg_grand") for s, t in _rebuild_calls(full))
+
+
+def test_schema_bump_retires_entries_cached_without_the_opaque_base_marker(tmp_path, monkeypatch):
+    """An entry from schema 6 has no `python_opaque_bases` marker; replaying it would
+    walk through `Mid(lib.Thing, Base)` and bind `Grand.run -> Base.m`."""
+    import graphify.cache as cache_mod
+    from graphify.cache import save_cached
+    from graphify.extract import extract_python
+
+    paths = _write(tmp_path, _CHAIN)
+    current_schema = cache_mod._AST_CACHE_SCHEMA
+    monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "same-version")
+    monkeypatch.setattr(cache_mod, "_AST_CACHE_SCHEMA", 6)  # last schema without the marker
+    monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set())
+    for p in paths:
+        stale = extract_python(p)
+        for node in stale["nodes"]:
+            (node.get("metadata") or {}).pop("python_opaque_bases", None)
+        save_cached(p, stale, root=tmp_path, cache_root=tmp_path, kind="ast")
+
+    monkeypatch.setattr(cache_mod, "_AST_CACHE_SCHEMA", current_schema)
+    monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set())
+    r = extract(paths, root=tmp_path, cache_root=tmp_path, parallel=False)
+    nodes = {n["id"]: n for n in r["nodes"]}
+    assert not [e for e in r["edges"] if e["relation"] == "calls"
+                and nodes[e["source"]]["label"] == ".run()" and nodes[e["source"]]["source_file"].endswith("grand.py")
+                and nodes[e["target"]]["label"] == ".m()"]