mirror of
https://github.com/safishamsi/graphify.git
synced 2026-10-13 15:40:05 +00:00
rohit-jsfreaky
0083ee78e5
A relative import of a module with no file behind it (`from .absent
import x`) built its target from the attempted absolute path, so the id
became e.g. `c_users_<user>_..._pkg_absent_py`. The root-relative id
remap never rewrites it (there is no node to anchor on), so the checkout
location and the OS username ended up in graph.json and every clone got
a different id.
Use the dotted module name relative to the scan root instead
(`pkg_absent`), the id an unresolved absolute import of the same module
already gets, so `from .absent` and `from pkg.absent` now meet on one
node. An import that climbs above the scan root falls back to a `ref`
id. Imports that resolve, and attempted paths that exist, are unchanged.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B8gwEpjajiWHVW3c9smuKf
(cherry picked from commit 8cdbb0ea4f)
566 lines
24 KiB
Python
566 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from graphify.extract import extract
|
|
from graphify.extractors.resolution import (
|
|
_SCAN_ROOT_NAMESPACE_CACHE,
|
|
_infer_scan_root_namespace,
|
|
_resolve_python_module_path,
|
|
)
|
|
|
|
|
|
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 _node_id(result: dict, label: str, source_file: str) -> str:
|
|
matches = [
|
|
node["id"]
|
|
for node in result["nodes"]
|
|
if node.get("label") == label and node.get("source_file") == source_file
|
|
]
|
|
assert len(matches) == 1
|
|
return matches[0]
|
|
|
|
|
|
def _has_edge(result: dict, source: str, target: str, relation: str) -> bool:
|
|
return any(
|
|
edge["source"] == source
|
|
and edge["target"] == target
|
|
and edge["relation"] == relation
|
|
for edge in result["edges"]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(("import_line", "receiver", "target"), [
|
|
("import requests", "requests", "requests"),
|
|
("import requests as rq", "rq", "requests"),
|
|
("import requests as Requests", "Requests", "requests"),
|
|
("import requests\nfrom requests import Session", "requests", "requests"),
|
|
("from requests import Session; import requests", "requests", "requests"),
|
|
("from other import value; import requests as rq", "rq", "requests"),
|
|
("import requests as Requests\nfrom requests import Session", "Requests", "requests"),
|
|
("import os", "os", "os"),
|
|
("import pkg.sub as sub", "sub", "pkg_sub"),
|
|
])
|
|
def test_external_plain_import_member_call_targets_module(
|
|
tmp_path: Path, import_line: str, receiver: str, target: str,
|
|
):
|
|
source = _write(tmp_path / "caller.py", (
|
|
f"{import_line}\n\n"
|
|
"def fetch():\n"
|
|
f" return {receiver}.get('/data')\n\n"
|
|
"def unrelated():\n"
|
|
" return 1\n"
|
|
))
|
|
result = extract([source], cache_root=tmp_path)
|
|
caller = _node_id(result, "fetch()", "caller.py")
|
|
calls = [e for e in result["edges"] if e["relation"] == "calls" and e["target"] == target]
|
|
assert len(calls) == 1
|
|
assert calls[0]["source"] == caller
|
|
assert calls[0]["source_file"] == "caller.py"
|
|
assert (calls[0]["confidence"], calls[0]["confidence_score"], calls[0]["context"],
|
|
calls[0]["source_location"], calls[0]["weight"]) == ("EXTRACTED", 1.0, "call", f"L{len(import_line.splitlines()) + 3}", 1.0)
|
|
assert not any(n.get("label") == "get()" for n in result["nodes"])
|
|
assert not any("_python_receiver_shadowed" in item for item in result["nodes"] + result["edges"])
|
|
|
|
|
|
@pytest.mark.parametrize("body", [
|
|
"def fetch(requests):\n return requests.get()\n",
|
|
"def fetch(*, requests=None):\n return requests.get()\n",
|
|
"def fetch():\n requests = object()\n return requests.get()\n",
|
|
"def fetch():\n requests, x = pair\n return requests.get()\n",
|
|
"def fetch():\n requests += 1\n return requests.get()\n",
|
|
"def fetch():\n (requests := object())\n return requests.get()\n",
|
|
"def fetch():\n for requests in xs: pass\n return requests.get()\n",
|
|
"def fetch():\n xs = [requests.get() for requests in things]\n return requests.get()\n",
|
|
"def fetch():\n with open('x') as requests: pass\n return requests.get()\n",
|
|
"def fetch():\n try: pass\n except Exception as requests: pass\n return requests.get()\n",
|
|
"def fetch():\n def requests(): pass\n return requests.get()\n",
|
|
"def fetch():\n class requests: pass\n return requests.get()\n",
|
|
"def fetch():\n del requests\n return requests.get()\n",
|
|
"def fetch():\n requests.value = 1\n return requests.get()\n",
|
|
"def fetch():\n import requests\n return requests.get()\n",
|
|
"def fetch():\n global requests\n return requests.get()\n",
|
|
"def fetch():\n nonlocal requests\n return requests.get()\n",
|
|
"def fetch():\n match x:\n case requests: pass\n return requests.get()\n",
|
|
"def fetch():\n return (lambda requests: requests.get())(object())\n",
|
|
"def fetch():\n return (lambda *, requests=None: requests.get())()\n",
|
|
"def fetch():\n return (lambda **requests: requests.get())()\n",
|
|
"requests = object()\ndef fetch():\n return requests.get()\n",
|
|
"type requests = object\ndef fetch():\n return requests.get()\n",
|
|
"def fetch():\n type requests[T] = list[T]\n return requests.get()\n",
|
|
"def fetch[requests]():\n return requests.get()\n",
|
|
"import builtins\nbuiltins.exec('requests = object()')\ndef fetch():\n return requests.get()\n",
|
|
"import builtins as bi\nbi.eval('globals().update(requests=object())')\ndef fetch():\n return requests.get()\n",
|
|
"from builtins import exec as run\nrun('requests = object()')\ndef fetch():\n return requests.get()\n",
|
|
"import builtins\nrun = builtins.exec\nrun('requests = object()')\ndef fetch():\n return requests.get()\n",
|
|
"run = exec\nrun('requests = object()')\ndef fetch():\n return requests.get()\n",
|
|
"def outer():\n requests = object()\n def fetch():\n return requests.get()\n",
|
|
"if flag:\n import requests\ndef fetch():\n return requests.get()\n",
|
|
"from elsewhere import *\ndef fetch():\n return requests.get()\n",
|
|
"def fetch():\n exec('requests = other')\n return requests.get()\n",
|
|
"def fetch():\n eval('globals().__setitem__(\\\"requests\\\", other)')\n return requests.get()\n",
|
|
"def fetch():\n globals()['requests'] = other\n return requests.get()\n",
|
|
"def fetch():\n locals()['requests'] = other\n return requests.get()\n",
|
|
"def fetch():\n setattr(requests, 'get', other)\n return requests.get()\n",
|
|
"def fetch():\n delattr(requests, 'get')\n return requests.get()\n",
|
|
"def mutate():\n global requests\n requests = other\ndef fetch():\n return requests.get()\n",
|
|
"import requests\ndef fetch():\n return requests.get()\n",
|
|
"import other as requests\ndef fetch():\n return requests.get()\n",
|
|
])
|
|
def test_external_module_fallback_rejects_receiver_shadow(tmp_path: Path, body: str):
|
|
source = _write(tmp_path / "caller.py", "import requests\n" + body)
|
|
result = extract([source], cache_root=tmp_path)
|
|
assert not any(e["relation"] == "calls" and e["target"] == "requests" for e in result["edges"])
|
|
|
|
|
|
@pytest.mark.parametrize("declaration", [
|
|
"from requests import Session",
|
|
"from other import value as requests",
|
|
])
|
|
def test_external_module_fallback_rejects_from_only_import(tmp_path: Path, declaration: str):
|
|
source = _write(tmp_path / "caller.py", f"{declaration}\ndef fetch():\n return requests.get()\n")
|
|
result = extract([source], cache_root=tmp_path)
|
|
assert not any(e["relation"] == "calls" and e["target"] in {"requests", "other"}
|
|
for e in result["edges"])
|
|
|
|
|
|
def test_overdeep_relative_import_is_unresolved_not_fatal(tmp_path: Path):
|
|
source = _write(
|
|
tmp_path / "pkg" / "mod.py",
|
|
"from ........................... import missing\n\n"
|
|
"def ok():\n"
|
|
" return 1\n",
|
|
)
|
|
|
|
assert _resolve_python_module_path("", source, tmp_path, level=27) is None
|
|
|
|
result = extract([source], cache_root=tmp_path)
|
|
|
|
assert _node_id(result, "mod.py", "pkg/mod.py")
|
|
assert _node_id(result, "ok()", "pkg/mod.py")
|
|
|
|
|
|
def _missing_import_targets(checkout: Path) -> set[str]:
|
|
_write(checkout / "pkg" / "__init__.py", "")
|
|
source = _write(
|
|
checkout / "pkg" / "app.py",
|
|
"from .absent import helper\n"
|
|
"from pkg.absent import other\n\n"
|
|
"def run():\n"
|
|
" return helper(), other()\n",
|
|
)
|
|
result = extract([source, checkout / "pkg" / "__init__.py"], cache_root=checkout)
|
|
return {e["target"] for e in result["edges"] if e["relation"] == "imports_from"}
|
|
|
|
|
|
def test_missing_relative_import_target_id_does_not_depend_on_the_checkout(tmp_path: Path):
|
|
"""A relative import of a module with no file behind it minted its target id
|
|
from the attempted absolute path (``..._pkg_absent_py``), which the
|
|
root-relative remap never rewrites: the checkout location and OS username
|
|
ended up in the graph. It now gets the dotted module name, the id an
|
|
unresolved absolute import of the same module already gets."""
|
|
first = _missing_import_targets(tmp_path / "clone_one")
|
|
second = _missing_import_targets(tmp_path / "elsewhere" / "clone_two")
|
|
|
|
assert first == second == {"pkg_absent"}
|
|
|
|
|
|
def test_ordinary_relative_import_still_resolves(tmp_path: Path):
|
|
target = _write(tmp_path / "pkg" / "sibling.py", "def helper():\n return 1\n")
|
|
source = _write(tmp_path / "pkg" / "mod.py", "from .sibling import helper\n")
|
|
|
|
assert _resolve_python_module_path("sibling", source, tmp_path, level=1) == target
|
|
|
|
result = extract([source, target], cache_root=tmp_path)
|
|
source_file = _node_id(result, "mod.py", "pkg/mod.py")
|
|
target_symbol = _node_id(result, "helper()", "pkg/sibling.py")
|
|
|
|
assert _has_edge(result, source_file, target_symbol, "imports")
|
|
|
|
|
|
def test_relative_subpackage_import_from_targets_package_init(tmp_path: Path):
|
|
# `from ...graphs import build_graph` where `graphs/` is a package (a dir
|
|
# with __init__.py, not a graphs.py module). The imports_from edge must
|
|
# target the package's __init__.py file node, matching the companion
|
|
# `imports` edge — not an absolute-scan-path slug for a nonexistent
|
|
# graphs.py that dangles per-checkout (#2455).
|
|
init = _write(tmp_path / "src/mypkg/__init__.py", "")
|
|
api_init = _write(tmp_path / "src/mypkg/api/__init__.py", "")
|
|
routes_init = _write(tmp_path / "src/mypkg/api/routes/__init__.py", "")
|
|
graphs_init = _write(
|
|
tmp_path / "src/mypkg/graphs/__init__.py",
|
|
"def build_graph():\n return {}\n",
|
|
)
|
|
health = _write(
|
|
tmp_path / "src/mypkg/api/routes/health.py",
|
|
"from ...graphs import build_graph\n",
|
|
)
|
|
|
|
result = extract(
|
|
[init, api_init, routes_init, graphs_init, health], cache_root=tmp_path
|
|
)
|
|
|
|
health_file = _node_id(result, "health.py", "src/mypkg/api/routes/health.py")
|
|
graphs_pkg = _node_id(result, "__init__.py", "src/mypkg/graphs/__init__.py")
|
|
|
|
assert _has_edge(result, health_file, graphs_pkg, "imports_from")
|
|
# No imports_from edge out of health may carry an unresolved absolute-path
|
|
# slug (the pre-fix `<scan>_src_mypkg_graphs_py` target).
|
|
health_targets = [
|
|
e["target"]
|
|
for e in result["edges"]
|
|
if e["source"] == health_file and e["relation"] == "imports_from"
|
|
]
|
|
assert all(t.endswith("graphs_init") for t in health_targets), health_targets
|
|
|
|
|
|
def test_absolute_package_import_targets_package_init(tmp_path: Path):
|
|
"""Absolute package imports must not leave dotted-name dangling edges (#3723)."""
|
|
files = [
|
|
_write(tmp_path / "pkg/__init__.py", ""),
|
|
_write(tmp_path / "pkg/sub/__init__.py", ""),
|
|
_write(tmp_path / "pkg/sub/thing.py", "def run():\n return 1\n"),
|
|
_write(tmp_path / "pkg/consumer.py", "from pkg import sub\n"),
|
|
_write(tmp_path / "user.py", "from pkg.sub import thing\n"),
|
|
]
|
|
|
|
result = extract(files, cache_root=tmp_path)
|
|
|
|
consumer = _node_id(result, "consumer.py", "pkg/consumer.py")
|
|
user = _node_id(result, "user.py", "user.py")
|
|
import_targets = {
|
|
edge["target"]
|
|
for edge in result["edges"]
|
|
if edge["relation"] == "imports_from"
|
|
and edge["source"] in {consumer, user}
|
|
}
|
|
|
|
assert {"pkg_init", "pkg_sub_init"} <= import_targets
|
|
assert "pkg" not in import_targets
|
|
assert "pkg_sub" not in import_targets
|
|
|
|
|
|
def test_plain_absolute_import_targets_package_module(tmp_path: Path):
|
|
package_init = _write(tmp_path / "pkg/__init__.py", "")
|
|
subpackage_init = _write(tmp_path / "pkg/sub/__init__.py", "")
|
|
consumer_path = _write(tmp_path / "app.py", "import pkg.sub\n")
|
|
|
|
result = extract(
|
|
[package_init, subpackage_init, consumer_path],
|
|
cache_root=tmp_path,
|
|
root=tmp_path,
|
|
parallel=False,
|
|
)
|
|
|
|
consumer = _node_id(result, "app.py", "app.py")
|
|
subpackage = _node_id(result, "__init__.py", "pkg/sub/__init__.py")
|
|
assert _has_edge(result, consumer, subpackage, "imports")
|
|
|
|
|
|
def test_nested_plain_import_target_is_stamped_for_incremental_remap(tmp_path: Path):
|
|
"""A changed importer can target an unchanged module outside its batch."""
|
|
_write(tmp_path / "src/pkg/__init__.py", "")
|
|
target = _write(tmp_path / "src/pkg/sub/__init__.py", "")
|
|
app_path = _write(tmp_path / "src/pkg/app.py", "import pkg.sub\n")
|
|
|
|
result = extract(
|
|
[app_path], cache_root=tmp_path / "cache", root=tmp_path, parallel=False
|
|
)
|
|
|
|
app = next(
|
|
node["id"] for node in result["nodes"] if node.get("label") == "app.py"
|
|
)
|
|
target_id = "src_pkg_sub_init"
|
|
assert _has_edge(result, app, target_id, "imports")
|
|
assert target.is_file()
|
|
|
|
|
|
def test_absolute_import_does_not_resolve_above_scan_root(tmp_path: Path):
|
|
scan_root = tmp_path / "scan"
|
|
source = _write(
|
|
scan_root / "app.py",
|
|
"from outside_pkg import thing\nimport outside_pkg\n",
|
|
)
|
|
_write(tmp_path / "outside_pkg/__init__.py", "")
|
|
_write(tmp_path / "outside_pkg/thing.py", "def run():\n return 1\n")
|
|
|
|
result = extract(
|
|
[source], cache_root=tmp_path / "cache", root=scan_root, parallel=False
|
|
)
|
|
|
|
app = _node_id(result, "app.py", "app.py")
|
|
targets = {
|
|
(edge["relation"], edge["target"])
|
|
for edge in result["edges"]
|
|
if edge["source"] == app
|
|
and edge["relation"] in ("imports", "imports_from")
|
|
}
|
|
assert targets == {
|
|
("imports", "outside_pkg"),
|
|
("imports_from", "outside_pkg"),
|
|
}
|
|
|
|
|
|
def test_absolute_from_import_keeps_namespace_package_submodule_edge(tmp_path: Path):
|
|
namespace_package = tmp_path / "namespace_pkg"
|
|
namespace_package.mkdir()
|
|
submodule = _write(
|
|
namespace_package / "subspace/worker.py", "def run():\n return 1\n"
|
|
)
|
|
consumer_path = _write(
|
|
tmp_path / "app.py", "from namespace_pkg.subspace import worker\n"
|
|
)
|
|
|
|
result = extract(
|
|
[consumer_path, submodule], cache_root=tmp_path, root=tmp_path, parallel=False
|
|
)
|
|
|
|
consumer = _node_id(result, "app.py", "app.py")
|
|
worker = _node_id(result, "worker.py", "namespace_pkg/subspace/worker.py")
|
|
assert _has_edge(result, consumer, worker, "imports_from")
|
|
|
|
|
|
def test_python_package_reexport_resolves_import_and_call_to_origin_symbol(tmp_path: Path):
|
|
origin = _write(tmp_path / "pkg/foo.py", "def Foo():\n return 1\n")
|
|
barrel = _write(tmp_path / "pkg/__init__.py", "from .foo import Foo as PublicFoo\n")
|
|
consumer = _write(
|
|
tmp_path / "app.py",
|
|
"from pkg import PublicFoo\n\n"
|
|
"def X():\n"
|
|
" return PublicFoo()\n",
|
|
)
|
|
|
|
result = extract([origin, barrel, consumer], cache_root=tmp_path)
|
|
|
|
origin_file = _node_id(result, "foo.py", "pkg/foo.py")
|
|
barrel_file = _node_id(result, "__init__.py", "pkg/__init__.py")
|
|
consumer_file = _node_id(result, "app.py", "app.py")
|
|
origin_symbol = _node_id(result, "Foo()", "pkg/foo.py")
|
|
consumer_symbol = _node_id(result, "X()", "app.py")
|
|
|
|
assert _has_edge(result, barrel_file, origin_file, "re_exports")
|
|
assert _has_edge(result, consumer_file, origin_symbol, "imports")
|
|
assert _has_edge(result, consumer_symbol, origin_symbol, "calls")
|
|
|
|
|
|
def test_python_parameter_return_and_generic_contexts(tmp_path: Path):
|
|
model = tmp_path / "pkg" / "model.py"
|
|
model.parent.mkdir(parents=True)
|
|
model.write_text(
|
|
"class Payload:\n"
|
|
" pass\n\n"
|
|
"class Result:\n"
|
|
" pass\n",
|
|
encoding="utf-8",
|
|
)
|
|
service = tmp_path / "pkg" / "service.py"
|
|
service.write_text(
|
|
"from .model import Payload, Result\n\n"
|
|
"def process(item: Payload) -> Result:\n"
|
|
" return Result()\n\n"
|
|
"def process_many(items: list[Payload]) -> Result:\n"
|
|
" return Result()\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = extract([model, service], cache_root=tmp_path)
|
|
labels = {node["id"]: node["label"] for node in result["nodes"]}
|
|
edges = [edge for edge in result["edges"] if edge.get("relation") == "references"]
|
|
pairs = {
|
|
(labels.get(e["source"], e["source"]), labels.get(e["target"], e["target"]), e.get("context"))
|
|
for e in edges
|
|
}
|
|
|
|
assert ("process()", "Payload", "parameter_type") in pairs
|
|
assert ("process()", "Result", "return_type") in pairs
|
|
assert ("process_many()", "Payload", "generic_arg") in pairs
|
|
|
|
|
|
def test_issue_3777_package_module_collision_phantom_cycle_absent(tmp_path: Path):
|
|
from graphify.analyze import find_import_cycles
|
|
from graphify.build import build_from_json
|
|
|
|
nettacker_py = _write(
|
|
tmp_path / "nettacker.py",
|
|
"from nettacker.main import run\n\ndef cli():\n run()\n",
|
|
)
|
|
init_py = _write(tmp_path / "nettacker/__init__.py", "")
|
|
main_py = _write(
|
|
tmp_path / "nettacker/main.py",
|
|
"from nettacker.core.app import Nettacker\n\ndef run():\n return Nettacker()\n",
|
|
)
|
|
app_py = _write(
|
|
tmp_path / "nettacker/core/app.py",
|
|
"from nettacker import logger\n\nclass Nettacker:\n def start(self):\n logger.log_info('start')\n",
|
|
)
|
|
logger_py = _write(
|
|
tmp_path / "nettacker/logger.py",
|
|
"def log_info(msg):\n print(msg)\n",
|
|
)
|
|
|
|
result = extract(
|
|
[nettacker_py, init_py, main_py, app_py, logger_py],
|
|
cache_root=tmp_path,
|
|
root=tmp_path,
|
|
)
|
|
|
|
app_file = _node_id(result, "app.py", "nettacker/core/app.py")
|
|
logger_file = _node_id(result, "logger.py", "nettacker/logger.py")
|
|
nettacker_file = _node_id(result, "nettacker.py", "nettacker.py")
|
|
|
|
assert _has_edge(result, app_file, logger_file, "imports_from")
|
|
assert not _has_edge(result, app_file, nettacker_file, "imports_from")
|
|
|
|
graph = build_from_json(result)
|
|
assert find_import_cycles(graph) == []
|
|
|
|
|
|
def test_issue_3777_nested_module_package_collision_resolves_to_submodule(tmp_path: Path):
|
|
from graphify.analyze import find_import_cycles
|
|
from graphify.build import build_from_json
|
|
|
|
runner_py = _write(
|
|
tmp_path / "pkg/runner.py",
|
|
"from pkg.runner.step import run\n\ndef start():\n run()\n",
|
|
)
|
|
init_py = _write(tmp_path / "pkg/runner/__init__.py", "")
|
|
step_py = _write(
|
|
tmp_path / "pkg/runner/step.py",
|
|
"from pkg.runner import helper\n\ndef run():\n helper.work()\n",
|
|
)
|
|
helper_py = _write(
|
|
tmp_path / "pkg/runner/helper.py",
|
|
"def work():\n pass\n",
|
|
)
|
|
|
|
result = extract(
|
|
[runner_py, init_py, step_py, helper_py],
|
|
cache_root=tmp_path,
|
|
root=tmp_path,
|
|
)
|
|
|
|
step_file = _node_id(result, "step.py", "pkg/runner/step.py")
|
|
helper_file = _node_id(result, "helper.py", "pkg/runner/helper.py")
|
|
runner_file = _node_id(result, "runner.py", "pkg/runner.py")
|
|
|
|
assert _has_edge(result, step_file, helper_file, "imports_from")
|
|
assert not _has_edge(result, step_file, runner_file, "imports_from")
|
|
|
|
graph = build_from_json(result)
|
|
assert find_import_cycles(graph) == []
|
|
|
|
|
|
def test_issue_3777_namespace_package_submodule_import(tmp_path: Path):
|
|
sub = _write(tmp_path / "ns/sub.py", "def helper():\n pass\n")
|
|
consumer = _write(tmp_path / "ns/consumer.py", "from ns import sub\n")
|
|
|
|
result = extract([sub, consumer], cache_root=tmp_path, root=tmp_path)
|
|
|
|
consumer_file = _node_id(result, "consumer.py", "ns/consumer.py")
|
|
sub_file = _node_id(result, "sub.py", "ns/sub.py")
|
|
|
|
assert _has_edge(result, consumer_file, sub_file, "imports_from")
|
|
|
|
|
|
def test_issue_3777_standalone_module_import_unaffected(tmp_path: Path):
|
|
standalone = _write(tmp_path / "standalone.py", "def fn():\n return 42\n")
|
|
consumer = _write(tmp_path / "consumer.py", "from standalone import fn\n")
|
|
|
|
result = extract([standalone, consumer], cache_root=tmp_path, root=tmp_path)
|
|
|
|
consumer_file = _node_id(result, "consumer.py", "consumer.py")
|
|
standalone_file = _node_id(result, "standalone.py", "standalone.py")
|
|
fn_symbol = _node_id(result, "fn()", "standalone.py")
|
|
|
|
assert _has_edge(result, consumer_file, standalone_file, "imports_from")
|
|
assert _has_edge(result, consumer_file, fn_symbol, "imports")
|
|
|
|
|
|
def test_nested_scan_root_resolves_full_namespace_import(tmp_path: Path) -> None:
|
|
_write(tmp_path / "Company" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "__init__.py", "")
|
|
lib_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "lib" / "delivery.py", "def deliver(): pass")
|
|
main_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "app" / "main.py", "from Company.Apps.Jobs.Team.lib import delivery")
|
|
|
|
root = tmp_path / "Company" / "Apps" / "Jobs" / "Team"
|
|
resolved = _resolve_python_module_path("Company.Apps.Jobs.Team.lib.delivery", main_path, root, 0)
|
|
assert resolved == lib_path
|
|
|
|
# E2E check
|
|
result = extract([main_path, lib_path], cache_root=tmp_path, root=root)
|
|
main_node = _node_id(result, "main.py", "app/main.py")
|
|
lib_node = _node_id(result, "delivery.py", "lib/delivery.py")
|
|
assert _has_edge(result, main_node, lib_node, "imports_from")
|
|
|
|
|
|
def test_nested_scan_root_does_not_resolve_third_party(tmp_path: Path) -> None:
|
|
_write(tmp_path / "Company" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "__init__.py", "")
|
|
main_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "app" / "main.py", "from thirdparty.foo import bar")
|
|
|
|
root = tmp_path / "Company" / "Apps" / "Jobs" / "Team"
|
|
resolved = _resolve_python_module_path("thirdparty.foo", main_path, root, 0)
|
|
assert resolved is None
|
|
|
|
|
|
def test_repo_root_scan_is_unaffected(tmp_path: Path) -> None:
|
|
_write(tmp_path / "Company" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "__init__.py", "")
|
|
lib_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "lib" / "delivery.py", "def deliver(): pass")
|
|
main_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "app" / "main.py", "from Company.Apps.Jobs.Team.lib import delivery")
|
|
|
|
root = tmp_path
|
|
resolved = _resolve_python_module_path("Company.Apps.Jobs.Team.lib.delivery", main_path, root, 0)
|
|
assert resolved == lib_path
|
|
|
|
|
|
def test_non_package_subdirectory_scan_is_unaffected(tmp_path: Path) -> None:
|
|
_write(tmp_path / "pkg" / "thing.py", "")
|
|
app_path = _write(tmp_path / "src" / "app.py", "from pkg import thing")
|
|
|
|
root = tmp_path / "src"
|
|
# No __init__.py above src/, so namespace inference should be empty
|
|
assert _infer_scan_root_namespace(root) == ""
|
|
|
|
|
|
def test_partial_namespace_prefix_is_not_stripped(tmp_path: Path) -> None:
|
|
_write(tmp_path / "Company" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "__init__.py", "")
|
|
main_path = _write(tmp_path / "Company" / "Apps" / "Jobs" / "Team" / "app" / "main.py", "from Company.AppService.foo import bar")
|
|
|
|
root = tmp_path / "Company" / "Apps" / "Jobs" / "Team"
|
|
resolved = _resolve_python_module_path("Company.AppService.foo", main_path, root, 0)
|
|
assert resolved is None
|
|
|
|
|
|
def test_infer_scan_root_namespace_caching(tmp_path: Path) -> None:
|
|
_SCAN_ROOT_NAMESPACE_CACHE.clear()
|
|
_write(tmp_path / "Company" / "__init__.py", "")
|
|
_write(tmp_path / "Company" / "Apps" / "__init__.py", "")
|
|
root = tmp_path / "Company" / "Apps"
|
|
|
|
ns1 = _infer_scan_root_namespace(root)
|
|
assert ns1 == "Company.Apps"
|
|
|
|
# Second call should be from cache. We can verify cache exists.
|
|
ns2 = _infer_scan_root_namespace(root)
|
|
assert ns1 == ns2
|
|
|
|
key = list(_SCAN_ROOT_NAMESPACE_CACHE.keys())[0]
|
|
assert _SCAN_ROOT_NAMESPACE_CACHE[key] == "Company.Apps"
|