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 `_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"