diff --git a/diffgraph/structural.py b/diffgraph/structural.py index 66c3f0b..09bc372 100644 --- a/diffgraph/structural.py +++ b/diffgraph/structural.py @@ -57,6 +57,7 @@ class _Import: module: str line: int snippet: str + bindings: Tuple[str, ...] @dataclass(frozen=True) @@ -148,7 +149,13 @@ def _name_child(node): def _parse_python( content: bytes, -) -> Tuple[List[_Symbol], List[_Import], List[_Call], Dict[Optional[str], set]]: +) -> Tuple[ + List[_Symbol], + List[_Import], + List[_Call], + Dict[Optional[str], set], + List[Tuple[str, int]], +]: # Do not silently replace undecodable source: the warning must identify the # exact side that could not be structurally analyzed. content.decode("utf-8") @@ -160,6 +167,7 @@ def _parse_python( imports: List[_Import] = [] calls: List[_Call] = [] bindings: Dict[Optional[str], set] = {} + module_rebindings: List[Tuple[str, int]] = [] symbol_occurrences: Dict[str, int] = {} def identifiers(node) -> set: @@ -206,6 +214,10 @@ 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. + module_rebindings.append((name, node.start_point[0] + 1)) elif node.type in ("import_statement", "import_from_statement"): snippet = _node_text(content, node) if node.type == "import_statement": @@ -218,11 +230,37 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: raw = _node_text(content, name_node) else: raw = _node_text(content, item) - imports.append(_Import(raw, node.start_point[0] + 1, snippet)) + if item.type == "aliased_import": + binding = _node_text(content, item.children[-1]) + else: + # ``import package.submodule`` binds ``package``. + binding = raw.split(".", 1)[0] + imports.append(_Import( + raw, node.start_point[0] + 1, snippet, (binding,) + )) else: module_node = node.child_by_field_name("module_name") if module_node is not None: - imports.append(_Import(_node_text(content, module_node), node.start_point[0] + 1, snippet)) + imported = [] + after_import = False + for child in node.children: + if child.type == "import": + after_import = True + continue + if not after_import or child.type not in ( + "dotted_name", "aliased_import" + ): + continue + if child.type == "aliased_import": + imported.append(_node_text(content, child.children[-1])) + else: + imported.append(_node_text(content, child).split(".", 1)[0]) + imports.append(_Import( + _node_text(content, module_node), + node.start_point[0] + 1, + snippet, + tuple(imported), + )) elif node.type == "call": function = node.child_by_field_name("function") if function is not None and function.type == "identifier": @@ -255,7 +293,12 @@ def visit(node, parents: Tuple[Tuple[str, str], ...] = ()) -> None: elif node.type in ("assignment", "annotated_assignment", "for_statement"): left = node.child_by_field_name("left") if left is not None: - bindings.setdefault(scope, set()).update(identifiers(left)) + bound_names = identifiers(left) + bindings.setdefault(scope, set()).update(bound_names) + if scope is None: + module_rebindings.extend( + (name, node.start_point[0] + 1) for name in bound_names + ) for child in node.children: visit(child, next_parents) @@ -265,6 +308,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, ) @@ -351,6 +395,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]]]], ) -> Optional[str]: """Resolve only syntax-grounded, same-file Python calls. @@ -374,7 +419,18 @@ def _resolve_call_target( current_name = current.parent if call.name in bindings.get(None, set()): - return None + # 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, []) + visible = [target for line, target in history if line <= call.line] + if visible and visible[-1] is not None: + return visible[-1] + # A later import must not hide a declaration that was already visible + # at this call site. Likewise, a declaration that replaces an import + # can still resolve through the ordinary local-symbol path. Assignments + # and loop targets have no matching local symbol and remain unresolved. candidates.append(call.name) for candidate in candidates: @@ -391,6 +447,33 @@ def _resolve_call_target( return None +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.""" + + targets: Dict[str, List[Tuple[int, Optional[str]]]] = {} + imported_bindings = set() + for (module, occurrence), item in imports.items(): + suffix = "" if occurrence == 0 else "#{}".format(occurrence) + target = "import::{}{}".format(module, suffix) + 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 + )) + imported_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]) + return targets + + def analyze_local_diff( repository: str = ".", *, staged: bool = False, pathspecs: Optional[Sequence[str]] = None, @@ -487,16 +570,16 @@ def analyze_local_diff( UnicodeDecodeError, ValueError, RuntimeError, OSError, TypeError ) try: - old_symbols, old_imports, _old_calls, _old_bindings = ( - _parse_python(old) if old is not None else ([], [], [], {}) + old_symbols, old_imports, _old_calls, _old_bindings, _old_rebindings = ( + _parse_python(old) if old is not None else ([], [], [], {}, []) ) except parser_errors as error: warnings.append(_warning("PARSE_FAILURE", entry.old_path or path, "pre-change: {}: {}".format(type(error).__name__, error))) skipped += 1 continue try: - new_symbols, new_imports, new_calls, new_bindings = ( - _parse_python(new) if new is not None else ([], [], [], {}) + new_symbols, new_imports, new_calls, new_bindings, new_rebindings = ( + _parse_python(new) if new is not None else ([], [], [], {}, []) ) except parser_errors as error: warnings.append(_warning("PARSE_FAILURE", path, "post-change: {}: {}".format(type(error).__name__, error))) @@ -545,6 +628,9 @@ def analyze_local_diff( old_import_map = _keyed_imports(old_imports) new_import_map = _keyed_imports(new_imports) + imported_call_targets = _imported_call_targets( + new_import_map, new_rebindings + ) for import_key in sorted(set(old_import_map) | set(new_import_map)): before = old_import_map.get(import_key) after = new_import_map.get(import_key) @@ -613,7 +699,9 @@ def analyze_local_diff( call_occurrences: Dict[Tuple[str, str], int] = {} for call in new_calls: - target_qname = _resolve_call_target(call, new_map, new_bindings) + target_qname = _resolve_call_target( + call, new_map, new_bindings, imported_call_targets + ) if target_qname is None: continue source = ( @@ -633,7 +721,11 @@ def analyze_local_diff( "source_id": source, "target_id": target, "analysis_source": "structural", - "resolution_method": "resolved", + "resolution_method": ( + "import_grounded" + if target_qname.startswith("import::") + else "resolved" + ), "confidence": None, "evidence": [ { diff --git a/tests/test_structural.py b/tests/test_structural.py index 71466bd..d91710b 100644 --- a/tests/test_structural.py +++ b/tests/test_structural.py @@ -1025,3 +1025,104 @@ def test_python_calls_are_conservative_schema_valid_and_golden(tmp_path): assert shadowed_callers.isdisjoint(item["source_id"] for item in calls) assert all("query=python-structure-v2" in item["evidence"][0]["detail"] for item in calls) assert all("blob=" in item["evidence"][0]["detail"] for item in calls) + + +def test_explicit_from_import_creates_import_grounded_call_edge(tmp_path): + root = repo(tmp_path) + write( + root, + "external_calls.py", + "from remote.worker import execute as run_remote\n\n" + "def caller():\n" + " run_remote()\n", + ) + git(root, "add", "external_calls.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 + call = calls[0] + assert call["source_id"] == "sym::external_calls.py::caller" + assert call["target_id"] == "sym::external_calls.py::import::remote.worker" + assert call["resolution_method"] == "import_grounded" + assert call["confidence"] is None + assert call["evidence"][0]["kind"] == "call_site" + assert call["evidence"][0]["snippet"] == "run_remote()" + assert "query=python-structure-v2" in call["evidence"][0]["detail"] + + +def test_rebound_import_does_not_create_import_grounded_call_edge(tmp_path): + root = repo(tmp_path) + write( + root, + "rebound_import.py", + "from remote.worker import execute as run_remote\n\n" + "run_remote = lambda: None\n\n" + "def caller():\n" + " run_remote()\n", + ) + git(root, "add", "rebound_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 calls == [] + + +@pytest.mark.parametrize("declaration", ["def run_remote():\n return None", "class run_remote:\n pass"]) +def test_module_declaration_rebinds_imported_alias(tmp_path, declaration): + root = repo(tmp_path) + write( + root, + "declaration_rebind.py", + "from remote.worker import execute as run_remote\n\n" + + declaration + "\n\n" + "def caller():\n" + " run_remote()\n", + ) + git(root, "add", "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 all(item["resolution_method"] != "import_grounded" for item in calls) + + +def test_import_binding_remains_visible_before_later_rebind(tmp_path): + root = repo(tmp_path) + write( + root, + "line_aware_rebind.py", + "from remote.worker import execute as run_remote\n\n" + "run_remote()\n\n" + "run_remote = lambda: None\n", + ) + git(root, "add", "line_aware_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 len(calls) == 1 + assert calls[0]["target_id"] == "sym::line_aware_rebind.py::import::remote.worker" + assert calls[0]["resolution_method"] == "import_grounded" + + +def test_later_import_does_not_hide_earlier_local_call(tmp_path): + root = repo(tmp_path) + write( + root, + "local_before_import.py", + "def run():\n" + " return None\n\n" + "run()\n\n" + "from remote.worker import execute as run\n", + ) + git(root, "add", "local_before_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]["target_id"] == "sym::local_before_import.py::run" + assert calls[0]["resolution_method"] == "resolved"