Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 37 additions & 18 deletions diffgraph/structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,7 @@ class _Import:
line: int
snippet: str
bindings: Tuple[str, ...]
scope: Optional[str]


@dataclass(frozen=True)
Expand Down Expand Up @@ -337,7 +338,8 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
# ``import package.submodule`` binds ``package``.
binding = raw.split(".", 1)[0]
imports.append(_Import(
raw, node.start_point[0] + 1, snippet, (binding,)
raw, node.start_point[0] + 1, snippet, (binding,),
parents[-1][0] if parents else None,
))
else:
# Tree-sitter exposes absolute modules through ``module_name``
Expand Down Expand Up @@ -373,6 +375,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None:
node.start_point[0] + 1,
snippet,
tuple(imported),
parents[-1][0] if parents else None,
))
elif not parents and node.type in (
"assignment", "annotated_assignment", "type_alias_statement"
Expand Down Expand Up @@ -549,7 +552,7 @@ def _resolve_call_target(
call: _Call,
symbols: Dict[str, _Symbol],
bindings: Dict[Optional[str], set],
imported_targets: Dict[str, List[Tuple[int, Optional[str]]]],
imported_targets: Dict[Optional[str], Dict[str, List[Tuple[int, Optional[str]]]]],
) -> Optional[str]:
"""Resolve only syntax-grounded, same-file Python calls.

Expand All @@ -563,21 +566,29 @@ def _resolve_call_target(
current = symbols.get(current_name)
if current is None:
break
# Function and method scopes participate in lexical lookup. Class
# namespaces do not: a bare name in a method never resolves through
# sibling class attributes or methods.
if current.kind in ("function", "method"):
# Function and method scopes participate in lexical lookup. A class
# body also needs its own imports, but a method must never resolve
# through its enclosing class namespace.
is_initial_class_body = (
current_name == call.caller and current.kind == "class"
)
if current.kind in ("function", "method") or is_initial_class_body:
if call.name in bindings.get(current_name, set()):
history = imported_targets.get(current_name, {}).get(call.name, [])
visible = [target for line, target in history if line <= call.line]
if visible:
return visible[-1]
Comment thread
coderabbitai[bot] marked this conversation as resolved.
return None
candidates.append("{}.{}".format(current_name, call.name))
if current.kind in ("function", "method"):
candidates.append("{}.{}".format(current_name, call.name))
current_name = current.parent

if call.name in bindings.get(None, set()):
# An explicit import is a deterministic external target. Other global
# bindings (for example an assignment) remain intentionally unresolved.
# Select the binding visible at this call site rather than applying a
# later top-level rebind retroactively.
history = imported_targets.get(call.name, [])
history = imported_targets.get(None, {}).get(call.name, [])
visible = [target for line, target in history if line <= call.line]
if visible and visible[-1] is not None:
return visible[-1]
Expand All @@ -604,27 +615,35 @@ def _resolve_call_target(
def _imported_call_targets(
imports: Dict[Tuple[str, int], _Import],
module_rebindings: List[Tuple[str, int]],
) -> Dict[str, List[Tuple[int, Optional[str]]]]:
"""Map each import binding to its conservative, line-aware history."""
) -> Dict[Optional[str], Dict[str, List[Tuple[int, Optional[str]]]]]:
"""Map import bindings to conservative, lexical and line-aware histories.

Imports inside a function bind only that function's local scope. Keeping
that scope alongside the source line permits a direct call to an imported
name without leaking the binding into sibling functions or the module.
"""

targets: Dict[str, List[Tuple[int, Optional[str]]]] = {}
imported_bindings = set()
targets: Dict[Optional[str], Dict[str, List[Tuple[int, Optional[str]]]]] = {}
imported_bindings: Dict[Optional[str], set] = {}
for (module, occurrence), item in imports.items():
suffix = "" if occurrence == 0 else "#{}".format(occurrence)
target = "import::{}{}".format(module, suffix)
scope_targets = targets.setdefault(item.scope, {})
scope_bindings = imported_bindings.setdefault(item.scope, set())
for binding in item.bindings:
# A later import of the same local name is intentionally
# unresolved, but calls before it retain the earlier binding.
targets.setdefault(binding, []).append((
item.line, None if binding in imported_bindings else target
scope_targets.setdefault(binding, []).append((
item.line, None if binding in scope_bindings else target
))
imported_bindings.add(binding)
scope_bindings.add(binding)
for binding, line in module_rebindings:
# A declaration, assignment, or loop target replaces the imported
# binding only for calls at or after its source line.
targets.setdefault(binding, []).append((line, None))
for history in targets.values():
history.sort(key=lambda item: item[0])
targets.setdefault(None, {}).setdefault(binding, []).append((line, None))
for scope_targets in targets.values():
for history in scope_targets.values():
history.sort(key=lambda item: item[0])
return targets


Expand Down
57 changes: 57 additions & 0 deletions tests/test_structural.py
Original file line number Diff line number Diff line change
Expand Up @@ -1370,6 +1370,63 @@ def test_explicit_from_import_creates_import_grounded_call_edge(tmp_path):
assert "query=python-structure-v2" in call["evidence"][0]["detail"]


def test_function_local_imports_ground_only_their_lexical_calls(tmp_path):
"""A local import is exact evidence for its own function, not its siblings."""
root = repo(tmp_path)
write(
root,
"local_import.py",
"def imports_and_calls():\n"
" from remote.worker import execute as run_remote\n"
" run_remote()\n\n"
"def sibling():\n"
" run_remote()\n",
)
git(root, "add", "local_import.py")

artifact = analyze_local_diff(str(root), staged=True)

assert_valid(artifact)
calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
assert len(calls) == 1
assert calls[0]["source_id"] == "sym::local_import.py::imports_and_calls"
assert calls[0]["target_id"] == "sym::local_import.py::import::remote.worker"
assert calls[0]["resolution_method"] == "import_grounded"
assert calls[0]["evidence"][0]["snippet"] == "run_remote()"


def test_class_body_imports_do_not_shadow_nested_method_lookups(tmp_path):
"""Class-body imports resolve there but remain outside method lexical scope."""
root = repo(tmp_path)
write(
root,
"class_import.py",
"def enclosing():\n"
" from outer.worker import execute as run_remote\n\n"
" class Worker:\n"
" from class_body.worker import execute as run_remote\n"
" run_remote()\n\n"
" def method(self):\n"
" run_remote()\n",
)
git(root, "add", "class_import.py")

artifact = analyze_local_diff(str(root), staged=True)

assert_valid(artifact)
calls = [item for item in artifact["relationships"] if item["kind"] == "calls"]
assert len(calls) == 2
assert [item["source_id"] for item in calls] == [
"sym::class_import.py::enclosing.Worker",
"sym::class_import.py::enclosing.Worker.method",
]
assert [item["target_id"] for item in calls] == [
"sym::class_import.py::import::class_body.worker",
"sym::class_import.py::import::outer.worker",
]
assert all(item["resolution_method"] == "import_grounded" for item in calls)


def test_relative_from_imports_preserve_package_evidence_and_call_bindings(tmp_path):
"""Relative imports are explicit package-local external dependencies.

Expand Down
Loading