diff --git a/graphify/extract.py b/graphify/extract.py index 30f31d329e..80e6daa136 100644 --- a/graphify/extract.py +++ b/graphify/extract.py @@ -2418,7 +2418,7 @@ def _emit_call(caller: str, target_nid: "str | None", rc: dict) -> None: file_aliases = import_alias_by_filenode.get(caller_file, {}) mods = [t for t in imported_by_filenode.get(caller_file, ()) if t in contains_children - and (_module_stem_key(t) == rkey or file_aliases.get(t) == rkey)] + and (file_aliases.get(t) == rkey if t in file_aliases else _module_stem_key(t) == rkey)] if len(mods) != 1: # not an imported module, or ambiguous -> bail continue children = contains_children[mods[0]].get(_key(callee), []) diff --git a/tests/test_extract.py b/tests/test_extract.py index da7a615f85..a5bf9e090c 100644 --- a/tests/test_extract.py +++ b/tests/test_extract.py @@ -1406,6 +1406,56 @@ def _alias_edges(result): assert len(_alias_edges(warm)) == 1, "aliased call edge vanished on warm cache (#2082)" +def test_python_aliased_import_stem_collision_resolution(tmp_path): + """Verify Issue #2428: two imported modules share the same filename stem ('service'), + with one aliased and one unaliased. The call to the unaliased module must resolve + and must not be blocked by a false ambiguity with the aliased module.""" + pkg = tmp_path / "pkg" + pkg.mkdir() + (pkg / "__init__.py").write_text("") + (pkg / "service.py").write_text("def register_user(name):\n return {'name': name}\n") + + other = tmp_path / "other" + other.mkdir() + (other / "__init__.py").write_text("") + (other / "service.py").write_text("def ping():\n return 'pong'\n") + + api = pkg / "api.py" + api.write_text( + "from pkg import service\n" + "from other import service as other_service\n\n" + "def register(payload):\n" + " other_service.ping()\n" + " return service.register_user(payload['name'])\n" + ) + result = extract( + [api, pkg / "service.py", other / "service.py", pkg / "__init__.py", other / "__init__.py"], + cache_root=tmp_path, + root=tmp_path, + ) + nodes = {n["id"]: n for n in result["nodes"]} + + # Verify call to other_service.ping() resolves to other/service.py -> ping() + ping_calls = [ + e for e in result["edges"] + if e["relation"] == "calls" + and "register" in nodes[e["source"]]["label"] + and "ping" in nodes[e["target"]]["label"] + and "other/service.py" in (nodes[e["target"]].get("source_file") or "") + ] + assert len(ping_calls) == 1, f"expected call to other_service.ping to resolve, got: {ping_calls}" + + # Verify call to service.register_user() resolves to pkg/service.py -> register_user() + register_user_calls = [ + e for e in result["edges"] + if e["relation"] == "calls" + and "register" in nodes[e["source"]]["label"] + and "register_user" in nodes[e["target"]]["label"] + and "pkg/service.py" in (nodes[e["target"]].get("source_file") or "") + ] + assert len(register_user_calls) == 1, f"expected call to service.register_user to resolve, got: {register_user_calls}" + + def test_python_qualified_call_resolves_when_method_name_collides_with_caller(tmp_path): """The real #1446 shape: a viewset action `approve()` delegates to a SERVICE action of the SAME name via `Service.approve()`. The bare-name in-file lookup