2196 lines
99 KiB
Python
2196 lines
99 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections import ChainMap
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Mapping
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.code_dependency_graph import (
|
|
BindingSite,
|
|
CodeDependencyGraph,
|
|
DeclarationSite,
|
|
DependencyDelta,
|
|
DependencySelector,
|
|
DependencyType,
|
|
IndirectReference,
|
|
ParentRelationship,
|
|
ReturnSite,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph_maintainer import (
|
|
CodeDependencyGraphMaintainer,
|
|
GraphSignature,
|
|
)
|
|
from band_aider.judgement.language_glob_strategy import LanguageGlobStrategy
|
|
from band_aider.judgement.repo_exec import RepoExec, current_repo_exec
|
|
from band_aider.judgement.source_discovery_strategy import SearchableBeat
|
|
|
|
|
|
@dataclass
|
|
class DependencyUpdateInstrumentation:
|
|
files_parsed: set[str] = field(default_factory=set)
|
|
declaration_sites_added: set[str] = field(default_factory=set)
|
|
declaration_sites_removed: set[str] = field(default_factory=set)
|
|
binding_sites_added: set[str] = field(default_factory=set)
|
|
binding_sites_removed: set[str] = field(default_factory=set)
|
|
return_sites_added: set[str] = field(default_factory=set)
|
|
return_sites_removed: set[str] = field(default_factory=set)
|
|
indirect_references_added: set[str] = field(default_factory=set)
|
|
indirect_references_removed: set[str] = field(default_factory=set)
|
|
indirect_references_revalidated: set[str] = field(default_factory=set)
|
|
parent_relationships_traversed: int = 0
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _StaticFacts:
|
|
"""Python discovery evidence used privately to maintain static source pairs."""
|
|
|
|
module_name: str
|
|
imported_modules: frozenset[str] = frozenset()
|
|
imported_names: frozenset[tuple[str, str, str]] = frozenset()
|
|
referenced_names: frozenset[str] = frozenset()
|
|
defined_public_symbols: frozenset[str] = frozenset()
|
|
|
|
|
|
@dataclass
|
|
class PythonAstDependencyGraphMaintainer(CodeDependencyGraphMaintainer):
|
|
"""Convert generic discovery beats into the live dependency graph."""
|
|
|
|
repo_exec: RepoExec = field(default_factory=current_repo_exec)
|
|
source_discovery: LanguageGlobStrategy | None = None
|
|
_graph: CodeDependencyGraph = field(default_factory=CodeDependencyGraph)
|
|
signature: GraphSignature | None = None
|
|
_is_frozen: bool = field(default=False, init=False, repr=False)
|
|
last_update_instrumentation: DependencyUpdateInstrumentation = field(
|
|
default_factory=DependencyUpdateInstrumentation,
|
|
init=False,
|
|
)
|
|
_beats_by_source: dict[str, tuple[SearchableBeat, ...]] = field(default_factory=dict, init=False, repr=False)
|
|
_type_kinds: dict[str, str] = field(default_factory=dict, init=False, repr=False)
|
|
_type_members: dict[str, tuple[frozenset[str], frozenset[str]]] = field(default_factory=dict, init=False, repr=False)
|
|
_callable_parameters: dict[str, tuple[tuple[str, str, int | None, str], ...]] = field(default_factory=dict, init=False, repr=False)
|
|
_callable_sources: dict[str, str] = field(default_factory=dict, init=False, repr=False)
|
|
_callable_ids_by_source: dict[str, set[str]] = field(default_factory=dict, init=False, repr=False)
|
|
_aliases_by_source: dict[str, dict[str, str | None]] = field(default_factory=dict, init=False, repr=False)
|
|
_static_facts_by_location: dict[str, _StaticFacts] = field(default_factory=dict, init=False, repr=False)
|
|
_test_facts_by_location: dict[str, _StaticFacts] = field(default_factory=dict, init=False, repr=False)
|
|
|
|
@property
|
|
def graph(self) -> CodeDependencyGraph:
|
|
return self._graph
|
|
|
|
@property
|
|
def is_frozen(self) -> bool:
|
|
return self._is_frozen or self._graph.is_frozen
|
|
|
|
def freeze(self) -> None:
|
|
if not self._is_frozen:
|
|
self._graph.freeze()
|
|
self._is_frozen = True
|
|
|
|
def _assert_mutable(self) -> None:
|
|
if self.is_frozen:
|
|
raise RuntimeError("PythonAstDependencyGraphMaintainer is frozen and cannot be mutated.")
|
|
|
|
@staticmethod
|
|
def _attrs(beat: SearchableBeat) -> dict[str, str]:
|
|
return dict(beat.attributes)
|
|
|
|
@staticmethod
|
|
def _json_list(value: str | None) -> list:
|
|
if not value:
|
|
return []
|
|
try:
|
|
result = json.loads(value)
|
|
except (TypeError, ValueError):
|
|
return []
|
|
return result if isinstance(result, list) else []
|
|
|
|
@staticmethod
|
|
def _dedupe(files: list[ClassMeta]) -> list[ClassMeta]:
|
|
result: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for item in files:
|
|
if item.location not in seen:
|
|
seen.add(item.location)
|
|
result.append(item)
|
|
return result
|
|
|
|
def _location_path(self, repo_root: Path, location: str) -> Path:
|
|
path = Path(location)
|
|
return path if path.is_absolute() else repo_root / path
|
|
|
|
def _class_meta_kind(
|
|
self,
|
|
class_meta: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> str:
|
|
"""Return the owned source/test population for one ClassMeta."""
|
|
|
|
location = class_meta.location
|
|
source_locations = {item.location for item in source_files}
|
|
test_locations = {item.location for item in test_files}
|
|
if location in source_locations or location in self._graph.source_by_location:
|
|
return "source"
|
|
if location in test_locations or location in self._graph.test_by_location:
|
|
return "test"
|
|
path = self._location_path(repo_root, location)
|
|
try:
|
|
path.relative_to(src_root)
|
|
return "source"
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
path.relative_to(test_root)
|
|
return "test"
|
|
except ValueError as error:
|
|
raise ValueError(f"ClassMeta location is outside source and test roots: {location}") from error
|
|
|
|
def _missing_package_init_paths(self, file_path: Path, root: Path) -> list[Path]:
|
|
try:
|
|
relative = file_path.relative_to(root)
|
|
except ValueError:
|
|
return []
|
|
missing: list[Path] = []
|
|
current = root
|
|
for part in relative.parts[:-1]:
|
|
current = current / part
|
|
init_path = current / "__init__.py"
|
|
if not self.repo_exec.exists(init_path):
|
|
missing.append(init_path)
|
|
return missing
|
|
|
|
def _ensure_package_chain_for_new_file(self, location: str, repo_root: Path, root: Path) -> None:
|
|
for init_path in self._missing_package_init_paths(self._location_path(repo_root, location), root):
|
|
self.repo_exec.write_text(init_path, "", encoding="utf-8")
|
|
|
|
def _configure_discovery(
|
|
self,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
repo_root: Path,
|
|
) -> LanguageGlobStrategy:
|
|
if self.source_discovery is None:
|
|
from band_aider.strategies.python.python_source_discovery import PythonSourceDiscovery
|
|
|
|
self.source_discovery = PythonSourceDiscovery(
|
|
src_root=src_root,
|
|
test_root=test_root,
|
|
repo_exec=self.repo_exec,
|
|
repo_root=repo_root,
|
|
)
|
|
return self.source_discovery
|
|
|
|
def _discover(
|
|
self,
|
|
locations: set[str],
|
|
source_by_location: dict[str, ClassMeta],
|
|
repo_root: Path,
|
|
discovery: LanguageGlobStrategy,
|
|
) -> dict[str, tuple[SearchableBeat, ...]]:
|
|
result: dict[str, tuple[SearchableBeat, ...]] = {}
|
|
for location in sorted(locations):
|
|
if location not in source_by_location:
|
|
continue
|
|
try:
|
|
source = self.repo_exec.read_text(self._location_path(repo_root, location))
|
|
values = discovery.searchable_beats(location, source)
|
|
except (OSError, SyntaxError, UnicodeDecodeError):
|
|
continue
|
|
beats = tuple(value for value in values if isinstance(value, SearchableBeat))
|
|
result[location] = beats
|
|
self.last_update_instrumentation.files_parsed.add(location)
|
|
return result
|
|
|
|
def _private_indexes_from_beats(
|
|
self,
|
|
beats_by_source: Mapping[str, tuple[SearchableBeat, ...]],
|
|
) -> tuple[
|
|
dict[str, str],
|
|
dict[str, tuple[frozenset[str], frozenset[str]]],
|
|
dict[str, tuple[tuple[str, str, int | None, str], ...]],
|
|
dict[str, str],
|
|
dict[str, set[str]],
|
|
dict[str, dict[str, str | None]],
|
|
]:
|
|
"""Derive private Python resolution facts from one beat population."""
|
|
|
|
type_kinds: dict[str, str] = {}
|
|
type_members: dict[str, tuple[frozenset[str], frozenset[str]]] = {}
|
|
callable_parameters: dict[str, tuple[tuple[str, str, int | None, str], ...]] = {}
|
|
callable_sources: dict[str, str] = {}
|
|
callable_ids_by_source: dict[str, set[str]] = {}
|
|
aliases_by_source: dict[str, dict[str, str | None]] = {}
|
|
for location, beats in beats_by_source.items():
|
|
aliases: dict[str, str | None] = {}
|
|
for beat in beats:
|
|
attrs = self._attrs(beat)
|
|
if beat.kind == "import":
|
|
local = attrs.get("alias") or attrs.get("module", "").split(".")[0]
|
|
imported = attrs.get("module", "")
|
|
if local and imported:
|
|
previous = aliases.get(local, imported)
|
|
aliases[local] = imported if previous == imported else None
|
|
elif beat.kind == "import_from":
|
|
local = attrs.get("alias") or attrs.get("name", "")
|
|
module = attrs.get("module", "")
|
|
name = attrs.get("name", "")
|
|
if local and module and name != "*":
|
|
imported = f"{module}.{name}"
|
|
previous = aliases.get(local, imported)
|
|
aliases[local] = imported if previous == imported else None
|
|
elif beat.kind == "dependency_type":
|
|
type_id = attrs.get("type_id", "")
|
|
if type_id:
|
|
type_kinds[type_id] = attrs.get("direct_abstraction_kind", "concrete")
|
|
type_members[type_id] = (
|
|
frozenset(str(item) for item in self._json_list(attrs.get("member_methods"))),
|
|
frozenset(str(item) for item in self._json_list(attrs.get("member_fields"))),
|
|
)
|
|
elif beat.kind == "dependency_callable":
|
|
callable_id = attrs.get("callable_id", "")
|
|
if callable_id:
|
|
parameters: list[tuple[str, str, int | None, str]] = []
|
|
for child in beat.children:
|
|
child_attrs = self._attrs(child)
|
|
ordinal_value = child_attrs.get("ordinal", "")
|
|
parameters.append((
|
|
child_attrs.get("name", ""),
|
|
child_attrs.get("parameter_kind", ""),
|
|
int(ordinal_value) if ordinal_value not in {"", "None"} else None,
|
|
child_attrs.get("annotation", ""),
|
|
))
|
|
if attrs.get("owner_type_id"):
|
|
parameters = [item for item in parameters if item[0] not in {"self", "cls"}]
|
|
positional_ordinal = 0
|
|
normalized: list[tuple[str, str, int | None, str]] = []
|
|
for name, kind, ordinal, annotation in parameters:
|
|
if kind in {"positional_only", "positional_or_keyword"}:
|
|
normalized.append((name, kind, positional_ordinal, annotation))
|
|
positional_ordinal += 1
|
|
else:
|
|
normalized.append((name, kind, ordinal, annotation))
|
|
parameters = normalized
|
|
callable_parameters[callable_id] = tuple(parameters)
|
|
callable_sources[callable_id] = location
|
|
callable_ids_by_source.setdefault(location, set()).add(callable_id)
|
|
aliases_by_source[location] = aliases
|
|
return (
|
|
type_kinds,
|
|
type_members,
|
|
callable_parameters,
|
|
callable_sources,
|
|
callable_ids_by_source,
|
|
aliases_by_source,
|
|
)
|
|
|
|
def _refresh_private_beat_indexes(self, beats_by_source: Mapping[str, tuple[SearchableBeat, ...]]) -> None:
|
|
"""Rebuild Python-only resolution facts at an explicit full-build boundary."""
|
|
|
|
(
|
|
self._type_kinds,
|
|
self._type_members,
|
|
self._callable_parameters,
|
|
self._callable_sources,
|
|
self._callable_ids_by_source,
|
|
self._aliases_by_source,
|
|
) = self._private_indexes_from_beats(beats_by_source)
|
|
|
|
def _expand_alias(self, name: str, aliases: dict[str, str | None]) -> str | None:
|
|
if not name:
|
|
return None
|
|
first, separator, rest = name.partition(".")
|
|
if first in aliases:
|
|
value = aliases[first]
|
|
if value is None:
|
|
return None
|
|
return f"{value}.{rest}" if separator else value
|
|
return name
|
|
|
|
def _resolve_type(
|
|
self,
|
|
name: str,
|
|
types: Mapping[str, DependencyType | None],
|
|
location: str,
|
|
aliases_by_source: Mapping[str, Mapping[str, str | None]] | None = None,
|
|
) -> str | None:
|
|
aliases = aliases_by_source if aliases_by_source is not None else self._aliases_by_source
|
|
expanded = self._expand_alias(name, dict(aliases.get(location, {})))
|
|
if not expanded:
|
|
return None
|
|
exact = [
|
|
key
|
|
for key, value in types.items()
|
|
if isinstance(value, DependencyType) and value.qualified_name == expanded
|
|
]
|
|
if len(exact) == 1:
|
|
return exact[0]
|
|
simple = [
|
|
key
|
|
for key, value in types.items()
|
|
if isinstance(value, DependencyType)
|
|
and value.local_name == expanded.rsplit(".", 1)[-1]
|
|
]
|
|
return simple[0] if len(simple) == 1 else None
|
|
|
|
def _resolve_callable(
|
|
self,
|
|
name: str,
|
|
location: str,
|
|
types: Mapping[str, DependencyType | None],
|
|
receiver: str = "",
|
|
aliases_by_source: Mapping[str, Mapping[str, str | None]] | None = None,
|
|
callable_parameters: Mapping[str, tuple[tuple[str, str, int | None, str], ...] | None] | None = None,
|
|
) -> tuple[str | None, str]:
|
|
aliases = aliases_by_source if aliases_by_source is not None else self._aliases_by_source
|
|
parameters = callable_parameters if callable_parameters is not None else self._callable_parameters
|
|
expanded = self._expand_alias(name, dict(aliases.get(location, {}))) or ""
|
|
type_id = self._resolve_type(expanded, types, location, aliases)
|
|
if type_id is not None:
|
|
type_record = types.get(type_id)
|
|
if isinstance(type_record, DependencyType):
|
|
return f"{type_record.qualified_name}.__init__", "constructor"
|
|
if receiver:
|
|
receiver_id = self._resolve_type(receiver, types, location, aliases)
|
|
if receiver_id is not None:
|
|
receiver_record = types.get(receiver_id)
|
|
owner = (
|
|
f"{receiver_record.qualified_name}.{expanded.rsplit('.', 1)[-1]}"
|
|
if isinstance(receiver_record, DependencyType)
|
|
else ""
|
|
)
|
|
if owner in parameters and parameters[owner] is not None:
|
|
return owner, "method"
|
|
if expanded in parameters and parameters[expanded] is not None:
|
|
return expanded, "function"
|
|
matches = [
|
|
item
|
|
for item, value in parameters.items()
|
|
if value is not None
|
|
and item.rsplit(".", 1)[-1] == expanded.rsplit(".", 1)[-1]
|
|
]
|
|
if len(matches) != 1:
|
|
return None, "function"
|
|
callable_id = matches[0]
|
|
owner = callable_id.rsplit(".", 1)[0]
|
|
kind = "method" if any(
|
|
isinstance(value, DependencyType) and value.qualified_name == owner
|
|
for value in types.values()
|
|
) else "function"
|
|
return callable_id, kind
|
|
|
|
def _declaration_for_call(
|
|
self,
|
|
location: str,
|
|
attrs: Mapping[str, str],
|
|
types: Mapping[str, DependencyType | None],
|
|
aliases_by_source: Mapping[str, Mapping[str, str | None]] | None = None,
|
|
callable_parameters: Mapping[str, tuple[tuple[str, str, int | None, str], ...] | None] | None = None,
|
|
callable_sources: Mapping[str, str | None] | None = None,
|
|
) -> tuple[str | None, str | None, str | None, int | None]:
|
|
"""Resolve one call argument to its exact tracked parameter declaration."""
|
|
|
|
if attrs.get("target_callable_is_local") == "True":
|
|
return None, None, None, None
|
|
callable_id, call_kind = self._resolve_callable(
|
|
attrs.get("target_callable_expression", ""),
|
|
location,
|
|
types,
|
|
attrs.get("receiver_expression", ""),
|
|
aliases_by_source,
|
|
callable_parameters,
|
|
)
|
|
if callable_id is None:
|
|
return None, None, None, None
|
|
parameters_by_callable = (
|
|
callable_parameters if callable_parameters is not None else self._callable_parameters
|
|
)
|
|
parameters = parameters_by_callable.get(callable_id) or ()
|
|
name = attrs.get("argument_name", "")
|
|
ordinal_value = attrs.get("positional_ordinal", "")
|
|
ordinal = int(ordinal_value) if ordinal_value not in {"", "None"} else None
|
|
entry: tuple[str, str, int | None, str] | None = None
|
|
if attrs.get("binding_kind") == "call_positional" and ordinal is not None:
|
|
positional = [
|
|
item
|
|
for item in parameters
|
|
if item[1] in {"positional_only", "positional_or_keyword"}
|
|
]
|
|
entry = positional[ordinal] if ordinal < len(positional) else None
|
|
elif name:
|
|
matches = [item for item in parameters if item[0] == name]
|
|
entry = matches[0] if len(matches) == 1 else None
|
|
if entry is None:
|
|
return None, callable_id, call_kind, ordinal
|
|
parameter_name, _parameter_kind, parameter_ordinal, _annotation = entry
|
|
declaration_kind = (
|
|
"constructor_parameter"
|
|
if call_kind == "constructor"
|
|
else "method_parameter"
|
|
if call_kind == "method"
|
|
else "function_parameter"
|
|
)
|
|
sources_by_callable = callable_sources if callable_sources is not None else self._callable_sources
|
|
declaration_source = sources_by_callable.get(callable_id) or location
|
|
declaration_id = "|".join(
|
|
(declaration_source, callable_id, declaration_kind, parameter_name)
|
|
)
|
|
return declaration_id, callable_id, call_kind, parameter_ordinal
|
|
|
|
def _records_from_beats(
|
|
self,
|
|
beats_by_source: Mapping[str, tuple[SearchableBeat, ...]],
|
|
source_by_location: Mapping[str, ClassMeta],
|
|
*,
|
|
base_types: Mapping[str, DependencyType | None] | None = None,
|
|
base_relationships: Mapping[str, ParentRelationship | None] | None = None,
|
|
base_declarations: Mapping[str, DeclarationSite | None] | None = None,
|
|
type_kinds: Mapping[str, str | None] | None = None,
|
|
aliases_by_source: Mapping[str, Mapping[str, str | None]] | None = None,
|
|
callable_parameters: Mapping[str, tuple[tuple[str, str, int | None, str], ...] | None] | None = None,
|
|
callable_sources: Mapping[str, str | None] | None = None,
|
|
refresh_private_indexes: bool = True,
|
|
) -> tuple[
|
|
dict[str, DependencyType],
|
|
dict[str, ParentRelationship],
|
|
dict[str, DeclarationSite],
|
|
dict[str, BindingSite],
|
|
dict[str, ReturnSite],
|
|
dict[str, _StaticFacts],
|
|
]:
|
|
if refresh_private_indexes:
|
|
self._refresh_private_beat_indexes(beats_by_source)
|
|
type_kinds = type_kinds if type_kinds is not None else self._type_kinds
|
|
aliases_by_source = aliases_by_source if aliases_by_source is not None else self._aliases_by_source
|
|
callable_parameters = (
|
|
callable_parameters
|
|
if callable_parameters is not None
|
|
else self._callable_parameters
|
|
)
|
|
callable_sources = callable_sources if callable_sources is not None else self._callable_sources
|
|
types: dict[str, DependencyType] = {}
|
|
for location, beats in beats_by_source.items():
|
|
source = source_by_location.get(location)
|
|
if source is None:
|
|
continue
|
|
for beat in beats:
|
|
if beat.kind != "dependency_type":
|
|
continue
|
|
attrs = self._attrs(beat)
|
|
type_id = attrs.get("type_id", "")
|
|
if type_id:
|
|
types[type_id] = DependencyType(
|
|
type_id=type_id,
|
|
qualified_name=attrs.get("qualified_name", ""),
|
|
local_name=attrs.get("local_name", ""),
|
|
source=source,
|
|
)
|
|
|
|
type_population: Mapping[str, DependencyType | None] = (
|
|
ChainMap(types, base_types) if base_types is not None else types
|
|
)
|
|
relationships: dict[str, ParentRelationship] = {}
|
|
for location, beats in beats_by_source.items():
|
|
for beat in beats:
|
|
if beat.kind != "parent_relationship":
|
|
continue
|
|
attrs = self._attrs(beat)
|
|
child_id = attrs.get("child_type_id", "")
|
|
parent_id = self._resolve_type(
|
|
attrs.get("parent_name", ""),
|
|
type_population,
|
|
location,
|
|
aliases_by_source,
|
|
)
|
|
if (
|
|
isinstance(type_population.get(child_id), DependencyType)
|
|
and isinstance(type_population.get(parent_id), DependencyType)
|
|
and child_id != parent_id
|
|
):
|
|
record = ParentRelationship(
|
|
parent_type_id=parent_id,
|
|
child_type_id=child_id,
|
|
source_location=location,
|
|
line=beat.start_line or 0,
|
|
column=0,
|
|
)
|
|
relationships[record.relationship_id] = record
|
|
|
|
relationship_population: Mapping[str, ParentRelationship | None] = (
|
|
ChainMap(relationships, base_relationships)
|
|
if base_relationships is not None
|
|
else relationships
|
|
)
|
|
eligible = {
|
|
type_id
|
|
for type_id, kind in type_kinds.items()
|
|
if kind in {"protocol", "abc"}
|
|
}
|
|
eligible.update(
|
|
record.parent_type_id
|
|
for record in relationship_population.values()
|
|
if isinstance(record, ParentRelationship)
|
|
)
|
|
declarations: dict[str, DeclarationSite] = {}
|
|
for location, beats in beats_by_source.items():
|
|
source = source_by_location.get(location)
|
|
if source is None:
|
|
continue
|
|
for beat in beats:
|
|
if beat.kind != "declaration_site":
|
|
continue
|
|
attrs = self._attrs(beat)
|
|
declared = self._resolve_type(
|
|
attrs.get("declared_name", ""),
|
|
type_population,
|
|
location,
|
|
aliases_by_source,
|
|
)
|
|
site_id = attrs.get("declaration_site_id", "")
|
|
if declared not in eligible or not site_id:
|
|
continue
|
|
declarations[site_id] = DeclarationSite(
|
|
declaration_site_id=site_id,
|
|
declared_type_id=declared,
|
|
source=source,
|
|
callable_id=attrs.get("callable_id") or None,
|
|
field_owner_id=attrs.get("field_owner_id") or None,
|
|
receiving_name=attrs.get("receiving_name", ""),
|
|
declaration_kind=attrs.get("declaration_kind", ""),
|
|
line=beat.start_line or 0,
|
|
column=0,
|
|
)
|
|
|
|
declaration_population: Mapping[str, DeclarationSite | None] = (
|
|
ChainMap(declarations, base_declarations)
|
|
if base_declarations is not None
|
|
else declarations
|
|
)
|
|
bindings: dict[str, BindingSite] = {}
|
|
for location, beats in beats_by_source.items():
|
|
source = source_by_location.get(location)
|
|
if source is None:
|
|
continue
|
|
for beat in beats:
|
|
if beat.kind != "binding_site":
|
|
continue
|
|
attrs = self._attrs(beat)
|
|
declared_site_id = attrs.get("declaration_site_id", "")
|
|
if attrs.get("binding_kind", "").startswith("call_"):
|
|
resolved_id, _callable_id, _call_kind, _ordinal = self._declaration_for_call(
|
|
location,
|
|
attrs,
|
|
type_population,
|
|
aliases_by_source,
|
|
callable_parameters,
|
|
callable_sources,
|
|
)
|
|
declared_site_id = resolved_id or ""
|
|
if not isinstance(declaration_population.get(declared_site_id), DeclarationSite):
|
|
continue
|
|
actual_id: str | None = None
|
|
returned_callable_id: str | None = None
|
|
value_kind = attrs.get("value_kind", "")
|
|
expression = attrs.get("actual_type_expression", "") or attrs.get("returned_callable_expression", "")
|
|
if value_kind == "direct":
|
|
actual_id = self._resolve_type(
|
|
expression,
|
|
type_population,
|
|
location,
|
|
aliases_by_source,
|
|
)
|
|
elif (
|
|
value_kind == "call"
|
|
and attrs.get("returned_callable_is_local") != "True"
|
|
):
|
|
callable_id, _kind = self._resolve_callable(
|
|
expression,
|
|
location,
|
|
type_population,
|
|
attrs.get("receiver_expression", ""),
|
|
aliases_by_source,
|
|
callable_parameters,
|
|
)
|
|
if callable_id is not None:
|
|
constructor_type = next(
|
|
(
|
|
item
|
|
for item, value in type_population.items()
|
|
if isinstance(value, DependencyType)
|
|
and f"{value.qualified_name}.__init__" == callable_id
|
|
),
|
|
None,
|
|
)
|
|
actual_id = constructor_type
|
|
returned_callable_id = None if actual_id is not None else callable_id
|
|
binding_kind = attrs.get("binding_kind", "")
|
|
if binding_kind.startswith("call_"):
|
|
_resolved_id, _resolved_callable, call_kind, _resolved_ordinal = self._declaration_for_call(
|
|
location,
|
|
attrs,
|
|
type_population,
|
|
aliases_by_source,
|
|
callable_parameters,
|
|
callable_sources,
|
|
)
|
|
if call_kind is not None:
|
|
binding_kind = f"{call_kind}_{binding_kind[5:]}"
|
|
record = BindingSite(
|
|
binding_site_id=attrs.get("binding_site_id", ""),
|
|
declaration_site_id=declared_site_id,
|
|
source=source,
|
|
binding_kind=binding_kind,
|
|
direct_actual_type_id=actual_id,
|
|
returned_callable_id=returned_callable_id,
|
|
line=beat.start_line or 0,
|
|
column=0,
|
|
)
|
|
if record.binding_site_id:
|
|
bindings[record.binding_site_id] = record
|
|
|
|
returns: dict[str, ReturnSite] = {}
|
|
for location, beats in beats_by_source.items():
|
|
source = source_by_location.get(location)
|
|
if source is None:
|
|
continue
|
|
for beat in beats:
|
|
if beat.kind != "return_site":
|
|
continue
|
|
attrs = self._attrs(beat)
|
|
declared_id = self._resolve_type(
|
|
attrs.get("declared_return_name", ""),
|
|
type_population,
|
|
location,
|
|
aliases_by_source,
|
|
)
|
|
if declared_id not in eligible:
|
|
continue
|
|
actual_id = self._resolve_type(
|
|
attrs.get("actual_type_expression", ""),
|
|
type_population,
|
|
location,
|
|
aliases_by_source,
|
|
)
|
|
returned_callable_id: str | None = None
|
|
if (
|
|
actual_id is None
|
|
and attrs.get("returned_callable_expression")
|
|
and attrs.get("returned_callable_is_local") != "True"
|
|
):
|
|
returned_callable_id, _kind = self._resolve_callable(
|
|
attrs["returned_callable_expression"],
|
|
location,
|
|
type_population,
|
|
aliases_by_source=aliases_by_source,
|
|
callable_parameters=callable_parameters,
|
|
)
|
|
if returned_callable_id is not None:
|
|
constructor_type = next(
|
|
(
|
|
item
|
|
for item, value in type_population.items()
|
|
if isinstance(value, DependencyType)
|
|
and f"{value.qualified_name}.__init__" == returned_callable_id
|
|
),
|
|
None,
|
|
)
|
|
if constructor_type is not None:
|
|
actual_id = constructor_type
|
|
returned_callable_id = None
|
|
record = ReturnSite(
|
|
return_site_id=attrs.get("return_site_id", ""),
|
|
callable_id=attrs.get("callable_id", ""),
|
|
declared_return_type_id=declared_id,
|
|
source=source,
|
|
direct_actual_type_id=actual_id,
|
|
returned_callable_id=returned_callable_id,
|
|
line=beat.start_line or 0,
|
|
column=0,
|
|
)
|
|
if record.return_site_id and record.callable_id:
|
|
returns[record.return_site_id] = record
|
|
|
|
static_facts: dict[str, _StaticFacts] = {}
|
|
for location, beats in beats_by_source.items():
|
|
fact_beat = next((beat for beat in beats if beat.kind == "static_facts"), None)
|
|
if fact_beat is None:
|
|
continue
|
|
attrs = self._attrs(fact_beat)
|
|
static_facts[location] = _StaticFacts(
|
|
module_name=attrs.get("module_name", ""),
|
|
imported_modules=frozenset(str(item) for item in self._json_list(attrs.get("imported_modules"))),
|
|
imported_names=frozenset(tuple(item) for item in self._json_list(attrs.get("imported_names")) if isinstance(item, list) and len(item) == 3),
|
|
referenced_names=frozenset(str(item) for item in self._json_list(attrs.get("referenced_names"))),
|
|
defined_public_symbols=frozenset(str(item) for item in self._json_list(attrs.get("defined_public_symbols"))),
|
|
)
|
|
return types, relationships, declarations, bindings, returns, static_facts
|
|
|
|
def _static_pair_exists(self, candidate: _StaticFacts, dependency: _StaticFacts) -> bool:
|
|
return (
|
|
dependency.module_name in candidate.imported_modules
|
|
or any(module == dependency.module_name for module, _name, _alias in candidate.imported_names)
|
|
)
|
|
|
|
def _static_pairs(self, facts: Mapping[str, _StaticFacts]) -> set[tuple[str, str]]:
|
|
return {
|
|
(candidate, dependency)
|
|
for candidate, candidate_facts in facts.items()
|
|
for dependency, dependency_facts in facts.items()
|
|
if candidate != dependency and self._static_pair_exists(candidate_facts, dependency_facts)
|
|
}
|
|
|
|
def _static_pair_delta_for_changed(
|
|
self,
|
|
old_dependencies: Mapping[str, set[str]],
|
|
old_dependents: Mapping[str, set[str]],
|
|
changed_facts: Mapping[str, _StaticFacts],
|
|
changed_locations: set[str],
|
|
) -> tuple[set[tuple[str, str]], set[tuple[str, str]]]:
|
|
"""Return only static pairs touching changed facts without rebuilding facts."""
|
|
|
|
old_touched: set[tuple[str, str]] = set()
|
|
for location in changed_locations:
|
|
old_touched.update(
|
|
(location, dependency)
|
|
for dependency in old_dependencies.get(location, set())
|
|
)
|
|
old_touched.update(
|
|
(dependent, location)
|
|
for dependent in old_dependents.get(location, set())
|
|
)
|
|
locations = set(self._static_facts_by_location) | set(changed_facts)
|
|
|
|
def fact_for(location: str) -> _StaticFacts | None:
|
|
if location in changed_locations:
|
|
return changed_facts.get(location)
|
|
return self._static_facts_by_location.get(location)
|
|
|
|
new_touched: set[tuple[str, str]] = set()
|
|
for candidate in changed_locations:
|
|
candidate_facts = fact_for(candidate)
|
|
if candidate_facts is None:
|
|
continue
|
|
for dependency in locations:
|
|
dependency_facts = fact_for(dependency)
|
|
if (
|
|
dependency != candidate
|
|
and dependency_facts is not None
|
|
and self._static_pair_exists(candidate_facts, dependency_facts)
|
|
):
|
|
new_touched.add((candidate, dependency))
|
|
for dependency in changed_locations:
|
|
dependency_facts = fact_for(dependency)
|
|
if dependency_facts is None:
|
|
continue
|
|
for candidate in locations:
|
|
candidate_facts = fact_for(candidate)
|
|
if (
|
|
candidate != dependency
|
|
and candidate_facts is not None
|
|
and self._static_pair_exists(candidate_facts, dependency_facts)
|
|
):
|
|
new_touched.add((candidate, dependency))
|
|
return new_touched - old_touched, old_touched - new_touched
|
|
|
|
def _nominal_path(
|
|
self,
|
|
actual_type_id: str,
|
|
declared_type_id: str,
|
|
relationships: Mapping[str, ParentRelationship | None],
|
|
) -> tuple[str, ...] | None:
|
|
if actual_type_id == declared_type_id:
|
|
return ()
|
|
by_child: dict[str, list[ParentRelationship]] = {}
|
|
for relationship in relationships.values():
|
|
if not isinstance(relationship, ParentRelationship):
|
|
continue
|
|
by_child.setdefault(relationship.child_type_id, []).append(relationship)
|
|
queue: list[tuple[str, tuple[str, ...]]] = [(actual_type_id, ())]
|
|
seen = {actual_type_id}
|
|
while queue:
|
|
current, path = queue.pop(0)
|
|
for relationship in sorted(by_child.get(current, ()), key=lambda item: item.relationship_id):
|
|
self.last_update_instrumentation.parent_relationships_traversed += 1
|
|
next_path = path + (relationship.relationship_id,)
|
|
if relationship.parent_type_id == declared_type_id:
|
|
return next_path
|
|
if relationship.parent_type_id not in seen:
|
|
seen.add(relationship.parent_type_id)
|
|
queue.append((relationship.parent_type_id, next_path))
|
|
return None
|
|
|
|
@staticmethod
|
|
def _return_sites_by_callable(
|
|
returns: Mapping[str, ReturnSite],
|
|
) -> dict[str, tuple[ReturnSite, ...]]:
|
|
"""Index one provisional return population by producing callable."""
|
|
|
|
indexed: dict[str, list[ReturnSite]] = {}
|
|
for site in returns.values():
|
|
if not isinstance(site, ReturnSite):
|
|
continue
|
|
indexed.setdefault(site.callable_id, []).append(site)
|
|
return {
|
|
callable_id: tuple(sorted(sites, key=lambda site: site.return_site_id))
|
|
for callable_id, sites in indexed.items()
|
|
}
|
|
|
|
def _return_provenance_for_callable(
|
|
self,
|
|
callable_id: str,
|
|
return_sites_by_callable: Mapping[str, tuple[ReturnSite, ...]],
|
|
) -> tuple[str, tuple[str, ...]] | None:
|
|
"""Resolve one bounded callable-return chain to a direct actual type."""
|
|
|
|
return_site_ids: list[str] = []
|
|
seen: set[str] = set()
|
|
current_callable_id: str | None = callable_id
|
|
while current_callable_id is not None:
|
|
if current_callable_id in seen:
|
|
return None
|
|
seen.add(current_callable_id)
|
|
candidates = return_sites_by_callable.get(current_callable_id, ())
|
|
if len(candidates) != 1:
|
|
return None
|
|
site = candidates[0]
|
|
return_site_ids.append(site.return_site_id)
|
|
if site.direct_actual_type_id is not None:
|
|
return site.direct_actual_type_id, tuple(return_site_ids)
|
|
current_callable_id = site.returned_callable_id
|
|
return None
|
|
|
|
def _binding_value_provenance(
|
|
self,
|
|
binding: BindingSite,
|
|
return_sites_by_callable: Mapping[str, tuple[ReturnSite, ...]],
|
|
) -> tuple[str, tuple[str, ...]] | None:
|
|
"""Resolve the direct or returned value supplied by one binding site."""
|
|
|
|
if binding.direct_actual_type_id is not None:
|
|
return binding.direct_actual_type_id, ()
|
|
if binding.returned_callable_id is None:
|
|
return None
|
|
return self._return_provenance_for_callable(
|
|
binding.returned_callable_id,
|
|
return_sites_by_callable,
|
|
)
|
|
|
|
def _callables_affected_by_return_changes(
|
|
self,
|
|
changed_return_ids: set[str],
|
|
provisional_returns: Mapping[str, ReturnSite | None],
|
|
local_return_ids_by_returned_callable: Mapping[str, set[str]],
|
|
) -> set[str]:
|
|
"""Use return indexes to find changed producers and bounded callers."""
|
|
|
|
affected = {
|
|
site.callable_id
|
|
for return_id in changed_return_ids
|
|
if isinstance(site := provisional_returns.get(return_id), ReturnSite)
|
|
}
|
|
pending = list(affected)
|
|
while pending:
|
|
callable_id = pending.pop()
|
|
return_ids = (
|
|
self._graph.return_site_ids_by_returned_callable.get(callable_id, set())
|
|
| local_return_ids_by_returned_callable.get(callable_id, set())
|
|
)
|
|
for return_id in return_ids:
|
|
site = provisional_returns.get(return_id)
|
|
if not isinstance(site, ReturnSite):
|
|
continue
|
|
caller_id = site.callable_id
|
|
if caller_id not in affected:
|
|
affected.add(caller_id)
|
|
pending.append(caller_id)
|
|
return affected
|
|
|
|
def _return_sites_for_binding_candidates(
|
|
self,
|
|
bindings: Mapping[str, BindingSite | None],
|
|
binding_ids: set[str],
|
|
returns: Mapping[str, ReturnSite | None],
|
|
local_return_ids_by_callable: Mapping[str, set[str]],
|
|
) -> dict[str, tuple[ReturnSite, ...]]:
|
|
"""Materialize only return-callable chains reachable from candidate bindings."""
|
|
|
|
pending = [
|
|
binding.returned_callable_id
|
|
for binding_id in binding_ids
|
|
if isinstance(binding := bindings.get(binding_id), BindingSite)
|
|
and binding.returned_callable_id is not None
|
|
]
|
|
result: dict[str, tuple[ReturnSite, ...]] = {}
|
|
while pending:
|
|
callable_id = pending.pop()
|
|
if callable_id in result:
|
|
continue
|
|
return_ids = (
|
|
self._graph.return_site_ids_by_callable.get(callable_id, set())
|
|
| local_return_ids_by_callable.get(callable_id, set())
|
|
)
|
|
sites = tuple(
|
|
sorted(
|
|
(
|
|
site
|
|
for return_id in return_ids
|
|
if isinstance(site := returns.get(return_id), ReturnSite)
|
|
),
|
|
key=lambda site: site.return_site_id,
|
|
)
|
|
)
|
|
result[callable_id] = sites
|
|
for site in sites:
|
|
if site.returned_callable_id is not None:
|
|
pending.append(site.returned_callable_id)
|
|
return result
|
|
|
|
def _binding_ids_reaching_actual_types(
|
|
self,
|
|
actual_type_ids: set[str],
|
|
returns: Mapping[str, ReturnSite | None],
|
|
local_binding_ids_by_actual_type: Mapping[str, set[str]],
|
|
local_binding_ids_by_returned_callable: Mapping[str, set[str]],
|
|
local_return_ids_by_actual_type: Mapping[str, set[str]],
|
|
local_return_ids_by_returned_callable: Mapping[str, set[str]],
|
|
) -> set[str]:
|
|
"""Follow only indexed direct and return-producing value paths."""
|
|
|
|
result = {
|
|
binding_id
|
|
for type_id in actual_type_ids
|
|
for binding_id in (
|
|
self._graph.binding_site_ids_by_actual_type.get(type_id, set())
|
|
| local_binding_ids_by_actual_type.get(type_id, set())
|
|
)
|
|
}
|
|
pending = [
|
|
site.callable_id
|
|
for type_id in actual_type_ids
|
|
for return_id in (
|
|
self._graph.return_site_ids_by_actual_type.get(type_id, set())
|
|
| local_return_ids_by_actual_type.get(type_id, set())
|
|
)
|
|
if isinstance(site := returns.get(return_id), ReturnSite)
|
|
]
|
|
seen: set[str] = set()
|
|
while pending:
|
|
callable_id = pending.pop()
|
|
if callable_id in seen:
|
|
continue
|
|
seen.add(callable_id)
|
|
result.update(
|
|
self._graph.binding_site_ids_by_returned_callable.get(callable_id, set())
|
|
)
|
|
result.update(local_binding_ids_by_returned_callable.get(callable_id, set()))
|
|
for return_id in (
|
|
self._graph.return_site_ids_by_returned_callable.get(callable_id, set())
|
|
| local_return_ids_by_returned_callable.get(callable_id, set())
|
|
):
|
|
site = returns.get(return_id)
|
|
if isinstance(site, ReturnSite):
|
|
pending.append(site.callable_id)
|
|
return result
|
|
|
|
def _resolve_reference(
|
|
self,
|
|
binding: BindingSite,
|
|
declarations: Mapping[str, DeclarationSite | None],
|
|
types: Mapping[str, DependencyType | None],
|
|
relationships: Mapping[str, ParentRelationship | None],
|
|
return_sites_by_callable: Mapping[str, tuple[ReturnSite, ...]],
|
|
type_kinds: Mapping[str, str | None] | None = None,
|
|
type_members: Mapping[str, tuple[frozenset[str], frozenset[str]] | None] | None = None,
|
|
) -> IndirectReference | None:
|
|
declaration = declarations.get(binding.declaration_site_id)
|
|
if not isinstance(declaration, DeclarationSite):
|
|
return None
|
|
provenance = self._binding_value_provenance(
|
|
binding,
|
|
return_sites_by_callable,
|
|
)
|
|
if provenance is None:
|
|
return None
|
|
actual_id, return_site_ids = provenance
|
|
if (
|
|
actual_id is None
|
|
or not isinstance(types.get(actual_id), DependencyType)
|
|
or actual_id == declaration.declared_type_id
|
|
):
|
|
return None
|
|
path = self._nominal_path(actual_id, declaration.declared_type_id, relationships)
|
|
if path is not None:
|
|
kind = "nominal"
|
|
structural = ()
|
|
else:
|
|
kinds = type_kinds if type_kinds is not None else self._type_kinds
|
|
members = type_members if type_members is not None else self._type_members
|
|
required_kind = kinds.get(declaration.declared_type_id) or "concrete"
|
|
required_methods, required_fields = members.get(
|
|
declaration.declared_type_id,
|
|
) or (frozenset(), frozenset())
|
|
actual_methods, actual_fields = members.get(actual_id) or (frozenset(), frozenset())
|
|
if required_kind != "protocol" or not required_methods <= actual_methods or not required_fields <= actual_fields:
|
|
return None
|
|
kind = "structural"
|
|
structural = tuple(sorted(required_methods | required_fields))
|
|
path = ()
|
|
return IndirectReference(
|
|
indirect_reference_id="|".join((declaration.declaration_site_id, binding.binding_site_id)),
|
|
declaration_site_id=declaration.declaration_site_id,
|
|
binding_site_id=binding.binding_site_id,
|
|
return_site_ids=return_site_ids,
|
|
declared_type_id=declaration.declared_type_id,
|
|
actual_type_id=actual_id,
|
|
compatibility_kind=kind,
|
|
parent_relationship_ids=tuple(path),
|
|
structural_member_names=structural,
|
|
)
|
|
|
|
def _all_reference_candidates(
|
|
self,
|
|
bindings: Mapping[str, BindingSite | None],
|
|
declarations: Mapping[str, DeclarationSite | None],
|
|
types: Mapping[str, DependencyType | None],
|
|
relationships: Mapping[str, ParentRelationship | None],
|
|
returns: Mapping[str, ReturnSite | None],
|
|
binding_ids: set[str],
|
|
return_sites_by_callable: Mapping[str, tuple[ReturnSite, ...]] | None = None,
|
|
type_kinds: Mapping[str, str | None] | None = None,
|
|
type_members: Mapping[str, tuple[frozenset[str], frozenset[str]] | None] | None = None,
|
|
) -> dict[str, IndirectReference]:
|
|
result: dict[str, IndirectReference] = {}
|
|
return_sites_by_callable = (
|
|
return_sites_by_callable
|
|
if return_sites_by_callable is not None
|
|
else self._return_sites_by_callable(returns)
|
|
)
|
|
for binding_id in binding_ids:
|
|
binding = bindings.get(binding_id)
|
|
if not isinstance(binding, BindingSite):
|
|
continue
|
|
reference = self._resolve_reference(
|
|
binding,
|
|
declarations,
|
|
types,
|
|
relationships,
|
|
return_sites_by_callable,
|
|
type_kinds,
|
|
type_members,
|
|
)
|
|
if reference is not None:
|
|
result[reference.indirect_reference_id] = reference
|
|
return result
|
|
|
|
@staticmethod
|
|
def _changed_owned_records(
|
|
old_records: Mapping[str, object],
|
|
old_owned_ids: set[str],
|
|
provisional_records: Mapping[str, object],
|
|
) -> tuple[set[str], dict[str, object], dict[str, object]]:
|
|
"""Compare only records owned by the parsed incremental population."""
|
|
|
|
removed = old_owned_ids - set(provisional_records)
|
|
added = {
|
|
record_id: record
|
|
for record_id, record in provisional_records.items()
|
|
if record_id not in old_records
|
|
}
|
|
changed = {
|
|
record_id: record
|
|
for record_id, record in provisional_records.items()
|
|
if record_id in old_records and old_records[record_id] != record
|
|
}
|
|
return removed, added, changed
|
|
|
|
@staticmethod
|
|
def _indexed_ids_for_sources(
|
|
index: Mapping[str, set[str]],
|
|
source_locations: set[str],
|
|
) -> set[str]:
|
|
return {
|
|
record_id
|
|
for location in source_locations
|
|
for record_id in index.get(location, set())
|
|
}
|
|
|
|
@staticmethod
|
|
def _local_index(
|
|
records: Mapping[str, object],
|
|
attribute: str,
|
|
) -> dict[str, set[str]]:
|
|
result: dict[str, set[str]] = {}
|
|
for record_id, record in records.items():
|
|
value = getattr(record, attribute)
|
|
if value is not None:
|
|
result.setdefault(value, set()).add(record_id)
|
|
return result
|
|
|
|
def _commit_full(
|
|
self,
|
|
source_by_location: dict[str, ClassMeta],
|
|
test_by_location: dict[str, ClassMeta],
|
|
types: dict[str, DependencyType],
|
|
relationships: dict[str, ParentRelationship],
|
|
declarations: dict[str, DeclarationSite],
|
|
bindings: dict[str, BindingSite],
|
|
returns: dict[str, ReturnSite],
|
|
references: dict[str, IndirectReference],
|
|
static_pairs: set[tuple[str, str]],
|
|
) -> None:
|
|
dependencies: dict[str, set[str]] = {}
|
|
dependents: dict[str, set[str]] = {}
|
|
for dependent, dependency in static_pairs:
|
|
dependencies.setdefault(dependent, set()).add(dependency)
|
|
dependents.setdefault(dependency, set()).add(dependent)
|
|
self._graph.source_by_location = dict(source_by_location)
|
|
self._graph.test_by_location = dict(test_by_location)
|
|
self._graph.source_dependencies = dependencies
|
|
self._graph.source_dependents = dependents
|
|
self._graph._replace_all_dependency_evidence(
|
|
type_definitions=types,
|
|
parent_relationships_by_id=relationships,
|
|
declaration_sites_by_id=declarations,
|
|
binding_sites_by_id=bindings,
|
|
return_sites_by_id=returns,
|
|
indirect_references_by_id=references,
|
|
)
|
|
|
|
def _test_matches_source(
|
|
self,
|
|
source: ClassMeta,
|
|
test: ClassMeta,
|
|
source_facts: _StaticFacts | None,
|
|
test_facts: _StaticFacts | None,
|
|
) -> bool:
|
|
if source_facts is not None and test_facts is not None:
|
|
if (
|
|
source_facts.module_name in test_facts.imported_modules
|
|
or any(
|
|
source_facts.module_name == item[0]
|
|
for item in test_facts.imported_names
|
|
)
|
|
):
|
|
return True
|
|
source_stem = Path(source.location).with_suffix("").name
|
|
test_stem = Path(test.location).with_suffix("").name
|
|
return test_stem in {source_stem, f"test_{source_stem}"}
|
|
|
|
def _map_tests(self, graph: CodeDependencyGraph, sources: list[ClassMeta], tests: list[ClassMeta], repo_root: Path, discovery: LanguageGlobStrategy) -> None:
|
|
for test in tests:
|
|
try:
|
|
values = discovery.searchable_beats(test.location, self.repo_exec.read_text(self._location_path(repo_root, test.location)))
|
|
except (OSError, SyntaxError, UnicodeDecodeError):
|
|
continue
|
|
self.last_update_instrumentation.files_parsed.add(test.location)
|
|
fact_beat = next((value for value in values if isinstance(value, SearchableBeat) and value.kind == "static_facts"), None)
|
|
test_attrs = self._attrs(fact_beat) if fact_beat is not None else {}
|
|
test_references = frozenset(str(item) for item in self._json_list(test_attrs.get("referenced_names")))
|
|
test_facts = _StaticFacts(
|
|
module_name=test_attrs.get("module_name", ""),
|
|
imported_modules=frozenset(str(item) for item in self._json_list(test_attrs.get("imported_modules"))),
|
|
imported_names=frozenset(
|
|
tuple(item)
|
|
for item in self._json_list(test_attrs.get("imported_names"))
|
|
if isinstance(item, list) and len(item) == 3
|
|
),
|
|
referenced_names=test_references,
|
|
)
|
|
self._test_facts_by_location[test.location] = test_facts
|
|
for source in sources:
|
|
source_facts = self._static_facts_by_location.get(source.location)
|
|
symbol_match = source_facts is not None and bool(test_references & source_facts.defined_public_symbols)
|
|
if symbol_match or self._test_matches_source(source, test, source_facts, test_facts):
|
|
graph._add_source_test_mapping(source.location, test.location)
|
|
|
|
def _test_mapping_delta_for_changed_sources(
|
|
self,
|
|
graph: CodeDependencyGraph,
|
|
source_map: Mapping[str, ClassMeta],
|
|
tests: list[ClassMeta],
|
|
facts: Mapping[str, _StaticFacts],
|
|
changed_sources: set[str],
|
|
) -> tuple[set[tuple[str, str]], set[tuple[str, str]]]:
|
|
"""Return only source/test pairs touched by changed source evidence."""
|
|
|
|
old_pairs = {
|
|
(source_location, test_location)
|
|
for source_location in changed_sources
|
|
for test_location in graph.tests_for_source.get(source_location, set())
|
|
}
|
|
new_pairs: set[tuple[str, str]] = set()
|
|
for source_location in changed_sources:
|
|
source = source_map.get(source_location)
|
|
if source is None:
|
|
continue
|
|
source_facts = facts.get(source_location)
|
|
for test in tests:
|
|
test_facts = self._test_facts_by_location.get(test.location)
|
|
symbol_match = (
|
|
test_facts is not None
|
|
and source_facts is not None
|
|
and bool(
|
|
test_facts.referenced_names
|
|
& source_facts.defined_public_symbols
|
|
)
|
|
)
|
|
if symbol_match or self._test_matches_source(source, test, source_facts, test_facts):
|
|
new_pairs.add((source_location, test.location))
|
|
return new_pairs - old_pairs, old_pairs - new_pairs
|
|
|
|
def build(self, source_files: list[ClassMeta], test_files: list[ClassMeta], repo_root: Path, src_root: Path, test_root: Path) -> CodeDependencyGraph:
|
|
self._assert_mutable()
|
|
self.last_update_instrumentation = DependencyUpdateInstrumentation()
|
|
discovery = self._configure_discovery(src_root, test_root, repo_root)
|
|
sources = self._dedupe(source_files)
|
|
tests = self._dedupe(test_files)
|
|
source_map = {item.location: item for item in sources}
|
|
self._beats_by_source = self._discover(set(source_map), source_map, repo_root, discovery)
|
|
types, relationships, declarations, bindings, returns, facts = self._records_from_beats(self._beats_by_source, source_map)
|
|
static_pairs = self._static_pairs(facts)
|
|
references = self._all_reference_candidates(bindings, declarations, types, relationships, returns, set(bindings))
|
|
self._commit_full(source_map, {item.location: item for item in tests}, types, relationships, declarations, bindings, returns, references, static_pairs)
|
|
self._static_facts_by_location = facts
|
|
self._test_facts_by_location = {}
|
|
self._graph.tests_for_source = {}
|
|
self._graph.sources_for_test = {}
|
|
for identity in declarations:
|
|
self.last_update_instrumentation.declaration_sites_added.add(identity)
|
|
for identity in bindings:
|
|
self.last_update_instrumentation.binding_sites_added.add(identity)
|
|
for identity in returns:
|
|
self.last_update_instrumentation.return_sites_added.add(identity)
|
|
for identity in references:
|
|
self.last_update_instrumentation.indirect_references_added.add(identity)
|
|
self._map_tests(self._graph, sources, tests, repo_root, discovery)
|
|
self.signature = (tuple(sorted(source_map)), tuple(sorted(item.location for item in tests)))
|
|
return self._graph
|
|
|
|
def _hierarchy_closure(
|
|
self,
|
|
relationships: Mapping[str, ParentRelationship | None],
|
|
seeds: set[str],
|
|
direction: str,
|
|
) -> set[str]:
|
|
adjacency: dict[str, set[str]] = {}
|
|
for relationship in relationships.values():
|
|
if not isinstance(relationship, ParentRelationship):
|
|
continue
|
|
if direction == "parents":
|
|
adjacency.setdefault(relationship.child_type_id, set()).add(relationship.parent_type_id)
|
|
else:
|
|
adjacency.setdefault(relationship.parent_type_id, set()).add(relationship.child_type_id)
|
|
result: set[str] = set()
|
|
pending = list(seeds)
|
|
while pending:
|
|
current = pending.pop()
|
|
for next_id in adjacency.get(current, ()):
|
|
if next_id not in result:
|
|
result.add(next_id)
|
|
pending.append(next_id)
|
|
return result
|
|
|
|
def _incremental(
|
|
self,
|
|
source_map: dict[str, ClassMeta],
|
|
changed: set[str],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
tests: list[ClassMeta],
|
|
) -> None:
|
|
"""Plan and apply one source-local delta against the live graph."""
|
|
|
|
graph = self._graph
|
|
discovery = self._configure_discovery(src_root, test_root, repo_root)
|
|
changed_existing = changed & set(source_map)
|
|
removed_locations = changed - set(source_map)
|
|
discovered = self._discover(changed_existing, source_map, repo_root, discovery)
|
|
|
|
def provisional_records(parsed_locations: set[str]) -> dict[str, object]:
|
|
old_type_ids_by_source = {
|
|
location: set(graph.type_ids_by_source.get(location, set()))
|
|
for location in parsed_locations
|
|
}
|
|
old_type_ids = set().union(*old_type_ids_by_source.values()) if old_type_ids_by_source else set()
|
|
old_relationship_ids = {
|
|
relationship_id
|
|
for type_id in old_type_ids
|
|
for relationship_id in graph.parent_relationship_ids_by_child.get(type_id, set())
|
|
}
|
|
old_declaration_ids = self._indexed_ids_for_sources(
|
|
graph.declaration_site_ids_by_source,
|
|
parsed_locations,
|
|
)
|
|
old_binding_ids = self._indexed_ids_for_sources(
|
|
graph.binding_site_ids_by_source,
|
|
parsed_locations,
|
|
)
|
|
old_return_ids = self._indexed_ids_for_sources(
|
|
graph.return_site_ids_by_source,
|
|
parsed_locations,
|
|
)
|
|
(
|
|
local_type_kinds,
|
|
local_type_members,
|
|
local_callable_parameters,
|
|
local_callable_sources,
|
|
local_callable_ids_by_source,
|
|
local_aliases_by_source,
|
|
) = self._private_indexes_from_beats(discovered)
|
|
old_callable_ids_by_source = {
|
|
location: set(self._callable_ids_by_source.get(location, set()))
|
|
for location in parsed_locations
|
|
}
|
|
for location in parsed_locations:
|
|
if old_callable_ids_by_source[location] or location not in self._beats_by_source:
|
|
continue
|
|
old_callable_ids_by_source[location].update(
|
|
self._private_indexes_from_beats(
|
|
{location: self._beats_by_source[location]}
|
|
)[4].get(location, set())
|
|
)
|
|
old_callable_ids = set().union(*old_callable_ids_by_source.values()) if old_callable_ids_by_source else set()
|
|
type_overrides: dict[str, DependencyType | None] = {type_id: None for type_id in old_type_ids}
|
|
kind_overrides: dict[str, str | None] = {type_id: None for type_id in old_type_ids}
|
|
member_overrides: dict[str, tuple[frozenset[str], frozenset[str]] | None] = {
|
|
type_id: None for type_id in old_type_ids
|
|
}
|
|
callable_parameter_overrides: dict[str, tuple[tuple[str, str, int | None, str], ...] | None] = {
|
|
callable_id: None for callable_id in old_callable_ids
|
|
}
|
|
callable_source_overrides: dict[str, str | None] = {
|
|
callable_id: None for callable_id in old_callable_ids
|
|
}
|
|
kind_overrides.update(local_type_kinds)
|
|
member_overrides.update(local_type_members)
|
|
callable_parameter_overrides.update(local_callable_parameters)
|
|
callable_source_overrides.update(local_callable_sources)
|
|
aliases_overrides = {
|
|
location: local_aliases_by_source.get(location, {})
|
|
for location in parsed_locations
|
|
}
|
|
types_base: Mapping[str, DependencyType | None] = ChainMap(
|
|
type_overrides,
|
|
graph.type_definitions,
|
|
)
|
|
relationships_base: Mapping[str, ParentRelationship | None] = ChainMap(
|
|
{relationship_id: None for relationship_id in old_relationship_ids},
|
|
graph.parent_relationships_by_id,
|
|
)
|
|
declarations_base: Mapping[str, DeclarationSite | None] = ChainMap(
|
|
{declaration_id: None for declaration_id in old_declaration_ids},
|
|
graph.declaration_sites_by_id,
|
|
)
|
|
type_kinds: Mapping[str, str | None] = ChainMap(kind_overrides, self._type_kinds)
|
|
type_members: Mapping[str, tuple[frozenset[str], frozenset[str]] | None] = ChainMap(
|
|
member_overrides,
|
|
self._type_members,
|
|
)
|
|
callable_parameters: Mapping[str, tuple[tuple[str, str, int | None, str], ...] | None] = ChainMap(
|
|
callable_parameter_overrides,
|
|
self._callable_parameters,
|
|
)
|
|
callable_sources: Mapping[str, str | None] = ChainMap(
|
|
callable_source_overrides,
|
|
self._callable_sources,
|
|
)
|
|
aliases_by_source: Mapping[str, Mapping[str, str | None]] = ChainMap(
|
|
aliases_overrides,
|
|
self._aliases_by_source,
|
|
)
|
|
(
|
|
local_types,
|
|
local_relationships,
|
|
local_declarations,
|
|
local_bindings,
|
|
local_returns,
|
|
local_facts,
|
|
) = self._records_from_beats(
|
|
discovered,
|
|
source_map,
|
|
base_types=types_base,
|
|
base_relationships=relationships_base,
|
|
base_declarations=declarations_base,
|
|
type_kinds=type_kinds,
|
|
aliases_by_source=aliases_by_source,
|
|
callable_parameters=callable_parameters,
|
|
callable_sources=callable_sources,
|
|
refresh_private_indexes=False,
|
|
)
|
|
return {
|
|
"old_type_ids_by_source": old_type_ids_by_source,
|
|
"old_relationship_ids": old_relationship_ids,
|
|
"old_declaration_ids": old_declaration_ids,
|
|
"old_binding_ids": old_binding_ids,
|
|
"old_return_ids": old_return_ids,
|
|
"old_callable_ids_by_source": old_callable_ids_by_source,
|
|
"local_indexes": (
|
|
local_type_kinds,
|
|
local_type_members,
|
|
local_callable_parameters,
|
|
local_callable_sources,
|
|
local_callable_ids_by_source,
|
|
local_aliases_by_source,
|
|
),
|
|
"types": ChainMap(local_types, types_base),
|
|
"relationships": ChainMap(local_relationships, relationships_base),
|
|
"declarations": ChainMap(local_declarations, declarations_base),
|
|
"bindings": ChainMap(
|
|
local_bindings,
|
|
ChainMap(
|
|
{binding_id: None for binding_id in old_binding_ids},
|
|
graph.binding_sites_by_id,
|
|
),
|
|
),
|
|
"returns": ChainMap(
|
|
local_returns,
|
|
ChainMap(
|
|
{return_id: None for return_id in old_return_ids},
|
|
graph.return_sites_by_id,
|
|
),
|
|
),
|
|
"local_types": local_types,
|
|
"local_relationships": local_relationships,
|
|
"local_declarations": local_declarations,
|
|
"local_bindings": local_bindings,
|
|
"local_returns": local_returns,
|
|
"local_facts": local_facts,
|
|
"type_kinds": type_kinds,
|
|
"type_members": type_members,
|
|
}
|
|
|
|
parsed_locations = set(discovered) | removed_locations
|
|
state = provisional_records(parsed_locations)
|
|
|
|
# A newly added source can make previously parsed imports resolvable.
|
|
# Its static edges do not exist in the live graph yet, so derive only
|
|
# the direct dependents from stored static facts before converting
|
|
# their local beats. This is the bounded repair needed when a split
|
|
# source is added after the sources which import one of its types.
|
|
new_source_types = {
|
|
type_id: record
|
|
for type_id, record in state["local_types"].items()
|
|
if (
|
|
type_id not in graph.type_definitions
|
|
and record.source.location not in graph.source_by_location
|
|
)
|
|
}
|
|
new_type_sources = {record.source.location for record in new_source_types.values()}
|
|
new_type_names = {
|
|
name
|
|
for record in new_source_types.values()
|
|
for name in (record.qualified_name, record.local_name)
|
|
}
|
|
|
|
def has_unresolved_type_use(location: str) -> bool:
|
|
aliases = dict(self._aliases_by_source.get(location, {}))
|
|
for beat in self._beats_by_source.get(location, ()):
|
|
attrs = self._attrs(beat)
|
|
expression = (
|
|
attrs.get("parent_name", "")
|
|
if beat.kind == "parent_relationship"
|
|
else attrs.get("declared_name", "")
|
|
if beat.kind == "declaration_site"
|
|
else attrs.get("declared_return_name", "")
|
|
if beat.kind == "return_site"
|
|
else ""
|
|
)
|
|
expanded = self._expand_alias(expression, aliases) or ""
|
|
if (
|
|
expanded in new_type_names
|
|
or expression.rsplit(".", 1)[-1] in new_type_names
|
|
):
|
|
return True
|
|
return False
|
|
|
|
new_declaration_sources = {
|
|
record.source.location
|
|
for declaration_id, record in state["local_declarations"].items()
|
|
if (
|
|
declaration_id not in graph.declaration_sites_by_id
|
|
and record.source.location not in graph.source_by_location
|
|
)
|
|
}
|
|
new_type_frontier = {
|
|
candidate_location
|
|
for source_location in new_type_sources
|
|
if (
|
|
dependency_facts := state["local_facts"].get(source_location)
|
|
) is not None
|
|
for candidate_location, candidate_facts in self._static_facts_by_location.items()
|
|
if (
|
|
candidate_location in source_map
|
|
and candidate_location != source_location
|
|
and self._static_pair_exists(candidate_facts, dependency_facts)
|
|
and has_unresolved_type_use(candidate_location)
|
|
)
|
|
}
|
|
new_declaration_frontier = {
|
|
candidate_location
|
|
for source_location in new_declaration_sources
|
|
if (
|
|
dependency_facts := state["local_facts"].get(source_location)
|
|
) is not None
|
|
for candidate_location, candidate_facts in self._static_facts_by_location.items()
|
|
if (
|
|
candidate_location in source_map
|
|
and candidate_location != source_location
|
|
and self._static_pair_exists(candidate_facts, dependency_facts)
|
|
)
|
|
}
|
|
|
|
# A first child or newly eligible abstraction discovers declarations
|
|
# only in static dependents of that type's defining source.
|
|
new_parent_sources: set[str] = set()
|
|
for relationship_id, relationship in state["local_relationships"].items():
|
|
if relationship_id in graph.parent_relationships_by_id:
|
|
continue
|
|
remaining_parent_relationships = (
|
|
graph.parent_relationship_ids_by_parent.get(relationship.parent_type_id, set())
|
|
- state["old_relationship_ids"]
|
|
)
|
|
if remaining_parent_relationships:
|
|
continue
|
|
parent = state["types"].get(relationship.parent_type_id)
|
|
if isinstance(parent, DependencyType):
|
|
new_parent_sources.add(parent.source.location)
|
|
local_type_kinds = state["local_indexes"][0]
|
|
for type_id, new_kind in local_type_kinds.items():
|
|
if (
|
|
self._type_kinds.get(type_id, "concrete") not in {"protocol", "abc"}
|
|
and new_kind in {"protocol", "abc"}
|
|
):
|
|
type_record = state["types"].get(type_id)
|
|
if isinstance(type_record, DependencyType):
|
|
new_parent_sources.add(type_record.source.location)
|
|
frontier = new_type_frontier | new_declaration_frontier | {
|
|
dependent
|
|
for parent_source in new_parent_sources
|
|
for dependent in graph.source_dependents.get(parent_source, set())
|
|
if dependent in source_map
|
|
}
|
|
frontier_to_discover = frontier - set(discovered)
|
|
if frontier_to_discover:
|
|
discovered.update(
|
|
self._discover(frontier_to_discover, source_map, repo_root, discovery)
|
|
)
|
|
parsed_locations = set(discovered) | removed_locations
|
|
state = provisional_records(parsed_locations)
|
|
|
|
# A newly tracked declaration makes call and assignment evidence in
|
|
# its static dependents relevant. This is a single bounded expansion,
|
|
# not a repository sweep.
|
|
declaration_frontier = {
|
|
dependent
|
|
for declaration_id, declaration in state["local_declarations"].items()
|
|
if (
|
|
declaration_id not in graph.declaration_sites_by_id
|
|
or graph.declaration_sites_by_id[declaration_id] != declaration
|
|
)
|
|
for dependent in graph.source_dependents.get(declaration.source.location, set())
|
|
if dependent in source_map
|
|
} - set(discovered)
|
|
if declaration_frontier:
|
|
discovered.update(
|
|
self._discover(declaration_frontier, source_map, repo_root, discovery)
|
|
)
|
|
parsed_locations = set(discovered) | removed_locations
|
|
state = provisional_records(parsed_locations)
|
|
|
|
local_types = state["local_types"]
|
|
local_relationships = state["local_relationships"]
|
|
local_declarations = state["local_declarations"]
|
|
local_bindings = state["local_bindings"]
|
|
local_returns = state["local_returns"]
|
|
provisional_types = state["types"]
|
|
provisional_relationships = state["relationships"]
|
|
provisional_declarations = state["declarations"]
|
|
provisional_bindings = state["bindings"]
|
|
provisional_returns = state["returns"]
|
|
types_removed, types_added, types_changed = self._changed_owned_records(
|
|
graph.type_definitions,
|
|
{
|
|
type_id
|
|
for ids in state["old_type_ids_by_source"].values()
|
|
for type_id in ids
|
|
},
|
|
local_types,
|
|
)
|
|
relationships_removed, relationships_added, relationships_changed = self._changed_owned_records(
|
|
graph.parent_relationships_by_id,
|
|
state["old_relationship_ids"],
|
|
local_relationships,
|
|
)
|
|
declarations_removed, declarations_added, declarations_changed = self._changed_owned_records(
|
|
graph.declaration_sites_by_id,
|
|
state["old_declaration_ids"],
|
|
local_declarations,
|
|
)
|
|
bindings_removed, bindings_added, bindings_changed = self._changed_owned_records(
|
|
graph.binding_sites_by_id,
|
|
state["old_binding_ids"],
|
|
local_bindings,
|
|
)
|
|
returns_removed, returns_added, returns_changed = self._changed_owned_records(
|
|
graph.return_sites_by_id,
|
|
state["old_return_ids"],
|
|
local_returns,
|
|
)
|
|
changed_type_ids = set(types_removed) | set(types_added) | set(types_changed)
|
|
changed_relationship_ids = (
|
|
set(relationships_removed) | set(relationships_added) | set(relationships_changed)
|
|
)
|
|
changed_declaration_ids = (
|
|
set(declarations_removed) | set(declarations_added) | set(declarations_changed)
|
|
)
|
|
changed_binding_ids = set(bindings_removed) | set(bindings_added) | set(bindings_changed)
|
|
changed_return_ids = set(returns_removed) | set(returns_added) | set(returns_changed)
|
|
|
|
static_pairs_added, static_pairs_removed = self._static_pair_delta_for_changed(
|
|
graph.source_dependencies,
|
|
graph.source_dependents,
|
|
state["local_facts"],
|
|
parsed_locations,
|
|
)
|
|
affected_refs: set[str] = set()
|
|
for relationship_id in changed_relationship_ids:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_parent_relationship.get(relationship_id, set())
|
|
)
|
|
for declaration_id in changed_declaration_ids:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_declaration_site.get(declaration_id, set())
|
|
)
|
|
for binding_id in changed_binding_ids:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_binding_site.get(binding_id, set())
|
|
)
|
|
for return_id in changed_return_ids:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_return_site.get(return_id, set())
|
|
)
|
|
for type_id in changed_type_ids:
|
|
affected_refs.update(graph.indirect_reference_ids_by_actual_type.get(type_id, set()))
|
|
for declaration_id in graph.declaration_site_ids_by_declared_type.get(type_id, set()):
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_declaration_site.get(declaration_id, set())
|
|
)
|
|
|
|
hierarchy_seeds = set(changed_type_ids)
|
|
for relationship_id in changed_relationship_ids:
|
|
for relationship in (
|
|
graph.parent_relationships_by_id.get(relationship_id),
|
|
provisional_relationships.get(relationship_id),
|
|
):
|
|
if isinstance(relationship, ParentRelationship):
|
|
hierarchy_seeds.add(relationship.parent_type_id)
|
|
hierarchy_seeds.add(relationship.child_type_id)
|
|
actual_candidates = (
|
|
hierarchy_seeds
|
|
| self._hierarchy_closure(graph.parent_relationships_by_id, hierarchy_seeds, "children")
|
|
| self._hierarchy_closure(provisional_relationships, hierarchy_seeds, "children")
|
|
)
|
|
declared_candidates = (
|
|
hierarchy_seeds
|
|
| self._hierarchy_closure(graph.parent_relationships_by_id, hierarchy_seeds, "parents")
|
|
| self._hierarchy_closure(provisional_relationships, hierarchy_seeds, "parents")
|
|
)
|
|
local_bindings_by_declaration = self._local_index(
|
|
local_bindings,
|
|
"declaration_site_id",
|
|
)
|
|
local_bindings_by_actual_type = self._local_index(
|
|
local_bindings,
|
|
"direct_actual_type_id",
|
|
)
|
|
local_bindings_by_returned_callable = self._local_index(
|
|
local_bindings,
|
|
"returned_callable_id",
|
|
)
|
|
local_declarations_by_declared_type = self._local_index(
|
|
local_declarations,
|
|
"declared_type_id",
|
|
)
|
|
local_returns_by_callable = self._local_index(local_returns, "callable_id")
|
|
local_returns_by_actual_type = self._local_index(
|
|
local_returns,
|
|
"direct_actual_type_id",
|
|
)
|
|
local_returns_by_returned_callable = self._local_index(
|
|
local_returns,
|
|
"returned_callable_id",
|
|
)
|
|
candidate_binding_ids = set(changed_binding_ids)
|
|
for declaration_id in changed_declaration_ids:
|
|
candidate_binding_ids.update(
|
|
graph.binding_site_ids_by_declaration_site.get(declaration_id, set())
|
|
)
|
|
candidate_binding_ids.update(
|
|
local_bindings_by_declaration.get(declaration_id, set())
|
|
)
|
|
candidate_binding_ids.update(
|
|
self._binding_ids_reaching_actual_types(
|
|
actual_candidates,
|
|
provisional_returns,
|
|
local_bindings_by_actual_type,
|
|
local_bindings_by_returned_callable,
|
|
local_returns_by_actual_type,
|
|
local_returns_by_returned_callable,
|
|
)
|
|
)
|
|
for declared_type_id in declared_candidates:
|
|
declaration_ids = (
|
|
graph.declaration_site_ids_by_declared_type.get(declared_type_id, set())
|
|
| local_declarations_by_declared_type.get(declared_type_id, set())
|
|
)
|
|
for declaration_id in declaration_ids:
|
|
candidate_binding_ids.update(
|
|
graph.binding_site_ids_by_declaration_site.get(declaration_id, set())
|
|
)
|
|
candidate_binding_ids.update(
|
|
local_bindings_by_declaration.get(declaration_id, set())
|
|
)
|
|
changed_callables = self._callables_affected_by_return_changes(
|
|
changed_return_ids,
|
|
provisional_returns,
|
|
local_returns_by_returned_callable,
|
|
)
|
|
for callable_id in changed_callables:
|
|
candidate_binding_ids.update(
|
|
graph.binding_site_ids_by_returned_callable.get(callable_id, set())
|
|
)
|
|
candidate_binding_ids.update(
|
|
local_bindings_by_returned_callable.get(callable_id, set())
|
|
)
|
|
for binding_id in candidate_binding_ids:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_binding_site.get(binding_id, set())
|
|
)
|
|
candidate_binding_ids.update(
|
|
graph.indirect_references_by_id[reference_id].binding_site_id
|
|
for reference_id in affected_refs
|
|
if reference_id in graph.indirect_references_by_id
|
|
)
|
|
new_candidates = candidate_binding_ids | set(local_bindings)
|
|
for binding_id in new_candidates:
|
|
affected_refs.update(
|
|
graph.indirect_reference_ids_by_binding_site.get(binding_id, set())
|
|
)
|
|
return_sites_by_callable = self._return_sites_for_binding_candidates(
|
|
provisional_bindings,
|
|
new_candidates,
|
|
provisional_returns,
|
|
local_returns_by_callable,
|
|
)
|
|
rebuilt = self._all_reference_candidates(
|
|
provisional_bindings,
|
|
provisional_declarations,
|
|
provisional_types,
|
|
provisional_relationships,
|
|
provisional_returns,
|
|
new_candidates,
|
|
return_sites_by_callable,
|
|
state["type_kinds"],
|
|
state["type_members"],
|
|
)
|
|
old_references_to_revalidate = {
|
|
reference_id: graph.indirect_references_by_id[reference_id]
|
|
for reference_id in affected_refs
|
|
if reference_id in graph.indirect_references_by_id
|
|
}
|
|
references_added = {
|
|
reference_id: reference
|
|
for reference_id, reference in rebuilt.items()
|
|
if reference_id not in graph.indirect_references_by_id
|
|
}
|
|
references_removed = {
|
|
reference_id
|
|
for reference_id in old_references_to_revalidate
|
|
if reference_id not in rebuilt
|
|
}
|
|
references_changed = {
|
|
reference_id: reference
|
|
for reference_id, reference in rebuilt.items()
|
|
if reference_id in graph.indirect_references_by_id
|
|
and graph.indirect_references_by_id[reference_id] != reference
|
|
}
|
|
tests_for_source_added, tests_for_source_removed = (
|
|
self._test_mapping_delta_for_changed_sources(
|
|
graph,
|
|
source_map,
|
|
tests,
|
|
state["local_facts"],
|
|
parsed_locations,
|
|
)
|
|
)
|
|
source_added = {
|
|
location: source_map[location]
|
|
for location in changed_existing
|
|
if location not in graph.source_by_location
|
|
}
|
|
source_changed = {
|
|
location: source_map[location]
|
|
for location in changed_existing
|
|
if location in graph.source_by_location
|
|
and graph.source_by_location[location] != source_map[location]
|
|
}
|
|
delta = DependencyDelta(
|
|
source_added=source_added,
|
|
source_removed=frozenset(removed_locations & set(graph.source_by_location)),
|
|
source_changed=source_changed,
|
|
static_pairs_added=frozenset(static_pairs_added),
|
|
static_pairs_removed=frozenset(static_pairs_removed),
|
|
types_added=types_added,
|
|
types_removed=frozenset(types_removed),
|
|
types_changed=types_changed,
|
|
relationships_added=relationships_added,
|
|
relationships_removed=frozenset(relationships_removed),
|
|
relationships_changed=relationships_changed,
|
|
declarations_added=declarations_added,
|
|
declarations_removed=frozenset(declarations_removed),
|
|
declarations_changed=declarations_changed,
|
|
bindings_added=bindings_added,
|
|
bindings_removed=frozenset(bindings_removed),
|
|
bindings_changed=bindings_changed,
|
|
returns_added=returns_added,
|
|
returns_removed=frozenset(returns_removed),
|
|
returns_changed=returns_changed,
|
|
references_added=references_added,
|
|
references_removed=frozenset(references_removed),
|
|
references_changed=references_changed,
|
|
tests_for_source_added=frozenset(tests_for_source_added),
|
|
tests_for_source_removed=frozenset(tests_for_source_removed),
|
|
)
|
|
graph._apply_dependency_delta(delta)
|
|
|
|
(
|
|
local_type_kinds,
|
|
local_type_members,
|
|
local_callable_parameters,
|
|
local_callable_sources,
|
|
local_callable_ids_by_source,
|
|
local_aliases_by_source,
|
|
) = state["local_indexes"]
|
|
for location in parsed_locations:
|
|
for type_id in state["old_type_ids_by_source"].get(location, set()):
|
|
self._type_kinds.pop(type_id, None)
|
|
self._type_members.pop(type_id, None)
|
|
for callable_id in state["old_callable_ids_by_source"].get(location, set()):
|
|
self._callable_parameters.pop(callable_id, None)
|
|
self._callable_sources.pop(callable_id, None)
|
|
self._callable_ids_by_source.pop(location, None)
|
|
self._aliases_by_source.pop(location, None)
|
|
self._static_facts_by_location.pop(location, None)
|
|
self._type_kinds.update(local_type_kinds)
|
|
self._type_members.update(local_type_members)
|
|
self._callable_parameters.update(local_callable_parameters)
|
|
self._callable_sources.update(local_callable_sources)
|
|
self._callable_ids_by_source.update(local_callable_ids_by_source)
|
|
self._aliases_by_source.update(local_aliases_by_source)
|
|
self._static_facts_by_location.update(state["local_facts"])
|
|
for location in removed_locations:
|
|
self._beats_by_source.pop(location, None)
|
|
self._beats_by_source.update(discovered)
|
|
|
|
self.last_update_instrumentation.declaration_sites_added.update(delta.declarations_added)
|
|
self.last_update_instrumentation.declaration_sites_removed.update(delta.declarations_removed)
|
|
self.last_update_instrumentation.binding_sites_added.update(delta.bindings_added)
|
|
self.last_update_instrumentation.binding_sites_removed.update(delta.bindings_removed)
|
|
self.last_update_instrumentation.return_sites_added.update(delta.returns_added)
|
|
self.last_update_instrumentation.return_sites_removed.update(delta.returns_removed)
|
|
self.last_update_instrumentation.indirect_references_added.update(delta.references_added)
|
|
self.last_update_instrumentation.indirect_references_removed.update(delta.references_removed)
|
|
for reference_id, old_reference in old_references_to_revalidate.items():
|
|
if graph.indirect_references_by_id.get(reference_id) != old_reference:
|
|
self.last_update_instrumentation.indirect_references_revalidated.add(reference_id)
|
|
|
|
def _upsert_source(
|
|
self,
|
|
source_file: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
creating = source_file.location not in self._graph.source_by_location
|
|
sources = self._dedupe([source_file, *source_files])
|
|
source_map = {item.location: item for item in sources}
|
|
if not self._graph.source_by_location:
|
|
if creating:
|
|
self._ensure_package_chain_for_new_file(source_file.location, repo_root, src_root)
|
|
return self.build(sources, self._dedupe(test_files), repo_root, src_root, test_root)
|
|
if creating and source_file.location not in self._graph.source_by_location:
|
|
self._ensure_package_chain_for_new_file(source_file.location, repo_root, src_root)
|
|
changed = {source_file.location} | (set(self._graph.source_by_location) - set(source_map))
|
|
self._incremental(source_map, changed, repo_root, src_root, test_root, self._dedupe(test_files))
|
|
self.signature = (tuple(sorted(source_map)), tuple(sorted(item.location for item in test_files)))
|
|
return self._graph
|
|
|
|
def _upsert_test(
|
|
self,
|
|
test_file: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
creating = test_file.location not in self._graph.test_by_location
|
|
sources = self._dedupe(source_files)
|
|
tests = self._dedupe([test_file, *test_files])
|
|
if not self._graph.source_by_location and sources:
|
|
return self.build(sources, tests, repo_root, src_root, test_root)
|
|
if creating and test_file.location not in self._graph.test_by_location:
|
|
self._ensure_package_chain_for_new_file(test_file.location, repo_root, test_root)
|
|
old_test = self._graph.test_by_location.get(test_file.location)
|
|
old_pairs = frozenset(
|
|
(source_location, test_file.location)
|
|
for source_location in self._graph.sources_for_test.get(test_file.location, set())
|
|
)
|
|
delta = DependencyDelta(
|
|
test_added={test_file.location: test_file} if old_test is None else {},
|
|
test_changed={test_file.location: test_file} if old_test is not None else {},
|
|
tests_for_source_removed=old_pairs,
|
|
)
|
|
self._graph._apply_dependency_delta(delta)
|
|
discovery = self._configure_discovery(src_root, test_root, repo_root)
|
|
self._map_tests(self._graph, sources, [test_file], repo_root, discovery)
|
|
self.signature = (
|
|
tuple(sorted(source.location for source in sources)),
|
|
tuple(sorted(test.location for test in tests)),
|
|
)
|
|
return self._graph
|
|
|
|
def upsert(
|
|
self,
|
|
class_meta: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
"""Create or update one ClassMeta slice and incrementally repair the graph."""
|
|
|
|
self._assert_mutable()
|
|
self.last_update_instrumentation = DependencyUpdateInstrumentation()
|
|
kind = self._class_meta_kind(
|
|
class_meta,
|
|
source_files,
|
|
test_files,
|
|
repo_root,
|
|
src_root,
|
|
test_root,
|
|
)
|
|
if kind == "source":
|
|
return self._upsert_source(
|
|
class_meta,
|
|
source_files,
|
|
test_files,
|
|
repo_root,
|
|
src_root,
|
|
test_root,
|
|
)
|
|
return self._upsert_test(
|
|
class_meta,
|
|
source_files,
|
|
test_files,
|
|
repo_root,
|
|
src_root,
|
|
test_root,
|
|
)
|
|
|
|
def delete(
|
|
self,
|
|
class_meta: ClassMeta,
|
|
source_files: list[ClassMeta],
|
|
test_files: list[ClassMeta],
|
|
repo_root: Path,
|
|
src_root: Path,
|
|
test_root: Path,
|
|
) -> CodeDependencyGraph:
|
|
"""Delete one ClassMeta slice and incrementally repair the graph."""
|
|
|
|
self._assert_mutable()
|
|
self.last_update_instrumentation = DependencyUpdateInstrumentation()
|
|
kind = self._class_meta_kind(
|
|
class_meta,
|
|
source_files,
|
|
test_files,
|
|
repo_root,
|
|
src_root,
|
|
test_root,
|
|
)
|
|
if kind == "source":
|
|
source_map = {
|
|
item.location: item
|
|
for item in self._dedupe(source_files)
|
|
if item.location != class_meta.location
|
|
}
|
|
self._incremental(
|
|
source_map,
|
|
{class_meta.location},
|
|
repo_root,
|
|
src_root,
|
|
test_root,
|
|
self._dedupe(test_files),
|
|
)
|
|
self.signature = (
|
|
tuple(sorted(source_map)),
|
|
tuple(sorted(test.location for test in test_files)),
|
|
)
|
|
return self._graph
|
|
|
|
old_pairs = frozenset(
|
|
(source_location, class_meta.location)
|
|
for source_location in self._graph.sources_for_test.get(class_meta.location, set())
|
|
)
|
|
self._graph._apply_dependency_delta(
|
|
DependencyDelta(
|
|
test_removed=frozenset({class_meta.location}),
|
|
tests_for_source_removed=old_pairs,
|
|
)
|
|
)
|
|
self._test_facts_by_location.pop(class_meta.location, None)
|
|
self.signature = (
|
|
tuple(sorted(source.location for source in source_files)),
|
|
tuple(sorted(test.location for test in test_files if test.location != class_meta.location)),
|
|
)
|
|
return self._graph
|
|
|
|
# Shallow domain walks and source projections belong to the maintainer.
|
|
def trace_types_defined_in(self, source_location: str) -> list[DependencyType]:
|
|
return [self._graph.type_definitions[item] for item in sorted(self._graph.type_ids_by_source.get(source_location, set()))]
|
|
|
|
def direct_tests_for(self, source_files: list[ClassMeta]) -> list[ClassMeta]:
|
|
locations = {
|
|
test_location
|
|
for source in source_files
|
|
for test_location in self._graph.tests_for_source.get(source.location, set())
|
|
}
|
|
return [self._graph.test_by_location[location] for location in sorted(locations) if location in self._graph.test_by_location]
|
|
|
|
def direct_parent_type_ids_of(self, child_type_id: str) -> set[str]:
|
|
return {self._graph.parent_relationships_by_id[item].parent_type_id for item in self._graph.parent_relationship_ids_by_child.get(child_type_id, set())}
|
|
|
|
def direct_child_type_ids_of(self, parent_type_id: str) -> set[str]:
|
|
return {self._graph.parent_relationships_by_id[item].child_type_id for item in self._graph.parent_relationship_ids_by_parent.get(parent_type_id, set())}
|
|
|
|
def trace_declaration_sites_in(self, source_location: str) -> list[DeclarationSite]:
|
|
return [self._graph.declaration_sites_by_id[item] for item in sorted(self._graph.declaration_site_ids_by_source.get(source_location, set()))]
|
|
|
|
def trace_declaration_sites_declaring(self, type_id: str) -> list[DeclarationSite]:
|
|
return [self._graph.declaration_sites_by_id[item] for item in sorted(self._graph.declaration_site_ids_by_declared_type.get(type_id, set()))]
|
|
|
|
def trace_declaration_sites_for_callable(self, callable_id: str) -> list[DeclarationSite]:
|
|
return [self._graph.declaration_sites_by_id[item] for item in sorted(self._graph.declaration_site_ids_by_callable.get(callable_id, set()))]
|
|
|
|
def trace_binding_sites_in(self, source_location: str) -> list[BindingSite]:
|
|
return [self._graph.binding_sites_by_id[item] for item in sorted(self._graph.binding_site_ids_by_source.get(source_location, set()))]
|
|
|
|
def trace_binding_sites_for_declaration(self, declaration_site_id: str) -> list[BindingSite]:
|
|
return [self._graph.binding_sites_by_id[item] for item in sorted(self._graph.binding_site_ids_by_declaration_site.get(declaration_site_id, set()))]
|
|
|
|
def trace_binding_sites_supplying(self, type_id: str) -> list[BindingSite]:
|
|
return [self._graph.binding_sites_by_id[item] for item in sorted(self._graph.binding_site_ids_by_actual_type.get(type_id, set()))]
|
|
|
|
def binding_sites_returned_by(self, callable_id: str) -> list[BindingSite]:
|
|
return [self._graph.binding_sites_by_id[item] for item in sorted(self._graph.binding_site_ids_by_returned_callable.get(callable_id, set()))]
|
|
|
|
def trace_return_sites_in(self, source_location: str) -> list[ReturnSite]:
|
|
return [self._graph.return_sites_by_id[item] for item in sorted(self._graph.return_site_ids_by_source.get(source_location, set()))]
|
|
|
|
def trace_return_sites_for_callable(self, callable_id: str) -> list[ReturnSite]:
|
|
return [self._graph.return_sites_by_id[item] for item in sorted(self._graph.return_site_ids_by_callable.get(callable_id, set()))]
|
|
|
|
def trace_return_sites_declaring(self, type_id: str) -> list[ReturnSite]:
|
|
return [self._graph.return_sites_by_id[item] for item in sorted(self._graph.return_site_ids_by_declared_type.get(type_id, set()))]
|
|
|
|
def trace_return_sites_returning(self, type_id: str) -> list[ReturnSite]:
|
|
return [self._graph.return_sites_by_id[item] for item in sorted(self._graph.return_site_ids_by_actual_type.get(type_id, set()))]
|
|
|
|
def trace_return_sites_calling(self, callable_id: str) -> list[ReturnSite]:
|
|
return [self._graph.return_sites_by_id[item] for item in sorted(self._graph.return_site_ids_by_returned_callable.get(callable_id, set()))]
|
|
|
|
def trace_indirect_references_for_declaration(self, declaration_site_id: str) -> list[IndirectReference]:
|
|
return [self._graph.indirect_references_by_id[item] for item in sorted(self._graph.indirect_reference_ids_by_declaration_site.get(declaration_site_id, set()))]
|
|
|
|
def trace_indirect_references_through_binding(self, binding_site_id: str) -> list[IndirectReference]:
|
|
return [self._graph.indirect_references_by_id[item] for item in sorted(self._graph.indirect_reference_ids_by_binding_site.get(binding_site_id, set()))]
|
|
|
|
def trace_indirect_references_through_return(self, return_site_id: str) -> list[IndirectReference]:
|
|
return [self._graph.indirect_references_by_id[item] for item in sorted(self._graph.indirect_reference_ids_by_return_site.get(return_site_id, set()))]
|
|
|
|
def trace_indirect_references_using_actual_type(self, type_id: str) -> list[IndirectReference]:
|
|
return [self._graph.indirect_references_by_id[item] for item in sorted(self._graph.indirect_reference_ids_by_actual_type.get(type_id, set()))]
|
|
|
|
def trace_indirect_references_using_parent_relationship(self, relationship_id: str) -> list[IndirectReference]:
|
|
return [self._graph.indirect_references_by_id[item] for item in sorted(self._graph.indirect_reference_ids_by_parent_relationship.get(relationship_id, set()))]
|
|
|
|
def trace_dynamic_dependencies_of(self, source_location: str) -> set[str]:
|
|
return {
|
|
self._graph.type_definitions[reference.actual_type_id].source.location
|
|
for reference_id in self._graph.indirect_reference_ids_by_dependent_source.get(source_location, set())
|
|
if (reference := self._graph.indirect_references_by_id.get(reference_id)) is not None
|
|
and reference.actual_type_id in self._graph.type_definitions
|
|
and self._graph.type_definitions[reference.actual_type_id].source.location != source_location
|
|
}
|
|
|
|
def trace_dynamic_dependents_of(self, source_location: str) -> set[str]:
|
|
return {
|
|
self._graph.declaration_sites_by_id[reference.declaration_site_id].source.location
|
|
for reference_id in self._graph.indirect_reference_ids_by_dependency_source.get(source_location, set())
|
|
if (reference := self._graph.indirect_references_by_id.get(reference_id)) is not None
|
|
and reference.declaration_site_id in self._graph.declaration_sites_by_id
|
|
and self._graph.declaration_sites_by_id[reference.declaration_site_id].source.location != source_location
|
|
}
|
|
|
|
def source_dependencies_of(self, source_location: str, selector: DependencySelector = DependencySelector.BOTH) -> set[str]:
|
|
static = set(self._graph.source_dependencies.get(source_location, set()))
|
|
dynamic = self.trace_dynamic_dependencies_of(source_location)
|
|
if selector is DependencySelector.STATIC:
|
|
return static
|
|
if selector is DependencySelector.DYNAMIC:
|
|
return dynamic - static
|
|
return static | dynamic
|
|
|
|
def source_dependents_of(self, source_location: str, selector: DependencySelector = DependencySelector.BOTH) -> set[str]:
|
|
static = set(self._graph.source_dependents.get(source_location, set()))
|
|
dynamic = self.trace_dynamic_dependents_of(source_location)
|
|
if selector is DependencySelector.STATIC:
|
|
return static
|
|
if selector is DependencySelector.DYNAMIC:
|
|
return dynamic - static
|
|
return static | dynamic
|
|
|
|
def dependency_kind_between(self, dependent: str, dependency: str) -> DependencySelector | None:
|
|
static = dependency in self._graph.source_dependencies.get(dependent, set())
|
|
if static:
|
|
return DependencySelector.STATIC
|
|
if dependency in self.trace_dynamic_dependencies_of(dependent):
|
|
return DependencySelector.DYNAMIC
|
|
return None
|
|
|
|
def upstream_sources_for(self, source_files: list[ClassMeta], selector: DependencySelector = DependencySelector.BOTH) -> list[ClassMeta]:
|
|
locations: set[str] = set()
|
|
for source in source_files:
|
|
if source.location in self._graph.source_by_location:
|
|
locations.update(self.source_dependencies_of(source.location, selector))
|
|
elif source.location in self._graph.test_by_location:
|
|
locations.update(self._graph.sources_for_test.get(source.location, set()))
|
|
return [self._graph.source_by_location[location] for location in sorted(locations) if location in self._graph.source_by_location]
|
|
|
|
def downstream_sources_for(self, source_files: list[ClassMeta], selector: DependencySelector = DependencySelector.BOTH) -> list[ClassMeta]:
|
|
locations = {dependent for source in source_files for dependent in self.source_dependents_of(source.location, selector)}
|
|
return [self._graph.source_by_location[location] for location in sorted(locations) if location in self._graph.source_by_location]
|