Fix loose sibling module import resolution

graphify/extract.py

@@ -266,6 +266,111 @@ def _repoint_python_package_imports(paths, all_nodes, all_edges, root) -> None:
                 e["target"] = alias_map[tgt]
 
 
+def _repoint_python_sibling_imports(paths, all_nodes, all_edges, root) -> None:
+    """Repoint Python sibling-import edges to the real file node in directories
+    without an __init__.py (#3430).
+
+    When a Python file below the scan root lives in a non-package directory
+    (no __init__.py in its immediate parent), plain imports of same-directory
+    modules (e.g. `scripts/main.py: import greeter`) target a bare name (`greeter`),
+    while the real file node is scan-root-relative (`scripts_greeter`). Because
+    the directory is not a package, `_repoint_python_package_imports` skips it
+    (levels == 0), leaving the edge dangling and causing downstream member calls
+    (`greeter.greet()`) to be dropped.
+
+    This pass is strictly importer-directory-local:
+    - Only applies when the importing file's parent directory has no __init__.py.
+    - Resolves only to unambiguous same-directory candidate modules in the scanned corpus.
+    - Never builds a global alias map and never searches outside the importer's directory.
+    - Preserves local aliases (e.g. `import greeter as g`).
+    """
+    try:
+        root = Path(root).resolve()
+    except OSError:
+        root = Path(root)
+
+    node_ids = {n.get("id") for n in all_nodes if isinstance(n, dict)}
+
+    # Index corpus modules by directory for non-package directories only.
+    # dir_path -> {module_name_id: set_of_file_node_ids}
+    dir_siblings: dict[Path, dict[str, set[str]]] = {}
+    for p in paths:
+        if p.suffix.lower() not in (".py", ".pyi"):
+            continue
+        try:
+            p_res = Path(p).resolve()
+            rel = p_res.relative_to(root)
+        except (ValueError, OSError):
+            continue
+
+        file_node = _file_node_id(rel)
+        if file_node not in node_ids:
+            continue
+
+        if p_res.name in ("__init__.py", "__init__.pyi"):
+            # Sibling package directory inside parent_dir (parent_dir / subpkg / __init__.py)
+            pkg_dir = p_res.parent
+            parent_dir = pkg_dir.parent
+            if not (parent_dir / "__init__.py").is_file() and not (parent_dir / "__init__.pyi").is_file():
+                mod_key = _make_id(pkg_dir.name)
+                dir_siblings.setdefault(parent_dir, {}).setdefault(mod_key, set()).add(file_node)
+            continue
+
+        d = p_res.parent
+        # PEP 328 guard: if the directory is a package, implicit relative imports
+        # are forbidden in Python 3. Do not index packages as loose sibling directories.
+        if (d / "__init__.py").is_file() or (d / "__init__.pyi").is_file():
+            continue
+
+        mod_key = _make_id(p_res.stem)
+        dir_siblings.setdefault(d, {}).setdefault(mod_key, set()).add(file_node)
+
+    # Require an unambiguous single candidate per module name in that directory.
+    dir_alias_map: dict[Path, dict[str, str]] = {
+        d: {
+            mod: next(iter(fns))
+            for mod, fns in mod_map.items()
+            if len(fns) == 1
+        }
+        for d, mod_map in dir_siblings.items()
+    }
+
+    if not dir_alias_map:
+        return
+
+    for e in all_edges:
+        if not (
+            isinstance(e, dict)
+            and e.get("relation") in ("imports", "imports_from")
+            and str(e.get("source_file", "")).lower().endswith((".py", ".pyi"))
+        ):
+            continue
+
+        sf = e.get("source_file")
+        if not sf:
+            continue
+        try:
+            sf_path = Path(sf)
+            if not sf_path.is_absolute():
+                sf_path = (root / sf_path).resolve()
+            else:
+                sf_path = sf_path.resolve()
+        except OSError:
+            continue
+
+        imp_dir = sf_path.parent
+        aliases = dir_alias_map.get(imp_dir)
+        if not aliases:
+            continue
+
+        tgt = e.get("target")
+        if tgt in aliases:
+            repointed = aliases[tgt]
+            if repointed != e.get("source"):
+                e["target"] = repointed
+
+
+
 SEMANTIC_RELATIONS = frozenset({
     "inherits", "implements", "mixes_in", "embeds", "references",
     "calls", "imports", "imports_from", "re_exports", "contains", "method",
@@ -6936,6 +7041,7 @@ def _learn(e: dict) -> None:
     # (src/) package root before the resolver/import-evidence passes run, so the
     # graph is identical regardless of scan root (#2072).
     _repoint_python_package_imports(paths, all_nodes, all_edges, root)
+    _repoint_python_sibling_imports(paths, all_nodes, all_edges, root)
     _merge_swift_extensions(per_file, all_nodes, all_edges)
     _merge_csharp_partial_class_nodes(per_file, all_nodes, all_edges, paths, root)
     _disambiguate_colliding_node_ids(all_nodes, all_edges, all_raw_calls, root)

tests/test_loose_sibling_import_resolution.py

@@ -0,0 +1,263 @@
+"""Regression tests for Graphify issue #3430:
+Loose/namespace-style sibling module imports below scan root.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+from graphify.build import build_from_json
+from graphify.extract import _file_node_id, _repoint_python_sibling_imports, extract
+
+
+def _edge_set(G):
+    """Return set of (relation, source, target) from G."""
+    return {
+        (d.get("relation"), d.get("_src", u), d.get("_tgt", v))
+        for u, v, d in G.edges(data=True)
+    }
+
+
+def test_primary_regression_loose_sibling_import_and_call(tmp_path):
+    """A. Primary regression:
+    repo/scripts/greeter.py + repo/scripts/main.py with:
+    import greeter
+    greeter.greet()
+    Extract from repo root and assert:
+    scripts_main imports scripts_greeter
+    scripts_main_run calls scripts_greeter_greet
+    """
+    scripts = tmp_path / "scripts"
+    scripts.mkdir(parents=True)
+    (scripts / "greeter.py").write_text("def greet():\n    return 42\n", encoding="utf-8")
+    (scripts / "main.py").write_text(
+        "import greeter\n\n"
+        "def run():\n"
+        "    return greeter.greet()\n",
+        encoding="utf-8",
+    )
+
+    paths = [scripts / "greeter.py", scripts / "main.py"]
+    res = extract(paths, root=tmp_path, parallel=False)
+    G = build_from_json(res, root=str(tmp_path), directed=True)
+
+    edges = _edge_set(G)
+    assert ("imports", "scripts_main", "scripts_greeter") in edges, (
+        f"scripts_main -> scripts_greeter import edge missing: {edges}"
+    )
+    assert ("calls", "scripts_main_run", "scripts_greeter_greet") in edges, (
+        f"scripts_main_run -> scripts_greeter_greet calls edge missing: {edges}"
+    )
+
+
+def test_scan_root_parity(tmp_path):
+    """B. Scan-root parity:
+    Extract the same two files with scripts/ as the root and verify the
+    equivalent unprefixed graph still resolves correctly and is isomorphic.
+    """
+    scripts = tmp_path / "scripts"
+    scripts.mkdir(parents=True)
+    (scripts / "greeter.py").write_text("def greet():\n    return 42\n", encoding="utf-8")
+    (scripts / "main.py").write_text(
+        "import greeter\n\n"
+        "def run():\n"
+        "    return greeter.greet()\n",
+        encoding="utf-8",
+    )
+
+    paths = [scripts / "greeter.py", scripts / "main.py"]
+
+    # Scan from repo root
+    root_res = extract(paths, root=tmp_path, parallel=False)
+    root_G = build_from_json(root_res, root=str(tmp_path), directed=True)
+
+    # Scan from scripts/ root
+    scripts_res = extract(paths, root=scripts, parallel=False)
+    scripts_G = build_from_json(scripts_res, root=str(scripts), directed=True)
+
+    scripts_edges = _edge_set(scripts_G)
+    assert ("imports", "main", "greeter") in scripts_edges
+    assert ("calls", "main_run", "greeter_greet") in scripts_edges
+
+    # Map root_edges to stripped prefix forms
+    stripped_root_edges = {
+        (
+            rel,
+            src[8:] if src.startswith("scripts_") else src,
+            tgt[8:] if tgt.startswith("scripts_") else tgt,
+        )
+        for rel, src, tgt in _edge_set(root_G)
+    }
+    assert stripped_root_edges == scripts_edges, (
+        f"Parity mismatch between scan roots:\n"
+        f"Root-scanned (stripped): {stripped_root_edges}\n"
+        f"Scripts-scanned: {scripts_edges}"
+    )
+
+
+def test_same_name_modules_in_separate_loose_directories(tmp_path):
+    """C. Same-name modules in separate loose directories:
+    scripts/helper.py + scripts/main.py
+    tools/helper.py + tools/main.py
+    Each importer must resolve only to its own directory's helper.
+    """
+    scripts = tmp_path / "scripts"
+    tools = tmp_path / "tools"
+    scripts.mkdir(parents=True)
+    tools.mkdir(parents=True)
+
+    (scripts / "helper.py").write_text("def work():\n    return 'scripts'\n", encoding="utf-8")
+    (scripts / "main.py").write_text(
+        "import helper\n\n"
+        "def run():\n"
+        "    return helper.work()\n",
+        encoding="utf-8",
+    )
+
+    (tools / "helper.py").write_text("def work():\n    return 'tools'\n", encoding="utf-8")
+    (tools / "main.py").write_text(
+        "import helper\n\n"
+        "def run():\n"
+        "    return helper.work()\n",
+        encoding="utf-8",
+    )
+
+    paths = [
+        scripts / "helper.py",
+        scripts / "main.py",
+        tools / "helper.py",
+        tools / "main.py",
+    ]
+    res = extract(paths, root=tmp_path, parallel=False)
+    G = build_from_json(res, root=str(tmp_path), directed=True)
+
+    edges = _edge_set(G)
+    # scripts resolution
+    assert ("imports", "scripts_main", "scripts_helper") in edges
+    assert ("calls", "scripts_main_run", "scripts_helper_work") in edges
+    assert ("imports", "scripts_main", "tools_helper") not in edges
+    assert ("calls", "scripts_main_run", "tools_helper_work") not in edges
+
+    # tools resolution
+    assert ("imports", "tools_main", "tools_helper") in edges
+    assert ("calls", "tools_main_run", "tools_helper_work") in edges
+    assert ("imports", "tools_main", "scripts_helper") not in edges
+    assert ("calls", "tools_main_run", "scripts_helper_work") not in edges
+
+
+def test_package_isolation_pep328(tmp_path):
+    """D. Package isolation:
+    pkg/__init__.py + pkg/helper.py + pkg/app.py with `import helper`
+    Verify the new loose-sibling pass does NOT repoint this package import.
+    """
+    pkg = tmp_path / "pkg"
+    pkg.mkdir(parents=True)
+    (pkg / "__init__.py").write_text("", encoding="utf-8")
+    (pkg / "helper.py").write_text("def work():\n    return 1\n", encoding="utf-8")
+    (pkg / "app.py").write_text(
+        "import helper\n\n"
+        "def run():\n"
+        "    return helper.work()\n",
+        encoding="utf-8",
+    )
+
+    paths = [pkg / "__init__.py", pkg / "helper.py", pkg / "app.py"]
+    res = extract(paths, root=tmp_path, parallel=False)
+
+    # In extract result, pkg_app's import must NOT be repointed to pkg_helper
+    import_edges = [
+        e for e in res["edges"]
+        if e.get("source") == "pkg_app" and e.get("relation") == "imports"
+    ]
+    assert len(import_edges) == 1
+    assert import_edges[0]["target"] == "helper", (
+        f"Package import in pkg/app.py was falsely repointed to sibling pkg_helper: {import_edges[0]}"
+    )
+
+    # No member call was resolved (package isolation)
+    call_edges = [e for e in res["edges"] if e.get("relation") == "calls"]
+    assert not call_edges, f"Spurious call edge resolved inside package: {call_edges}"
+
+
+def test_no_cross_directory_leakage(tmp_path):
+    """E. No cross-directory leakage:
+    scripts/utils.py + src/app.py with `import utils`
+    Verify src/app.py does not acquire an import/call relationship to scripts/utils.py.
+    """
+    scripts = tmp_path / "scripts"
+    src = tmp_path / "src"
+    scripts.mkdir(parents=True)
+    src.mkdir(parents=True)
+
+    (scripts / "utils.py").write_text("def helper():\n    return 'util'\n", encoding="utf-8")
+    (src / "app.py").write_text(
+        "import utils\n\n"
+        "def run():\n"
+        "    return utils.helper()\n",
+        encoding="utf-8",
+    )
+
+    paths = [scripts / "utils.py", src / "app.py"]
+    res = extract(paths, root=tmp_path, parallel=False)
+
+    # In extract result, src_app's import must NOT be repointed to scripts_utils
+    import_edges = [
+        e for e in res["edges"]
+        if e.get("source") == "src_app" and e.get("relation") == "imports"
+    ]
+    assert len(import_edges) == 1
+    assert import_edges[0]["target"] == "utils", (
+        f"Unrelated src/app.py import was leaked to scripts_utils: {import_edges[0]}"
+    )
+
+    # No member call resolved
+    call_edges = [
+        e for e in res["edges"]
+        if e.get("source") == "src_app_run" and e.get("relation") == "calls"
+    ]
+    assert not call_edges, f"Spurious cross-directory call edge resolved: {call_edges}"
+
+
+def test_aliased_import_resolves_member_call(tmp_path):
+    """F. Aliased import:
+    scripts/greeter.py + scripts/main.py with:
+    import greeter as g
+    g.greet()
+    Verify the calls edge resolves to scripts_greeter_greet and the
+    import alias metadata remains intact during repointing.
+    """
+    scripts = tmp_path / "scripts"
+    scripts.mkdir(parents=True)
+    (scripts / "greeter.py").write_text("def greet():\n    return 42\n", encoding="utf-8")
+    (scripts / "main.py").write_text(
+        "import greeter as g\n\n"
+        "def run():\n"
+        "    return g.greet()\n",
+        encoding="utf-8",
+    )
+
+    paths = [scripts / "greeter.py", scripts / "main.py"]
+
+    # Verify directly on _repoint_python_sibling_imports that local_alias is preserved
+    all_nodes = [
+        {"id": "scripts_greeter", "source_file": str(scripts / "greeter.py")},
+        {"id": "scripts_main", "source_file": str(scripts / "main.py")},
+    ]
+    edge = {
+        "source": "scripts_main",
+        "target": "greeter",
+        "relation": "imports",
+        "source_file": str(scripts / "main.py"),
+        "local_alias": "g",
+    }
+    all_edges = [edge]
+    _repoint_python_sibling_imports(paths, all_nodes, all_edges, root=tmp_path)
+
+    assert edge["target"] == "scripts_greeter"
+    assert edge["local_alias"] == "g", "local_alias metadata must remain intact after repointing"
+
+    # Full extraction pipeline verify
+    res = extract(paths, root=tmp_path, parallel=False)
+    G = build_from_json(res, root=str(tmp_path), directed=True)
+    edges = _edge_set(G)
+    assert ("imports", "scripts_main", "scripts_greeter") in edges
+    assert ("calls", "scripts_main_run", "scripts_greeter_greet") in edges