diff --git a/diffgraph/structural.py b/diffgraph/structural.py index dbff6b8..e48d061 100644 --- a/diffgraph/structural.py +++ b/diffgraph/structural.py @@ -260,7 +260,7 @@ def _parse_python( imports: List[_Import] = [] calls: List[_Call] = [] bindings: Dict[Optional[str], set] = {} - module_rebindings: List[Tuple[str, int]] = [] + rebindings: List[Tuple[Optional[str], str, int]] = [] symbol_occurrences: Dict[str, int] = {} def identifiers(node) -> set: @@ -282,6 +282,17 @@ def as_target_identifiers(node) -> set: found.update(as_target_identifiers(child)) return found + def assignment_target_identifiers(node) -> set: + """Return names assigned by an assignment target, not names it reads.""" + if node.type in ("attribute", "subscript"): + return set() + if node.type == "identifier": + return {_node_text(content, node)} + found = set() + for child in node.children: + found.update(assignment_target_identifiers(child)) + return found + def case_pattern_identifiers(node) -> set: """Return capture names assigned by a Python structural match pattern. @@ -399,6 +410,12 @@ def is_declaration_header(node) -> bool: ) return False + def record_rebindings( + scope: Optional[str], names: set, position: int + ) -> None: + """Record lexical rebindings after the expression that creates them.""" + rebindings.extend((scope, name, position) for name in names) + def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: next_parents = parents if node.type in ("class_definition", "function_definition"): @@ -447,17 +464,15 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: ) ) next_parents = (*parents, (qname, kind)) - if not parents: - # A top-level declaration overwrites an imported binding at - # runtime just like a top-level assignment does, but only - # after the declaration header has been evaluated. - body = node.child_by_field_name("body") - binding_position = ( - body.start_byte - if node.type == "function_definition" and body is not None - else node.end_byte - ) - module_rebindings.append((name, binding_position)) + # A declaration overwrites an imported binding in its enclosing + # scope, but only after its declaration header is evaluated. + body = node.child_by_field_name("body") + binding_position = ( + body.start_byte + if node.type == "function_definition" and body is not None + else node.end_byte + ) + record_rebindings(parent, {name}, binding_position) elif node.type in ("import_statement", "import_from_statement"): snippet = _node_text(content, node) if node.type == "import_statement": @@ -594,15 +609,13 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: if name_node is not None and name_node.type == "identifier": name = _node_text(content, name_node) bindings.setdefault(scope, set()).add(name) - if scope is None: - module_rebindings.append((name, node.start_byte)) + record_rebindings(scope, {name}, node.start_byte) elif node.type == "case_clause": # Capture patterns bind their names before the case body executes. # Keep them lexical and line-aware just like assignment bindings. bound_names = case_pattern_identifiers(node) bindings.setdefault(scope, set()).update(bound_names) - if scope is None: - module_rebindings.extend((name, node.start_byte) for name in bound_names) + record_rebindings(scope, bound_names, node.start_byte) elif node.type in ( "assignment", "annotated_assignment", "for_statement", "named_expression" ): @@ -614,7 +627,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: "name" if node.type == "named_expression" else "left" ) if left is not None: - bound_names = identifiers(left) + bound_names = assignment_target_identifiers(left) binding_scope = scope if node.type == "named_expression": # Lambdas have their own lexical scope, but are not @@ -625,15 +638,16 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: elif is_declaration_header(node): binding_scope = parents[-2][0] if len(parents) > 1 else None bindings.setdefault(binding_scope, set()).update(bound_names) - if binding_scope is None: - binding_position = node.end_byte - if node.type == "for_statement": - iterable = node.child_by_field_name("right") - if iterable is not None: - binding_position = iterable.end_byte - module_rebindings.extend( - (name, binding_position) for name in bound_names - ) + binding_position = node.end_byte + if node.type == "for_statement": + iterable = node.child_by_field_name("right") + if iterable is not None: + binding_position = iterable.end_byte + annotation_only = node.type in ( + "assignment", "annotated_assignment" + ) and not any(child.type == "=" for child in node.children) + if not annotation_only: + record_rebindings(binding_scope, bound_names, binding_position) elif ( node.type == "as_pattern" and node.parent is not None @@ -650,8 +664,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: if target is not None: bound_names = as_target_identifiers(target) bindings.setdefault(scope, set()).update(bound_names) - if scope is None: - module_rebindings.extend((name, node.start_byte) for name in bound_names) + record_rebindings(scope, bound_names, target.start_byte) for child in node.children: visit(child, next_parents) @@ -661,7 +674,7 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: sorted(imports, key=lambda item: (item.line, item.module, item.snippet)), sorted(calls, key=lambda item: (item.line, item.caller or "", item.name, item.snippet)), bindings, - module_rebindings, + rebindings, ) @@ -801,7 +814,7 @@ def _resolve_call_target( def _imported_call_targets( imports: Dict[Tuple[str, int], _Import], - module_rebindings: List[Tuple[str, int]], + rebindings: List[Tuple[Optional[str], str, int]], ) -> Dict[Optional[str], Dict[str, List[Tuple[int, Optional[str]]]]]: """Map import bindings to conservative, lexical and line-aware histories. @@ -824,10 +837,10 @@ def _imported_call_targets( item.position, None if binding in scope_bindings else target )) scope_bindings.add(binding) - for binding, position in module_rebindings: + for scope, binding, position in rebindings: # A declaration, assignment, or loop target replaces the imported # binding after its source expression has been evaluated. - targets.setdefault(None, {}).setdefault(binding, []).append((position, None)) + targets.setdefault(scope, {}).setdefault(binding, []).append((position, None)) for scope_targets in targets.values(): for history in scope_targets.values(): history.sort(key=lambda item: item[0]) diff --git a/tests/test_structural.py b/tests/test_structural.py index 2e47aec..1ac12fe 100644 --- a/tests/test_structural.py +++ b/tests/test_structural.py @@ -1492,6 +1492,93 @@ def test_as_pattern_bindings_do_not_create_import_grounded_call_edges(tmp_path): assert calls == [] +def test_as_binding_keeps_import_visible_in_context_expression(tmp_path): + """An ``as`` alias binds only after its context expression runs.""" + root = repo(tmp_path) + write( + root, + "as_binding_order.py", + "from remote.worker import execute as run_remote\n\n" + "with run_remote() as run_remote:\n" + " run_remote()\n", + ) + git(root, "add", "as_binding_order.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]["resolution_method"] == "import_grounded" + assert calls[0]["evidence"][0]["line_start"] == 3 + + +def test_local_import_rebinding_stops_import_grounded_call_edges(tmp_path): + """A local assignment must replace a same-scope imported call target.""" + root = repo(tmp_path) + write( + root, + "local_import_rebinding.py", + "def caller():\n" + " from remote.worker import execute as run_remote\n" + " run_remote()\n" + " run_remote = lambda: None\n" + " run_remote()\n", + ) + git(root, "add", "local_import_rebinding.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_rebinding.py::caller" + assert calls[0]["resolution_method"] == "import_grounded" + assert calls[0]["evidence"][0]["line_start"] == 3 + + +def test_attribute_and_annotation_only_assignments_do_not_shadow_imports(tmp_path): + """Only assignment targets with a value replace an imported binding.""" + root = repo(tmp_path) + write( + root, + "assignment_binding_precision.py", + "from remote.worker import execute as run_remote\n\n" + "run_remote.result = None\n" + "run_remote: object\n" + "run_remote()\n", + ) + git(root, "add", "assignment_binding_precision.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]["resolution_method"] == "import_grounded" + + +@pytest.mark.parametrize("declaration", ["def run_remote():\n pass", "class run_remote:\n pass"]) +def test_nested_declaration_rebinds_function_local_import(tmp_path, declaration): + """Nested declarations replace same-scope imported call targets.""" + root = repo(tmp_path) + write( + root, + "nested_declaration_rebind.py", + "def caller():\n" + " from remote.worker import execute as run_remote\n\n" + + " " + declaration.replace("\n", "\n ") + "\n\n" + " run_remote()\n", + ) + git(root, "add", "nested_declaration_rebind.py") + + artifact = analyze_local_diff(str(root), staged=True) + + assert_valid(artifact) + calls = [item for item in artifact["relationships"] if item["kind"] == "calls"] + assert calls == [] + + def test_named_expression_bindings_do_not_create_import_grounded_call_edges(tmp_path): """A walrus target shadows an import for calls after the assignment.""" root = repo(tmp_path)