1232 lines
55 KiB
Python
1232 lines
55 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import fields
|
|
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.judgement.code_dependency_graph import (
|
|
BindingSite,
|
|
CodeDependencyGraph,
|
|
DeclarationSite,
|
|
DependencySelector,
|
|
DependencyType,
|
|
IndirectReference,
|
|
ParentRelationship,
|
|
ReturnSite,
|
|
)
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import (
|
|
DependencyUpdateInstrumentation,
|
|
PythonAstDependencyGraphMaintainer,
|
|
)
|
|
from tests.support.memory_path import MemoryPath
|
|
from band_aider.judgement.code_dependency_graph_maintainer import CodeDependencyGraphMaintainer
|
|
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
|
|
from band_aider.judgement.frozen_code_dependency_graph_maintainer import FrozenCodeDependencyGraphMaintainer
|
|
from band_aider.judgement.python_ast_dependency_strategy import PythonAstDependencyStrategy
|
|
from band_aider.judgement.source_discovery_strategy import SearchableBeat, SearchableBeatValue
|
|
from band_aider.strategies.python.python_source_discovery import PythonSourceDiscovery
|
|
|
|
|
|
def _meta(location: str) -> ClassMeta:
|
|
return ClassMeta(location, location, [], {})
|
|
|
|
|
|
def _project(
|
|
tmp_path: MemoryPath,
|
|
files: dict[str, str],
|
|
) -> tuple[PythonAstDependencyGraphMaintainer, CodeDependencyGraph, list[ClassMeta], MemoryPath]:
|
|
"""Build one dependency graph from a complete in-memory source population."""
|
|
|
|
src = tmp_path / "src"
|
|
src.mkdir(parents=True)
|
|
metas: list[ClassMeta] = []
|
|
for name, source in files.items():
|
|
(src / name).write_text(source)
|
|
metas.append(_meta(f"src/{name}"))
|
|
maintainer = PythonAstDependencyGraphMaintainer(repo_exec=tmp_path.repository)
|
|
return maintainer, maintainer.build(metas, [], tmp_path, src, tmp_path / "tests"), metas, src
|
|
|
|
|
|
def _refresh(
|
|
maintainer: PythonAstDependencyGraphMaintainer,
|
|
metas: list[ClassMeta],
|
|
src: MemoryPath,
|
|
tmp_path: MemoryPath,
|
|
location: str,
|
|
source: str,
|
|
) -> None:
|
|
"""Replace one memory-backed source and incrementally update its graph records."""
|
|
|
|
(src / (location[4:] if location.startswith("src/") else location)).write_text(source)
|
|
current = [_meta(meta.location) for meta in metas]
|
|
maintainer.upsert(
|
|
next(item for item in current if item.location == location),
|
|
current,
|
|
[],
|
|
tmp_path,
|
|
src,
|
|
tmp_path / "tests",
|
|
)
|
|
|
|
|
|
class _RecordingPythonSourceDiscovery(PythonSourceDiscovery):
|
|
"""Record nominal discovery calls while preserving Python-source semantics."""
|
|
|
|
def __init__(self, src_root: MemoryPath, test_root: MemoryPath) -> None:
|
|
"""Bind the recorder to the owning in-memory repository."""
|
|
|
|
super().__init__(src_root, test_root, repo_exec=src_root.repository)
|
|
self.calls: list[tuple[str, str]] = []
|
|
|
|
def searchable_beats(self, file_name: str, source: str) -> list[SearchableBeatValue]:
|
|
"""Record the supplied evidence before performing real Python discovery."""
|
|
|
|
self.calls.append((file_name, source))
|
|
return super().searchable_beats(file_name, source)
|
|
|
|
|
|
class _PrecomputedPythonSourceDiscovery(_RecordingPythonSourceDiscovery):
|
|
"""Return a fixed precomputed beat population through the nominal strategy API."""
|
|
|
|
def __init__(
|
|
self,
|
|
src_root: MemoryPath,
|
|
test_root: MemoryPath,
|
|
beats: tuple[SearchableBeatValue, ...],
|
|
) -> None:
|
|
"""Store the stationary precomputed discovery result."""
|
|
|
|
super().__init__(src_root, test_root)
|
|
self._beats = beats
|
|
|
|
def searchable_beats(self, file_name: str, source: str) -> list[SearchableBeatValue]:
|
|
"""Record the request and return its fixed precomputed evidence."""
|
|
|
|
self.calls.append((file_name, source))
|
|
return list(self._beats)
|
|
|
|
|
|
def _reverse(adjacency: dict[str, set[str]]) -> dict[str, set[str]]:
|
|
return {
|
|
destination: {
|
|
source
|
|
for source, destinations in adjacency.items()
|
|
if destination in destinations
|
|
}
|
|
for destination in {
|
|
destination
|
|
for destinations in adjacency.values()
|
|
for destination in destinations
|
|
}
|
|
}
|
|
|
|
|
|
def _assert_canonical_indexes_coherent(graph: CodeDependencyGraph) -> None:
|
|
def index(records, key):
|
|
result: dict[str, set[str]] = {}
|
|
for record_id, record in records.items():
|
|
value = key(record)
|
|
if value is not None:
|
|
result.setdefault(value, set()).add(record_id)
|
|
return result
|
|
|
|
assert graph.source_dependents == _reverse(graph.source_dependencies)
|
|
assert graph.sources_for_test == _reverse(graph.tests_for_source)
|
|
assert graph.type_ids_by_source == index(
|
|
graph.type_definitions,
|
|
lambda record: record.source.location,
|
|
)
|
|
assert graph.parent_relationship_ids_by_child == index(
|
|
graph.parent_relationships_by_id,
|
|
lambda record: record.child_type_id,
|
|
)
|
|
assert graph.parent_relationship_ids_by_parent == index(
|
|
graph.parent_relationships_by_id,
|
|
lambda record: record.parent_type_id,
|
|
)
|
|
assert graph.declaration_site_ids_by_source == index(
|
|
graph.declaration_sites_by_id,
|
|
lambda record: record.source.location,
|
|
)
|
|
assert graph.declaration_site_ids_by_declared_type == index(
|
|
graph.declaration_sites_by_id,
|
|
lambda record: record.declared_type_id,
|
|
)
|
|
assert graph.declaration_site_ids_by_callable == index(
|
|
graph.declaration_sites_by_id,
|
|
lambda record: record.callable_id,
|
|
)
|
|
assert graph.binding_site_ids_by_source == index(
|
|
graph.binding_sites_by_id,
|
|
lambda record: record.source.location,
|
|
)
|
|
assert graph.binding_site_ids_by_declaration_site == index(
|
|
graph.binding_sites_by_id,
|
|
lambda record: record.declaration_site_id,
|
|
)
|
|
assert graph.binding_site_ids_by_actual_type == index(
|
|
graph.binding_sites_by_id,
|
|
lambda record: record.direct_actual_type_id,
|
|
)
|
|
assert graph.binding_site_ids_by_returned_callable == index(
|
|
graph.binding_sites_by_id,
|
|
lambda record: record.returned_callable_id,
|
|
)
|
|
assert graph.return_site_ids_by_source == index(
|
|
graph.return_sites_by_id,
|
|
lambda record: record.source.location,
|
|
)
|
|
assert graph.return_site_ids_by_callable == index(
|
|
graph.return_sites_by_id,
|
|
lambda record: record.callable_id,
|
|
)
|
|
assert graph.return_site_ids_by_declared_type == index(
|
|
graph.return_sites_by_id,
|
|
lambda record: record.declared_return_type_id,
|
|
)
|
|
assert graph.return_site_ids_by_actual_type == index(
|
|
graph.return_sites_by_id,
|
|
lambda record: record.direct_actual_type_id,
|
|
)
|
|
assert graph.return_site_ids_by_returned_callable == index(
|
|
graph.return_sites_by_id,
|
|
lambda record: record.returned_callable_id,
|
|
)
|
|
assert graph.indirect_reference_ids_by_declaration_site == index(
|
|
graph.indirect_references_by_id,
|
|
lambda record: record.declaration_site_id,
|
|
)
|
|
assert graph.indirect_reference_ids_by_binding_site == index(
|
|
graph.indirect_references_by_id,
|
|
lambda record: record.binding_site_id,
|
|
)
|
|
expected_return_references: dict[str, set[str]] = {}
|
|
expected_parent_references: dict[str, set[str]] = {}
|
|
expected_actual_references: dict[str, set[str]] = {}
|
|
expected_dependent_references: dict[str, set[str]] = {}
|
|
expected_dependency_references: dict[str, set[str]] = {}
|
|
for reference_id, reference in graph.indirect_references_by_id.items():
|
|
for return_id in reference.return_site_ids:
|
|
expected_return_references.setdefault(return_id, set()).add(reference_id)
|
|
for relationship_id in reference.parent_relationship_ids:
|
|
expected_parent_references.setdefault(relationship_id, set()).add(reference_id)
|
|
expected_actual_references.setdefault(reference.actual_type_id, set()).add(reference_id)
|
|
declaration = graph.declaration_sites_by_id[reference.declaration_site_id]
|
|
expected_dependent_references.setdefault(
|
|
declaration.source.location,
|
|
set(),
|
|
).add(reference_id)
|
|
actual = graph.type_definitions[reference.actual_type_id]
|
|
expected_dependency_references.setdefault(actual.source.location, set()).add(
|
|
reference_id
|
|
)
|
|
assert graph.indirect_reference_ids_by_return_site == expected_return_references
|
|
assert graph.indirect_reference_ids_by_actual_type == expected_actual_references
|
|
assert graph.indirect_reference_ids_by_parent_relationship == expected_parent_references
|
|
assert graph.indirect_reference_ids_by_dependent_source == expected_dependent_references
|
|
assert graph.indirect_reference_ids_by_dependency_source == expected_dependency_references
|
|
|
|
|
|
def _canonical_graph_state(graph: CodeDependencyGraph) -> dict[str, object]:
|
|
return {
|
|
item.name: getattr(graph, item.name)
|
|
for item in fields(CodeDependencyGraph)
|
|
if item.name != "_frozen"
|
|
}
|
|
|
|
|
|
def test_domain_records_have_only_canonical_fields() -> None:
|
|
assert [item.name for item in fields(DependencyType)] == ["type_id", "qualified_name", "local_name", "source"]
|
|
assert "consumer" not in {item.name for item in fields(DeclarationSite)}
|
|
assert "argument_name" not in {item.name for item in fields(BindingSite)}
|
|
assert "actual_type_id" not in {item.name for item in fields(ReturnSite)}
|
|
assert not hasattr(CodeDependencyGraph(), "declaration_ids_by_consumer")
|
|
assert not hasattr(CodeDependencyGraph(), "binding_ids_by_consumer")
|
|
assert not hasattr(CodeDependencyGraph(), "dynamic_dependencies")
|
|
assert not hasattr(CodeDependencyGraph(), "_dynamic_reference_counts")
|
|
assert not hasattr(CodeDependencyGraph(), "static_facts_by_location")
|
|
assert [item.name for item in fields(DependencyUpdateInstrumentation)] == [
|
|
"files_parsed",
|
|
"declaration_sites_added",
|
|
"declaration_sites_removed",
|
|
"binding_sites_added",
|
|
"binding_sites_removed",
|
|
"return_sites_added",
|
|
"return_sites_removed",
|
|
"indirect_references_added",
|
|
"indirect_references_removed",
|
|
"indirect_references_revalidated",
|
|
"parent_relationships_traversed",
|
|
]
|
|
|
|
|
|
def test_graph_fields_are_language_neutral_canonical_records_and_indexes() -> None:
|
|
assert {item.name for item in fields(CodeDependencyGraph)} == {
|
|
"source_by_location",
|
|
"test_by_location",
|
|
"source_dependencies",
|
|
"source_dependents",
|
|
"type_definitions",
|
|
"type_ids_by_source",
|
|
"parent_relationships_by_id",
|
|
"parent_relationship_ids_by_child",
|
|
"parent_relationship_ids_by_parent",
|
|
"declaration_sites_by_id",
|
|
"declaration_site_ids_by_source",
|
|
"declaration_site_ids_by_declared_type",
|
|
"declaration_site_ids_by_callable",
|
|
"binding_sites_by_id",
|
|
"binding_site_ids_by_source",
|
|
"binding_site_ids_by_declaration_site",
|
|
"binding_site_ids_by_actual_type",
|
|
"binding_site_ids_by_returned_callable",
|
|
"return_sites_by_id",
|
|
"return_site_ids_by_source",
|
|
"return_site_ids_by_callable",
|
|
"return_site_ids_by_declared_type",
|
|
"return_site_ids_by_actual_type",
|
|
"return_site_ids_by_returned_callable",
|
|
"indirect_references_by_id",
|
|
"indirect_reference_ids_by_declaration_site",
|
|
"indirect_reference_ids_by_binding_site",
|
|
"indirect_reference_ids_by_return_site",
|
|
"indirect_reference_ids_by_actual_type",
|
|
"indirect_reference_ids_by_parent_relationship",
|
|
"indirect_reference_ids_by_dependent_source",
|
|
"indirect_reference_ids_by_dependency_source",
|
|
"tests_for_source",
|
|
"sources_for_test",
|
|
"_frozen",
|
|
}
|
|
|
|
|
|
def test_python_discovery_emits_dependency_beats_for_protocol_and_return_evidence(tmp_path: MemoryPath) -> None:
|
|
discovery = PythonSourceDiscovery(tmp_path / "src", tmp_path / "tests", repo_exec=tmp_path.repository)
|
|
source = """from typing import Protocol
|
|
class Repository(Protocol):
|
|
def required(self, value: int, /, *, flag: bool): ...
|
|
class SqlRepository(Repository):
|
|
pass
|
|
def create() -> Repository:
|
|
return SqlRepository()
|
|
"""
|
|
beats = [
|
|
beat
|
|
for beat in discovery.searchable_beats("src/repository.py", source)
|
|
if isinstance(beat, SearchableBeat)
|
|
]
|
|
kinds = {beat.kind for beat in beats}
|
|
assert {"dependency_type", "parent_relationship", "dependency_callable", "return_site", "static_facts"} <= kinds
|
|
assert any(dict(beat.attributes).get("direct_abstraction_kind") == "protocol" for beat in beats if beat.kind == "dependency_type")
|
|
assert any(dict(child.attributes).get("parameter_kind") == "positional_only" for beat in beats if beat.kind == "dependency_callable" for child in beat.children)
|
|
assert any(dict(beat.attributes).get("returned_callable_expression") == "SqlRepository" for beat in beats if beat.kind == "return_site")
|
|
|
|
|
|
def test_graph_maintainer_consumes_language_beats_through_its_injected_discovery_contract(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""The graph builds from explicit language discovery evidence over memory only."""
|
|
|
|
src_root = tmp_path / "src"
|
|
src_root.mkdir(parents=True)
|
|
source = "class Widget:\n pass\n"
|
|
(src_root / "widget.py").write_text(source)
|
|
widget = _meta("src/widget.py")
|
|
discovery = _RecordingPythonSourceDiscovery(src_root, tmp_path / "tests")
|
|
maintainer = PythonAstDependencyGraphMaintainer(
|
|
repo_exec=tmp_path.repository,
|
|
source_discovery=discovery,
|
|
)
|
|
|
|
graph = maintainer.build([widget], [], tmp_path, src_root, tmp_path / "tests")
|
|
strategy = PythonAstDependencyStrategy(maintainer)
|
|
|
|
assert discovery.calls == [("src/widget.py", source)]
|
|
assert graph.source_by_location == {"src/widget.py": widget}
|
|
assert strategy.graph is graph
|
|
|
|
|
|
def test_dependency_beats_expose_all_required_generic_evidence(tmp_path: MemoryPath) -> None:
|
|
discovery = PythonSourceDiscovery(
|
|
tmp_path / "src",
|
|
tmp_path / "tests",
|
|
repo_exec=tmp_path.repository,
|
|
)
|
|
beats = [
|
|
beat
|
|
for beat in discovery.searchable_beats(
|
|
"src/example.py",
|
|
"""from typing import Protocol
|
|
class Repository(Protocol):
|
|
name: str
|
|
def fetch(self, value: int, /, *, refresh: bool): ...
|
|
class SqlRepository(Repository):
|
|
name: str
|
|
def fetch(self, value: int, /, *, refresh: bool): ...
|
|
class Service:
|
|
repository: Repository = SqlRepository()
|
|
def __init__(self, repository: Repository):
|
|
self.repository: Repository = repository
|
|
def configure(self, repository: Repository):
|
|
self.repository = repository
|
|
def create() -> Repository:
|
|
return SqlRepository()
|
|
def configure(repository: Repository):
|
|
return None
|
|
Service(SqlRepository()).configure(repository=SqlRepository())
|
|
configure(SqlRepository())
|
|
""",
|
|
)
|
|
if isinstance(beat, SearchableBeat)
|
|
]
|
|
by_kind: dict[str, list[SearchableBeat]] = {}
|
|
for beat in beats:
|
|
by_kind.setdefault(beat.kind, []).append(beat)
|
|
assert {
|
|
"dependency_type",
|
|
"parent_relationship",
|
|
"dependency_callable",
|
|
"declaration_site",
|
|
"binding_site",
|
|
"return_site",
|
|
"static_facts",
|
|
} <= set(by_kind)
|
|
assert {
|
|
"type_id",
|
|
"qualified_name",
|
|
"local_name",
|
|
"source",
|
|
"direct_abstraction_kind",
|
|
"member_methods",
|
|
"member_fields",
|
|
"owner_key",
|
|
} <= set(dict(by_kind["dependency_type"][0].attributes))
|
|
assert {
|
|
"child_type_id",
|
|
"parent_name",
|
|
"source",
|
|
"owner_key",
|
|
} <= set(dict(by_kind["parent_relationship"][0].attributes))
|
|
callable_beat = by_kind["dependency_callable"][0]
|
|
assert {
|
|
"callable_id",
|
|
"owner_type_id",
|
|
"return_annotation",
|
|
"source",
|
|
"owner_key",
|
|
} <= set(dict(callable_beat.attributes))
|
|
assert {
|
|
"name",
|
|
"parameter_kind",
|
|
"ordinal",
|
|
"annotation",
|
|
"owner_key",
|
|
} <= set(dict(callable_beat.children[0].attributes))
|
|
assert {
|
|
"declaration_site_id",
|
|
"declared_name",
|
|
"callable_id",
|
|
"field_owner_id",
|
|
"receiving_name",
|
|
"declaration_kind",
|
|
"source",
|
|
"owner_key",
|
|
} <= set(dict(by_kind["declaration_site"][0].attributes))
|
|
assert {
|
|
"binding_site_id",
|
|
"declaration_site_id",
|
|
"binding_kind",
|
|
"value_kind",
|
|
"actual_type_expression",
|
|
"returned_callable_expression",
|
|
"target_callable_expression",
|
|
"receiver_expression",
|
|
"argument_name",
|
|
"positional_ordinal",
|
|
"source",
|
|
"owner_key",
|
|
} <= set(dict(by_kind["binding_site"][0].attributes))
|
|
assert {
|
|
"return_site_id",
|
|
"callable_id",
|
|
"declared_return_name",
|
|
"actual_type_expression",
|
|
"returned_callable_expression",
|
|
"source",
|
|
"owner_key",
|
|
} <= set(dict(by_kind["return_site"][0].attributes))
|
|
assert all(
|
|
not any(value.__class__.__module__ == "ast" for value in beat.children)
|
|
for beat in beats
|
|
)
|
|
|
|
|
|
def test_maintainer_consumes_precomputed_beats_without_parsing_python(tmp_path: MemoryPath) -> None:
|
|
"""The maintainer consumes supplied nominal discovery evidence once."""
|
|
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
source = "class Repository: pass\n"
|
|
(src / "repository.py").write_text(source)
|
|
discovery = PythonSourceDiscovery(src, tmp_path / "tests", repo_exec=tmp_path.repository)
|
|
cached_beats = tuple(discovery.searchable_beats("src/repository.py", source))
|
|
cached = _PrecomputedPythonSourceDiscovery(src, tmp_path / "tests", cached_beats)
|
|
maintainer = PythonAstDependencyGraphMaintainer(
|
|
repo_exec=tmp_path.repository,
|
|
source_discovery=cached,
|
|
)
|
|
maintainer.build([_meta("src/repository.py")], [], tmp_path, src, tmp_path / "tests")
|
|
assert cached.calls == [("src/repository.py", source)]
|
|
|
|
|
|
def test_initial_build_discovers_each_source_once(tmp_path: MemoryPath) -> None:
|
|
"""Initial graph construction asks its nominal discovery strategy once per source."""
|
|
|
|
src = tmp_path / "src"
|
|
src.mkdir()
|
|
sources = {
|
|
"src/a.py": "from b import B\nclass A: pass\n",
|
|
"src/b.py": "class B: pass\n",
|
|
}
|
|
for location, source in sources.items():
|
|
(tmp_path / location).write_text(source)
|
|
counting = _RecordingPythonSourceDiscovery(src, tmp_path / "tests")
|
|
maintainer = PythonAstDependencyGraphMaintainer(
|
|
repo_exec=tmp_path.repository,
|
|
source_discovery=counting,
|
|
)
|
|
maintainer.build(
|
|
[_meta(location) for location in sources],
|
|
[],
|
|
tmp_path,
|
|
src,
|
|
tmp_path / "tests",
|
|
)
|
|
assert [file_name for file_name, _source in counting.calls] == sorted(sources)
|
|
assert maintainer.last_update_instrumentation.files_parsed == set(sources)
|
|
|
|
|
|
def test_incompatible_binding_is_persisted_and_parent_change_revalidates_without_reparsing_wiring(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "from typing import Protocol\nclass Repository(Protocol):\n def required(self): ...\n",
|
|
"sql.py": "class SqlRepository: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
assert graph.binding_sites_by_id and not graph.indirect_references_by_id
|
|
|
|
class IndexedOnlyDict(dict):
|
|
def items(self):
|
|
raise AssertionError("parent invalidation scanned a full canonical population")
|
|
|
|
graph.binding_sites_by_id = IndexedOnlyDict(graph.binding_sites_by_id)
|
|
graph.indirect_references_by_id = IndexedOnlyDict(graph.indirect_references_by_id)
|
|
_refresh(maintainer, metas, src, tmp_path, "src/sql.py", "from contract import Repository\nclass SqlRepository(Repository): pass\n")
|
|
assert graph.indirect_references_by_id
|
|
assert maintainer.dependency_kind_between("src/service.py", "src/sql.py") is DependencySelector.DYNAMIC
|
|
assert "src/bootstrap.py" not in maintainer.last_update_instrumentation.files_parsed
|
|
|
|
|
|
def test_one_reference_per_binding_and_dynamic_source_indexes_preserve_multiplicity(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, first: Repository, second: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository(), SqlRepository())\n",
|
|
})
|
|
assert len(graph.indirect_references_by_id) == 2
|
|
assert len({item.binding_site_id for item in graph.indirect_references_by_id.values()}) == 2
|
|
assert "src/sql.py" not in graph.source_dependencies.get("src/service.py", set())
|
|
assert maintainer.trace_dynamic_dependencies_of("src/service.py") == {"src/sql.py"}
|
|
assert maintainer.trace_dynamic_dependents_of("src/sql.py") == {"src/service.py"}
|
|
_refresh(maintainer, metas, src, tmp_path, "src/bootstrap.py", "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n")
|
|
assert len(graph.indirect_references_by_id) == 1
|
|
assert maintainer.trace_dynamic_dependencies_of("src/service.py") == {"src/sql.py"}
|
|
|
|
|
|
def test_same_source_dynamic_reference_remains_canonical_but_is_not_projected(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, _, _ = _project(tmp_path, {
|
|
"service.py": "class Parent: pass\nclass Child(Parent): pass\nclass Service:\n def __init__(self, parent: Parent):\n self.parent = parent\nService(Child())\n",
|
|
})
|
|
|
|
assert graph.indirect_references_by_id
|
|
assert graph.indirect_reference_ids_by_dependent_source["src/service.py"]
|
|
assert graph.indirect_reference_ids_by_dependency_source["src/service.py"]
|
|
assert maintainer.trace_dynamic_dependencies_of("src/service.py") == set()
|
|
assert maintainer.trace_dynamic_dependents_of("src/service.py") == set()
|
|
|
|
frozen = FrozenCodeDependencyGraphMaintainer(_graph=graph)
|
|
assert frozen.trace_dynamic_dependencies_of("src/service.py") == set()
|
|
assert frozen.trace_dynamic_dependents_of("src/service.py") == set()
|
|
|
|
|
|
def test_delete_removes_owned_bindings_and_references_using_a_deleted_actual_type(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
reference_id = next(iter(graph.indirect_references_by_id))
|
|
bootstrap = next(meta for meta in metas if meta.location == "src/bootstrap.py")
|
|
remaining = [meta for meta in metas if meta is not bootstrap]
|
|
maintainer.delete(bootstrap, remaining, [], tmp_path, src, tmp_path / "tests")
|
|
assert not graph.binding_site_ids_by_source.get("src/bootstrap.py")
|
|
assert reference_id not in graph.indirect_references_by_id
|
|
assert maintainer.last_update_instrumentation.indirect_references_removed == {
|
|
reference_id
|
|
}
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
maintainer, graph, metas, src = _project(tmp_path / "actual", {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
reference_id = next(iter(graph.indirect_references_by_id))
|
|
sql = next(meta for meta in metas if meta.location == "src/sql.py")
|
|
remaining = [meta for meta in metas if meta is not sql]
|
|
maintainer.delete(sql, remaining, [], tmp_path / "actual", src, tmp_path / "actual" / "tests")
|
|
assert reference_id not in graph.indirect_references_by_id
|
|
assert maintainer.last_update_instrumentation.indirect_references_removed == {
|
|
reference_id
|
|
}
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
|
|
def test_alternate_nominal_path_updates_proof_in_place(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\nclass Middle(Repository): pass\n",
|
|
"child.py": "from contract import Repository, Middle\nclass Child(Middle, Repository): pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, child: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom child import Child\nService(Child())\n",
|
|
})
|
|
reference = next(iter(graph.indirect_references_by_id.values()))
|
|
direct = next(item for item in graph.parent_relationships_by_id.values() if item.child_type_id == reference.actual_type_id and item.parent_type_id == reference.declared_type_id)
|
|
_refresh(maintainer, metas, src, tmp_path, "src/child.py", "from contract import Middle\nclass Child(Middle): pass\n")
|
|
updated = next(iter(graph.indirect_references_by_id.values()))
|
|
assert direct.relationship_id not in updated.parent_relationship_ids
|
|
assert updated.parent_relationship_ids
|
|
|
|
|
|
def test_direct_and_chained_return_provenance_are_ordered(tmp_path: MemoryPath) -> None:
|
|
_, direct, _, _ = _project(tmp_path / "direct", {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"factory.py": "from contract import Repository\nfrom sql import SqlRepository\ndef create() -> Repository:\n return SqlRepository()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import create\nService(create())\n",
|
|
})
|
|
direct_reference = next(iter(direct.indirect_references_by_id.values()))
|
|
assert len(direct.return_sites_by_id) == 1
|
|
assert direct_reference.return_site_ids == tuple(direct.return_sites_by_id)
|
|
|
|
_, graph, _, _ = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"factory.py": "from contract import Repository\nfrom sql import SqlRepository\ndef inner() -> Repository:\n return SqlRepository()\ndef create() -> Repository:\n return inner()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import create\nService(create())\n",
|
|
})
|
|
reference = next(iter(graph.indirect_references_by_id.values()))
|
|
assert [
|
|
graph.return_sites_by_id[return_id].callable_id.rsplit(".", 1)[-1]
|
|
for return_id in reference.return_site_ids
|
|
] == ["create", "inner"]
|
|
|
|
|
|
def test_lazy_parent_factory_tracks_call_argument_provenance_but_local_return_fails_closed(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
_, graph, _, _ = _project(tmp_path, {
|
|
"contract.py": "from typing import Protocol\nclass Parent(Protocol): pass\n",
|
|
"child.py": "from contract import Parent\nclass Child(Parent): pass\nclass WrongChild(Parent): pass\n",
|
|
"factory.py": (
|
|
"from contract import Parent\n"
|
|
"from child import Child, WrongChild\n"
|
|
"def parent_factory() -> Parent:\n"
|
|
" return Child()\n"
|
|
"def parent() -> Parent:\n"
|
|
" return WrongChild()\n"
|
|
"def lazy(parent: Parent) -> Parent:\n"
|
|
" return parent()\n"
|
|
),
|
|
"bootstrap.py": "from factory import lazy, parent_factory\nlazy(parent_factory())\n",
|
|
})
|
|
|
|
lazy_parent = next(
|
|
declaration
|
|
for declaration in graph.declaration_sites_by_id.values()
|
|
if declaration.callable_id == "factory.lazy"
|
|
and declaration.receiving_name == "parent"
|
|
)
|
|
binding = next(
|
|
candidate
|
|
for candidate in graph.binding_sites_by_id.values()
|
|
if candidate.declaration_site_id == lazy_parent.declaration_site_id
|
|
)
|
|
parent_factory_return = next(
|
|
site
|
|
for site in graph.return_sites_by_id.values()
|
|
if site.callable_id == "factory.parent_factory"
|
|
)
|
|
reference = next(
|
|
candidate
|
|
for candidate in graph.indirect_references_by_id.values()
|
|
if candidate.binding_site_id == binding.binding_site_id
|
|
)
|
|
|
|
assert lazy_parent.declared_type_id == "src/contract.py:contract.Parent"
|
|
assert binding.binding_kind == "function_positional"
|
|
assert binding.returned_callable_id == "factory.parent_factory"
|
|
assert reference.declaration_site_id == lazy_parent.declaration_site_id
|
|
assert reference.return_site_ids == (parent_factory_return.return_site_id,)
|
|
assert graph.type_definitions[reference.actual_type_id].qualified_name == "child.Child"
|
|
|
|
lazy_return = next(
|
|
site
|
|
for site in graph.return_sites_by_id.values()
|
|
if site.callable_id == "factory.lazy"
|
|
)
|
|
assert lazy_return.direct_actual_type_id is None
|
|
assert lazy_return.returned_callable_id is None
|
|
assert all(
|
|
lazy_return.return_site_id not in candidate.return_site_ids
|
|
for candidate in graph.indirect_references_by_id.values()
|
|
)
|
|
|
|
|
|
def test_local_callable_collisions_fail_closed_for_call_targets_and_values(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
factory_source = (
|
|
"from contract import Parent\n"
|
|
"from child import Child, WrongChild\n"
|
|
"def target(value: Parent): pass\n"
|
|
"def receiver(value: Parent): pass\n"
|
|
"def factory() -> Parent:\n"
|
|
" return WrongChild()\n"
|
|
"def invoke(target, factory):\n"
|
|
" target(Child())\n"
|
|
" receiver(factory())\n"
|
|
)
|
|
discovery = PythonSourceDiscovery(
|
|
tmp_path / "src",
|
|
tmp_path / "tests",
|
|
repo_exec=tmp_path.repository,
|
|
)
|
|
beats = [
|
|
beat
|
|
for beat in discovery.searchable_beats("src/factory.py", factory_source)
|
|
if isinstance(beat, SearchableBeat)
|
|
]
|
|
target_binding = next(
|
|
dict(beat.attributes)
|
|
for beat in beats
|
|
if beat.kind == "binding_site"
|
|
and dict(beat.attributes).get("target_callable_expression") == "target"
|
|
)
|
|
value_binding = next(
|
|
dict(beat.attributes)
|
|
for beat in beats
|
|
if beat.kind == "binding_site"
|
|
and dict(beat.attributes).get("target_callable_expression") == "receiver"
|
|
)
|
|
assert target_binding["target_callable_is_local"] == "True"
|
|
assert value_binding["returned_callable_is_local"] == "True"
|
|
|
|
_, graph, _, _ = _project(tmp_path, {
|
|
"contract.py": "from typing import Protocol\nclass Parent(Protocol): pass\n",
|
|
"child.py": "from contract import Parent\nclass Child(Parent): pass\nclass WrongChild(Parent): pass\n",
|
|
"factory.py": factory_source,
|
|
})
|
|
target_declaration = next(
|
|
declaration
|
|
for declaration in graph.declaration_sites_by_id.values()
|
|
if declaration.callable_id == "factory.target"
|
|
)
|
|
receiver_declaration = next(
|
|
declaration
|
|
for declaration in graph.declaration_sites_by_id.values()
|
|
if declaration.callable_id == "factory.receiver"
|
|
)
|
|
|
|
assert not any(
|
|
binding.declaration_site_id == target_declaration.declaration_site_id
|
|
for binding in graph.binding_sites_by_id.values()
|
|
)
|
|
receiver_binding = next(
|
|
binding
|
|
for binding in graph.binding_sites_by_id.values()
|
|
if binding.declaration_site_id == receiver_declaration.declaration_site_id
|
|
)
|
|
assert receiver_binding.direct_actual_type_id is None
|
|
assert receiver_binding.returned_callable_id is None
|
|
assert not any(
|
|
reference.binding_site_id == receiver_binding.binding_site_id
|
|
for reference in graph.indirect_references_by_id.values()
|
|
)
|
|
|
|
|
|
def test_cyclic_unresolved_and_ambiguous_return_chains_fail_closed(tmp_path: MemoryPath) -> None:
|
|
_, cyclic, _, _ = _project(tmp_path / "cyclic", {
|
|
"contract.py": "class Repository: pass\nclass Impl(Repository): pass\n",
|
|
"factory.py": "from contract import Repository\ndef first() -> Repository:\n return second()\ndef second() -> Repository:\n return first()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import first\nService(first())\n",
|
|
})
|
|
assert not cyclic.indirect_references_by_id
|
|
_, unresolved, _, _ = _project(tmp_path / "unresolved", {
|
|
"contract.py": "class Repository: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"factory.py": "from contract import Repository\ndef create() -> Repository:\n return missing()\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import create\nService(create())\n",
|
|
})
|
|
assert not unresolved.indirect_references_by_id
|
|
_, ambiguous, _, _ = _project(tmp_path / "ambiguous", {
|
|
"contract.py": "class Repository: pass\n",
|
|
"impl.py": "from contract import Repository\nclass Impl(Repository): pass\n",
|
|
"factory.py": "from contract import Repository\nfrom impl import Impl\ndef create(flag: bool) -> Repository:\n if flag:\n return Impl()\n return Impl()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import create\nService(create(True))\n",
|
|
})
|
|
assert not ambiguous.indirect_references_by_id
|
|
|
|
|
|
def test_inner_return_update_revalidates_an_unchanged_outer_binding(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"factory.py": "from contract import Repository\ndef inner() -> Repository:\n return unresolved()\ndef outer() -> Repository:\n return inner()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import outer\nService(outer())\n",
|
|
})
|
|
assert not graph.indirect_references_by_id
|
|
|
|
_refresh(
|
|
maintainer,
|
|
metas,
|
|
src,
|
|
tmp_path,
|
|
"src/factory.py",
|
|
"from contract import Repository\nfrom sql import SqlRepository\ndef inner() -> Repository:\n return SqlRepository()\ndef outer() -> Repository:\n return inner()\n",
|
|
)
|
|
|
|
reference = next(iter(graph.indirect_references_by_id.values()))
|
|
assert len(reference.return_site_ids) == 2
|
|
assert maintainer.last_update_instrumentation.files_parsed == {"src/factory.py"}
|
|
|
|
|
|
def test_changed_return_revalidates_only_references_with_that_provenance(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"other.py": "from contract import Repository\nclass OtherRepository(Repository): pass\n",
|
|
"first_factory.py": "from contract import Repository\nfrom sql import SqlRepository\ndef create() -> Repository:\n return SqlRepository()\n",
|
|
"second_factory.py": "from contract import Repository\nfrom sql import SqlRepository\ndef create() -> Repository:\n return SqlRepository()\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, first: Repository, second: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom first_factory import create as first\nfrom second_factory import create as second\nService(first(), second())\n",
|
|
})
|
|
references_by_binding = {
|
|
reference.binding_site_id: reference
|
|
for reference in graph.indirect_references_by_id.values()
|
|
}
|
|
changed_reference = next(
|
|
reference
|
|
for binding_id, reference in references_by_binding.items()
|
|
if graph.binding_sites_by_id[binding_id].returned_callable_id == "first_factory.create"
|
|
)
|
|
untouched_reference = next(
|
|
reference
|
|
for binding_id, reference in references_by_binding.items()
|
|
if graph.binding_sites_by_id[binding_id].returned_callable_id == "second_factory.create"
|
|
)
|
|
_refresh(
|
|
maintainer,
|
|
metas,
|
|
src,
|
|
tmp_path,
|
|
"src/first_factory.py",
|
|
"from contract import Repository\nfrom other import OtherRepository\ndef create() -> Repository:\n return OtherRepository()\n",
|
|
)
|
|
assert graph.indirect_references_by_id[changed_reference.indirect_reference_id].actual_type_id != changed_reference.actual_type_id
|
|
assert graph.indirect_references_by_id[untouched_reference.indirect_reference_id] == untouched_reference
|
|
assert maintainer.last_update_instrumentation.indirect_references_revalidated == {
|
|
changed_reference.indirect_reference_id
|
|
}
|
|
assert maintainer.last_update_instrumentation.files_parsed == {"src/first_factory.py"}
|
|
|
|
|
|
def test_static_overlap_keeps_dynamic_evidence_but_static_classification_wins(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, _, _ = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"service.py": "from contract import Repository\nfrom sql import SqlRepository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
assert graph.indirect_references_by_id
|
|
assert maintainer.dependency_kind_between("src/service.py", "src/sql.py") is DependencySelector.STATIC
|
|
assert maintainer.source_dependencies_of("src/service.py", DependencySelector.DYNAMIC) == set()
|
|
|
|
|
|
def test_graph_and_interfaces_expose_canonical_maintainer_operations() -> None:
|
|
for method_name in (
|
|
"dependency_kind_between",
|
|
"source_dependencies_of",
|
|
"source_dependents_of",
|
|
"trace_dynamic_dependencies_of",
|
|
"trace_dynamic_dependents_of",
|
|
"upstream_sources_for",
|
|
"downstream_sources_for",
|
|
"trace_ancestors_of",
|
|
"trace_descendants_of",
|
|
"replace_source",
|
|
"replace_test",
|
|
"add_source_dependency",
|
|
"remove_source_dependency",
|
|
"add_source_test_mapping",
|
|
"remove_source_test_mapping",
|
|
):
|
|
assert not hasattr(CodeDependencyGraph, method_name)
|
|
for method_name in (
|
|
"build",
|
|
"upsert",
|
|
"delete",
|
|
"freeze",
|
|
"trace_types_defined_in",
|
|
"trace_declaration_sites_in",
|
|
"trace_binding_sites_in",
|
|
"trace_return_sites_in",
|
|
"trace_dynamic_dependencies_of",
|
|
"source_dependencies_of",
|
|
"dependency_kind_between",
|
|
"binding_sites_returned_by",
|
|
):
|
|
assert hasattr(CodeDependencyGraphMaintainer, method_name)
|
|
for implementation in (
|
|
PythonAstDependencyGraphMaintainer,
|
|
FrozenCodeDependencyGraphMaintainer,
|
|
):
|
|
assert not {"create", "update", "trace_direct_parents_of", "trace_direct_children_of", "trace_ancestors_of", "trace_descendants_of", "trace_binding_sites_targeting_callable"} & set(dir(implementation))
|
|
assert {"build", "upsert", "delete", "freeze", "binding_sites_returned_by"} <= set(dir(implementation))
|
|
assert hasattr(DependencyDiscoveryStrategy, "graph")
|
|
|
|
|
|
def test_one_hop_traversal_lives_on_the_maintainer_and_recursion_on_the_strategy(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
maintainer, graph, metas, _ = _project(tmp_path, {
|
|
"leaf.py": "def leaf():\n return 1\n",
|
|
"middle.py": "from leaf import leaf\ndef middle():\n return leaf()\n",
|
|
"root.py": "from middle import middle\ndef root():\n return middle()\n",
|
|
})
|
|
leaf = next(meta for meta in metas if meta.location == "src/leaf.py")
|
|
middle = next(meta for meta in metas if meta.location == "src/middle.py")
|
|
root = next(meta for meta in metas if meta.location == "src/root.py")
|
|
strategy = PythonAstDependencyStrategy(maintainer)
|
|
assert maintainer.downstream_sources_for([leaf]) == [middle]
|
|
assert strategy.downstream_sources_for([leaf], recursive=True) == [middle, root]
|
|
assert not {
|
|
"trace_ancestors_of",
|
|
"trace_descendants_of",
|
|
"trace_direct_parents_of",
|
|
"trace_direct_children_of",
|
|
} & set(dir(maintainer))
|
|
assert not {
|
|
"source_dependencies_of",
|
|
"source_dependents_of",
|
|
"dependency_kind_between",
|
|
} & set(dir(graph))
|
|
|
|
|
|
def test_upsert_handles_absent_and_existing_records_and_delete_keeps_indexes_coherent(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
})
|
|
assert not graph.declaration_sites_by_id
|
|
implementation_path = src / "sql.py"
|
|
implementation_path.write_text(
|
|
"from contract import Repository\nclass SqlRepository(Repository): pass\n"
|
|
)
|
|
implementation = _meta("src/sql.py")
|
|
sources = [*metas, implementation]
|
|
|
|
maintainer.upsert(implementation, sources, [], tmp_path, src, tmp_path / "tests")
|
|
implementation_type = next(
|
|
dependency_type
|
|
for dependency_type in maintainer.trace_types_defined_in("src/sql.py")
|
|
if dependency_type.local_name == "SqlRepository"
|
|
)
|
|
assert graph.has_source("src/sql.py")
|
|
assert graph.declaration_sites_by_id
|
|
assert maintainer.last_update_instrumentation.files_parsed == {
|
|
"src/service.py",
|
|
"src/sql.py",
|
|
}
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
implementation_path.write_text("class SqlRepository: pass\n")
|
|
maintainer.upsert(implementation, sources, [], tmp_path, src, tmp_path / "tests")
|
|
assert not maintainer.direct_parent_type_ids_of(implementation_type.type_id)
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
maintainer.delete(implementation, metas, [], tmp_path, src, tmp_path / "tests")
|
|
assert not graph.has_source("src/sql.py")
|
|
assert not maintainer.trace_types_defined_in("src/sql.py")
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
tests_root = tmp_path / "tests"
|
|
tests_root.mkdir(exist_ok=True)
|
|
test_path = tests_root / "test_other.py"
|
|
test_path.write_text("from service import Service\n")
|
|
test_meta = _meta("tests/test_other.py")
|
|
maintainer.upsert(test_meta, metas, [test_meta], tmp_path, src, tests_root)
|
|
assert graph.sources_for_test["tests/test_other.py"] == {"src/service.py"}
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
test_path.write_text("value = 1\n")
|
|
maintainer.upsert(test_meta, metas, [test_meta], tmp_path, src, tests_root)
|
|
assert "tests/test_other.py" not in graph.sources_for_test
|
|
|
|
maintainer.delete(test_meta, metas, [], tmp_path, src, tests_root)
|
|
assert "tests/test_other.py" not in graph.test_by_location
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
|
|
def test_incremental_delta_preserves_unaffected_map_identity_and_parse_frontier(tmp_path: MemoryPath) -> None:
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"other.py": "class Other: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
types_identity = id(graph.type_definitions)
|
|
static_identity = id(graph.source_dependencies)
|
|
implementation = next(
|
|
dependency_type
|
|
for dependency_type in graph.type_definitions.values()
|
|
if dependency_type.local_name == "SqlRepository"
|
|
)
|
|
implementation_parent_index = graph.parent_relationship_ids_by_child[
|
|
implementation.type_id
|
|
]
|
|
service_declaration_index = graph.declaration_site_ids_by_source["src/service.py"]
|
|
dynamic_reference_index = graph.indirect_reference_ids_by_dependent_source[
|
|
"src/service.py"
|
|
]
|
|
_refresh(maintainer, metas, src, tmp_path, "src/other.py", "class Other:\n pass\n")
|
|
assert id(graph.type_definitions) == types_identity
|
|
assert id(graph.source_dependencies) == static_identity
|
|
assert graph.parent_relationship_ids_by_child[implementation.type_id] is implementation_parent_index
|
|
assert graph.declaration_site_ids_by_source["src/service.py"] is service_declaration_index
|
|
assert graph.indirect_reference_ids_by_dependent_source["src/service.py"] is dynamic_reference_index
|
|
assert maintainer.last_update_instrumentation.files_parsed == {"src/other.py"}
|
|
|
|
|
|
def test_incremental_upsert_does_not_construct_a_graph_shell_or_rebuild_references(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""An unrelated source update preserves existing canonical reference records."""
|
|
|
|
maintainer, graph, metas, src = _project(tmp_path, {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "from contract import Repository\nclass SqlRepository(Repository): pass\n",
|
|
"other.py": "class Other: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
})
|
|
binding_records = graph.binding_sites_by_id
|
|
reference_records = graph.indirect_references_by_id
|
|
existing_references = set(reference_records)
|
|
|
|
_refresh(maintainer, metas, src, tmp_path, "src/other.py", "class Other:\n pass\n")
|
|
|
|
assert graph.indirect_references_by_id
|
|
assert set(graph.indirect_references_by_id) == existing_references
|
|
assert graph.binding_sites_by_id is binding_records
|
|
assert graph.indirect_references_by_id is reference_records
|
|
assert maintainer.last_update_instrumentation.files_parsed == {"src/other.py"}
|
|
|
|
|
|
def test_clean_build_and_equivalent_incremental_upsert_have_identical_canonical_state(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
initial_files = {
|
|
"contract.py": "class Repository: pass\n",
|
|
"sql.py": "class SqlRepository: pass\n",
|
|
"service.py": "from contract import Repository\nclass Service:\n def __init__(self, repository: Repository): pass\n",
|
|
"bootstrap.py": "from service import Service\nfrom sql import SqlRepository\nService(SqlRepository())\n",
|
|
}
|
|
final_sql = "from contract import Repository\nclass SqlRepository(Repository): pass\n"
|
|
incremental, incremental_graph, incremental_metas, incremental_src = _project(
|
|
tmp_path / "incremental",
|
|
initial_files,
|
|
)
|
|
_refresh(
|
|
incremental,
|
|
incremental_metas,
|
|
incremental_src,
|
|
tmp_path / "incremental",
|
|
"src/sql.py",
|
|
final_sql,
|
|
)
|
|
final_files = {**initial_files, "sql.py": final_sql}
|
|
_, clean_graph, _, _ = _project(tmp_path / "clean", final_files)
|
|
assert _canonical_graph_state(incremental_graph) == _canonical_graph_state(clean_graph)
|
|
_assert_canonical_indexes_coherent(incremental_graph)
|
|
_assert_canonical_indexes_coherent(clean_graph)
|
|
|
|
|
|
def test_incremental_monolith_to_split_matches_clean_build_in_both_orders(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Rehome every source-owned record without scanning an unrelated source."""
|
|
|
|
monolith = """class Parent: pass
|
|
class Child(Parent): pass
|
|
class Service:
|
|
def __init__(self, parent: Parent):
|
|
self.parent = parent
|
|
def parent_factory() -> Parent:
|
|
return Child()
|
|
Service(parent_factory())
|
|
"""
|
|
split_files = {
|
|
"parent.py": "class Parent: pass\n",
|
|
"child.py": "from parent import Parent\nclass Child(Parent): pass\n",
|
|
"factory.py": "from parent import Parent\nfrom child import Child\ndef parent_factory() -> Parent:\n return Child()\n",
|
|
"service.py": "from parent import Parent\nclass Service:\n def __init__(self, parent: Parent):\n self.parent = parent\n",
|
|
"bootstrap.py": "from service import Service\nfrom factory import parent_factory\nService(parent_factory())\n",
|
|
}
|
|
orders = {
|
|
"dependencies_first": tuple(split_files),
|
|
"dependents_first": tuple(reversed(tuple(split_files))),
|
|
}
|
|
final_files = {"unrelated.py": "class Unrelated: pass\n", **split_files}
|
|
|
|
for order_name, order in orders.items():
|
|
root = tmp_path / order_name
|
|
maintainer, graph, metas, src = _project(
|
|
root,
|
|
{
|
|
"all.py": monolith,
|
|
"unrelated.py": final_files["unrelated.py"],
|
|
},
|
|
)
|
|
old_location = "src/all.py"
|
|
old_type_ids = set(graph.type_ids_by_source[old_location])
|
|
old_relationship_ids = set(graph.parent_relationships_by_id)
|
|
old_declaration_ids = set(graph.declaration_sites_by_id)
|
|
old_binding_ids = set(graph.binding_sites_by_id)
|
|
old_return_ids = set(graph.return_sites_by_id)
|
|
old_reference_ids = set(graph.indirect_references_by_id)
|
|
types_identity = id(graph.type_definitions)
|
|
static_identity = id(graph.source_dependencies)
|
|
unrelated_type_index = graph.type_ids_by_source["src/unrelated.py"]
|
|
old_meta = next(meta for meta in metas if meta.location == old_location)
|
|
remaining = [meta for meta in metas if meta.location != old_location]
|
|
(src / "all.py").unlink()
|
|
maintainer.delete(old_meta, remaining, [], root, src, root / "tests")
|
|
assert maintainer.last_update_instrumentation.files_parsed == set()
|
|
assert old_location not in graph.source_by_location
|
|
assert old_type_ids.isdisjoint(graph.type_definitions)
|
|
assert old_relationship_ids.isdisjoint(graph.parent_relationships_by_id)
|
|
assert old_declaration_ids.isdisjoint(graph.declaration_sites_by_id)
|
|
assert old_binding_ids.isdisjoint(graph.binding_sites_by_id)
|
|
assert old_return_ids.isdisjoint(graph.return_sites_by_id)
|
|
assert old_reference_ids.isdisjoint(graph.indirect_references_by_id)
|
|
assert id(graph.type_definitions) == types_identity
|
|
assert id(graph.source_dependencies) == static_identity
|
|
assert graph.type_ids_by_source["src/unrelated.py"] is unrelated_type_index
|
|
|
|
for name in order:
|
|
location = f"src/{name}"
|
|
(src / name).write_text(split_files[name])
|
|
meta = _meta(location)
|
|
remaining.append(meta)
|
|
maintainer.upsert(meta, remaining, [], root, src, root / "tests")
|
|
expected_frontier = (
|
|
{f"src/{item}" for item in split_files}
|
|
if order_name == "dependents_first" and name == "parent.py"
|
|
else {location}
|
|
)
|
|
assert maintainer.last_update_instrumentation.files_parsed == expected_frontier
|
|
assert id(graph.type_definitions) == types_identity
|
|
assert id(graph.source_dependencies) == static_identity
|
|
assert graph.type_ids_by_source["src/unrelated.py"] is unrelated_type_index
|
|
|
|
assert set(graph.source_by_location) == {"src/unrelated.py", *{
|
|
f"src/{name}" for name in split_files
|
|
}}
|
|
assert {record.source.location for record in graph.type_definitions.values()} == {
|
|
"src/unrelated.py",
|
|
"src/parent.py",
|
|
"src/child.py",
|
|
"src/service.py",
|
|
}
|
|
assert {record.source_location for record in graph.parent_relationships_by_id.values()} == {
|
|
"src/child.py"
|
|
}
|
|
assert {record.source.location for record in graph.declaration_sites_by_id.values()} == {
|
|
"src/service.py"
|
|
}
|
|
assert {record.source.location for record in graph.binding_sites_by_id.values()} == {
|
|
"src/bootstrap.py"
|
|
}
|
|
assert {record.source.location for record in graph.return_sites_by_id.values()} == {
|
|
"src/factory.py"
|
|
}
|
|
assert all(old_location not in record_id for record_id in graph.parent_relationships_by_id)
|
|
assert all(old_location not in record_id for record_id in graph.declaration_sites_by_id)
|
|
assert all(old_location not in record_id for record_id in graph.binding_sites_by_id)
|
|
assert all(old_location not in record_id for record_id in graph.return_sites_by_id)
|
|
assert all(old_location not in record_id for record_id in graph.indirect_references_by_id)
|
|
assert all(
|
|
old_location not in index
|
|
for index in (
|
|
graph.source_dependencies,
|
|
graph.source_dependents,
|
|
graph.type_ids_by_source,
|
|
graph.declaration_site_ids_by_source,
|
|
graph.binding_site_ids_by_source,
|
|
graph.return_site_ids_by_source,
|
|
graph.indirect_reference_ids_by_dependent_source,
|
|
graph.indirect_reference_ids_by_dependency_source,
|
|
maintainer._beats_by_source,
|
|
maintainer._callable_ids_by_source,
|
|
maintainer._aliases_by_source,
|
|
maintainer._static_facts_by_location,
|
|
)
|
|
)
|
|
assert graph.source_dependencies == {
|
|
"src/child.py": {"src/parent.py"},
|
|
"src/factory.py": {"src/parent.py", "src/child.py"},
|
|
"src/service.py": {"src/parent.py"},
|
|
"src/bootstrap.py": {"src/factory.py", "src/service.py"},
|
|
}
|
|
assert graph.source_dependents == _reverse(graph.source_dependencies)
|
|
assert maintainer.trace_dynamic_dependencies_of("src/service.py") == {
|
|
"src/child.py"
|
|
}
|
|
assert maintainer.trace_dynamic_dependents_of("src/child.py") == {
|
|
"src/service.py"
|
|
}
|
|
_assert_canonical_indexes_coherent(graph)
|
|
|
|
_, clean_graph, _, _ = _project(tmp_path / f"{order_name}_clean", final_files)
|
|
assert _canonical_graph_state(graph) == _canonical_graph_state(clean_graph)
|
|
_assert_canonical_indexes_coherent(clean_graph)
|
|
|
|
|
|
def test_graph_contains_no_recursive_or_compatibility_policy_methods() -> None:
|
|
graph_methods = set(dir(CodeDependencyGraph()))
|
|
assert not {
|
|
"trace_ancestors_of",
|
|
"trace_dynamic_dependencies_of",
|
|
"source_dependencies_of",
|
|
"dependency_kind_between",
|
|
} & graph_methods
|