655 lines
22 KiB
Python
655 lines
22 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from tests.band_aider_loop_impl import *
|
|
from band_aider.judgement.code_dependency_graph import CodeDependencyGraph, DependencySelector
|
|
from band_aider.judgement.composite_dependency_discovery_strategy import (
|
|
CompositeDependencyDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.dependency_discovery_strategy import (
|
|
DependencyDirection,
|
|
DependencyTraversalCache,
|
|
DependencyTraversalQuery,
|
|
)
|
|
from band_aider.judgement.python_ast_dependency_graph_maintainer import (
|
|
PythonAstDependencyGraphMaintainer,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
def _python_ast_strategy_from_manager(file_manager) -> PythonAstDependencyStrategy:
|
|
"""Return a Python AST dependency strategy through the explicit maintainer seam."""
|
|
|
|
maintainer = PythonAstDependencyGraphMaintainer(repo_exec=file_manager.repo_exec)
|
|
maintainer.build(
|
|
file_manager.src_classes,
|
|
file_manager.test_classes,
|
|
file_manager.repo_root,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
)
|
|
return PythonAstDependencyStrategy(maintainer)
|
|
|
|
|
|
def _graph_strategy(
|
|
*,
|
|
sources: list[ClassMeta],
|
|
tests: list[ClassMeta] | None = None,
|
|
dependencies: dict[str, set[str]] | None = None,
|
|
dependents: dict[str, set[str]] | None = None,
|
|
sources_for_test: dict[str, set[str]] | None = None,
|
|
) -> PythonAstDependencyStrategy:
|
|
"""Return a Python AST strategy backed by an in-memory graph."""
|
|
|
|
graph = CodeDependencyGraph(
|
|
source_by_location={source.location: source for source in sources},
|
|
test_by_location={test.location: test for test in tests or []},
|
|
source_dependencies=dependencies or {},
|
|
source_dependents=dependents or {},
|
|
sources_for_test=sources_for_test or {},
|
|
)
|
|
return PythonAstDependencyStrategy(
|
|
PythonAstDependencyGraphMaintainer(_graph=graph, repo_exec=MemoryRepoExec())
|
|
)
|
|
|
|
|
|
class _CountingInnerStrategy(PythonAstDependencyStrategy):
|
|
"""Record composite traversal cache isolation through a nominal strategy."""
|
|
|
|
def __init__(self, upstream: list[ClassMeta], downstream: list[ClassMeta]) -> None:
|
|
self.upstream = upstream
|
|
self.downstream = downstream
|
|
self.upstream_calls = 0
|
|
self.downstream_calls = 0
|
|
self.cache_args: list[DependencyTraversalCache | None] = []
|
|
super().__init__(PythonAstDependencyGraphMaintainer(repo_exec=MemoryRepoExec()))
|
|
|
|
def upstream_sources_for(
|
|
self,
|
|
files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
test_root: Path | None = None,
|
|
recursive: bool = False,
|
|
dependency_cache: DependencyTraversalCache | None = None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
del files, all_source_files, src_root, test_root, recursive, selector
|
|
self.upstream_calls += 1
|
|
self.cache_args.append(dependency_cache)
|
|
return list(self.upstream)
|
|
|
|
def downstream_sources_for(
|
|
self,
|
|
source_files: list[ClassMeta],
|
|
all_source_files: list[ClassMeta] | None = None,
|
|
src_root: Path | None = None,
|
|
recursive: bool = False,
|
|
dependency_cache: DependencyTraversalCache | None = None,
|
|
selector: DependencySelector = DependencySelector.BOTH,
|
|
) -> list[ClassMeta]:
|
|
del source_files, all_source_files, src_root, recursive, selector
|
|
self.downstream_calls += 1
|
|
self.cache_args.append(dependency_cache)
|
|
return list(self.downstream)
|
|
|
|
|
|
class _CompositeDependencyStrategy(CompositeDependencyDiscoveryStrategy):
|
|
"""Supply the graph property omitted by the production protocol declaration."""
|
|
|
|
@property
|
|
def graph(self) -> CodeDependencyGraph:
|
|
"""Expose the primary maintained graph through the nominal test subtype."""
|
|
|
|
return self.maintainer.graph
|
|
|
|
|
|
def test_python_ast_dependency_strategy_finds_direct_tests_and_downstream_sources(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify deterministic import/path evidence drives dependency discovery."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.src_root / "bar.py",
|
|
"from foo import foo\n\ndef bar():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.test_root / "test_foo.py",
|
|
"from foo import foo\n\ndef test_foo():\n assert foo() == 1\n"
|
|
)
|
|
file_manager.resync()
|
|
|
|
strategy = _python_ast_strategy_from_manager(file_manager)
|
|
source = file_manager.current_src("src/foo.py")
|
|
assert source is not None
|
|
|
|
assert [test.location for test in strategy.direct_tests_for(
|
|
[source],
|
|
file_manager.test_classes,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
)] == [
|
|
"tests/test_foo.py"
|
|
]
|
|
assert [source.location for source in strategy.downstream_sources_for(
|
|
[source],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
)] == [
|
|
"src/bar.py"
|
|
]
|
|
|
|
|
|
def test_python_ast_dependency_strategy_finds_upstream_sources_for_source_and_test(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify graph-backed upstream source lookup works for source and test inputs."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(file_manager.src_root / "base.py", "def base():\n return 1\n")
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.src_root / "foo.py",
|
|
"from base import base\n\ndef foo():\n return base()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.test_root / "test_foo.py",
|
|
"from foo import foo\nfrom base import base\n\ndef test_foo():\n assert foo() == base()\n"
|
|
)
|
|
file_manager.resync()
|
|
|
|
strategy = _python_ast_strategy_from_manager(file_manager)
|
|
source = file_manager.current_src("src/foo.py")
|
|
test = file_manager.current_test("tests/test_foo.py")
|
|
assert source is not None
|
|
assert test is not None
|
|
|
|
assert [item.location for item in strategy.upstream_sources_for(
|
|
[source],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
)] == ["src/base.py"]
|
|
assert [item.location for item in strategy.upstream_sources_for(
|
|
[test],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
)] == ["src/base.py", "src/foo.py"]
|
|
|
|
|
|
def test_python_ast_dependency_strategy_recurses_and_builds_joint_beam(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
"""Verify recursive reachability and deferred beam intersection."""
|
|
|
|
file_manager = _file_manager(tmp_path)
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.src_root / "mid.py",
|
|
"from foo import foo\n\ndef mid():\n return foo()\n"
|
|
)
|
|
file_manager.repo_exec.write_text(
|
|
file_manager.src_root / "leaf.py",
|
|
"from mid import mid\n\ndef leaf():\n return mid()\n"
|
|
)
|
|
file_manager.resync()
|
|
strategy = _python_ast_strategy_from_manager(file_manager)
|
|
root = file_manager.current_src("src/foo.py")
|
|
leaf = file_manager.current_src("src/leaf.py")
|
|
assert root is not None
|
|
assert leaf is not None
|
|
|
|
assert [source.location for source in strategy.downstream_sources_for(
|
|
[root],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
recursive=True,
|
|
)] == ["src/leaf.py", "src/mid.py"]
|
|
assert [source.location for source in strategy.upstream_sources_for(
|
|
[leaf],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
file_manager.test_root,
|
|
recursive=True,
|
|
)] == ["src/foo.py", "src/mid.py"]
|
|
assert [source.location for source in strategy.joint_beam_sources_for(
|
|
[root],
|
|
[leaf],
|
|
file_manager.src_classes,
|
|
file_manager.src_root,
|
|
)] == ["src/foo.py", "src/leaf.py", "src/mid.py"]
|
|
|
|
|
|
def test_python_ast_dependency_strategy_joint_beam_includes_current_endpoint() -> None:
|
|
"""Verify a deferred beam includes its relevant current endpoint."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
target = ClassMeta("src/target.py", "target", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[a, b, c, target],
|
|
dependencies={
|
|
b.location: {a.location},
|
|
c.location: {b.location},
|
|
target.location: {c.location},
|
|
},
|
|
dependents={
|
|
a.location: {b.location},
|
|
b.location: {c.location},
|
|
c.location: {target.location},
|
|
},
|
|
)
|
|
|
|
assert [source.location for source in strategy.joint_beam_sources_for(
|
|
[a],
|
|
[target],
|
|
)] == [a.location, b.location, c.location, target.location]
|
|
|
|
|
|
def test_python_ast_dependency_strategy_joint_beam_excludes_unrelated_current_endpoint() -> None:
|
|
"""Verify a deferred beam excludes a current source without a target path."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
target = ClassMeta("src/target.py", "target", [], {})
|
|
x = ClassMeta("src/x.py", "x", [], {})
|
|
y = ClassMeta("src/y.py", "y", [], {})
|
|
z = ClassMeta("src/z.py", "z", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[a, b, target, x, y, z],
|
|
dependencies={
|
|
b.location: {a.location},
|
|
target.location: {b.location},
|
|
y.location: {x.location},
|
|
z.location: {y.location},
|
|
},
|
|
dependents={
|
|
a.location: {b.location},
|
|
b.location: {target.location},
|
|
x.location: {y.location},
|
|
y.location: {z.location},
|
|
},
|
|
)
|
|
|
|
joint_locations = {
|
|
source.location
|
|
for source in strategy.joint_beam_sources_for([a, x], [target])
|
|
}
|
|
|
|
assert joint_locations == {a.location, b.location, target.location}
|
|
assert not joint_locations & {x.location, y.location, z.location}
|
|
|
|
|
|
def test_dependency_strategy_has_path_is_directional_and_stops_at_dead_relays() -> None:
|
|
"""Verify targeted reachability avoids full closure materialization and dead relays."""
|
|
|
|
x = ClassMeta("src/x.py", "x", [], {})
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
z = ClassMeta("src/z.py", "z", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[x, a, b, z],
|
|
dependencies={
|
|
a.location: {x.location},
|
|
b.location: {a.location},
|
|
z.location: {b.location},
|
|
},
|
|
dependents={
|
|
x.location: {a.location},
|
|
a.location: {b.location},
|
|
b.location: {z.location},
|
|
},
|
|
)
|
|
|
|
assert strategy.has_path(z.location, x.location)
|
|
assert strategy.has_path(
|
|
x.location,
|
|
z.location,
|
|
direction=DependencyDirection.DOWNSTREAM,
|
|
)
|
|
assert not strategy.has_path(
|
|
z.location,
|
|
x.location,
|
|
excluded_locations=frozenset({a.location}),
|
|
)
|
|
assert not strategy.has_path(
|
|
x.location,
|
|
z.location,
|
|
direction=DependencyDirection.UPSTREAM,
|
|
)
|
|
|
|
|
|
def test_composite_has_path_does_not_stitch_independent_dependency_views() -> None:
|
|
"""Verify composite reachability preserves per-view traversal boundaries."""
|
|
|
|
x = ClassMeta("src/x.py", "x", [], {})
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
z = ClassMeta("src/z.py", "z", [], {})
|
|
first = _graph_strategy(
|
|
sources=[x, a, z],
|
|
dependents={x.location: {a.location}},
|
|
)
|
|
second = _graph_strategy(
|
|
sources=[x, a, z],
|
|
dependents={a.location: {z.location}},
|
|
)
|
|
composite = _CompositeDependencyStrategy((first, second))
|
|
|
|
assert not composite.has_path(
|
|
x.location,
|
|
z.location,
|
|
direction=DependencyDirection.DOWNSTREAM,
|
|
)
|
|
|
|
|
|
def test_dependency_traversal_cache_keys_and_directional_entries() -> None:
|
|
"""Verify upstream/downstream caches use normalized directional query keys."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
d = ClassMeta("src/d.py", "d", [], {})
|
|
test = ClassMeta("tests/test_a.py", "test", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[a, b, c, d],
|
|
tests=[test],
|
|
dependencies={
|
|
"src/a.py": {"src/b.py"},
|
|
"src/b.py": {"src/d.py"},
|
|
"src/c.py": {"src/b.py", "src/d.py"},
|
|
},
|
|
dependents={
|
|
"src/b.py": {"src/a.py", "src/c.py"},
|
|
"src/d.py": {"src/b.py", "src/c.py"},
|
|
},
|
|
sources_for_test={"tests/test_a.py": {"src/a.py", "src/b.py"}},
|
|
)
|
|
cache = DependencyTraversalCache()
|
|
|
|
direct_upstream = strategy.upstream_sources_for(
|
|
[a, c, a],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
Path("tests"),
|
|
dependency_cache=cache,
|
|
)
|
|
recursive_upstream = strategy.upstream_sources_for(
|
|
[a],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
Path("tests"),
|
|
recursive=True,
|
|
dependency_cache=cache,
|
|
)
|
|
direct_downstream = strategy.downstream_sources_for(
|
|
[b],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
dependency_cache=cache,
|
|
)
|
|
recursive_downstream = strategy.downstream_sources_for(
|
|
[d],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
recursive=True,
|
|
dependency_cache=cache,
|
|
)
|
|
|
|
assert [source.location for source in direct_upstream] == ["src/b.py", "src/d.py"]
|
|
assert [source.location for source in recursive_upstream] == ["src/b.py", "src/d.py"]
|
|
assert [source.location for source in direct_downstream] == ["src/a.py", "src/c.py"]
|
|
assert [source.location for source in recursive_downstream] == [
|
|
"src/a.py",
|
|
"src/b.py",
|
|
"src/c.py",
|
|
]
|
|
assert DependencyTraversalQuery(("src/a.py", "src/c.py"), False) in cache.upstream
|
|
assert DependencyTraversalQuery(("src/a.py",), True) in cache.upstream
|
|
assert DependencyTraversalQuery(("src/b.py",), False) in cache.downstream
|
|
assert DependencyTraversalQuery(("src/d.py",), True) in cache.downstream
|
|
|
|
cached = strategy.upstream_sources_for(
|
|
[c, a],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
Path("tests"),
|
|
dependency_cache=cache,
|
|
)
|
|
direct_upstream.append(a)
|
|
assert [source.location for source in cached] == ["src/b.py", "src/d.py"]
|
|
assert [source.location for source in cache.upstream[
|
|
DependencyTraversalQuery(("src/a.py", "src/c.py"), False)
|
|
]] == ["src/b.py", "src/d.py"]
|
|
|
|
upstream_from_test = strategy.upstream_sources_for(
|
|
[test],
|
|
[a, b, c, d],
|
|
Path("src"),
|
|
Path("tests"),
|
|
dependency_cache=cache,
|
|
)
|
|
assert [source.location for source in upstream_from_test] == ["src/a.py", "src/b.py"]
|
|
assert not hasattr(strategy, "dependency_cache")
|
|
|
|
|
|
def test_directional_cache_entries_do_not_cross_contaminate() -> None:
|
|
"""Verify a cached upstream value never satisfies a downstream query."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[a, b, c],
|
|
dependencies={"src/a.py": {"src/b.py"}},
|
|
dependents={"src/a.py": {"src/c.py"}},
|
|
)
|
|
cache = DependencyTraversalCache()
|
|
query = DependencyTraversalQuery(("src/a.py",), False)
|
|
cache.upstream[query] = (b,)
|
|
|
|
downstream = strategy.downstream_sources_for(
|
|
[a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
dependency_cache=cache,
|
|
)
|
|
|
|
assert [source.location for source in downstream] == ["src/c.py"]
|
|
assert cache.downstream[query] == (c,)
|
|
|
|
|
|
def test_composite_dependency_strategy_caches_final_merged_directional_results() -> None:
|
|
"""Verify composite caches only final merged results and isolates inner cache calls."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
first = _CountingInnerStrategy([b], [c])
|
|
second = _CountingInnerStrategy([c], [b])
|
|
composite = _CompositeDependencyStrategy((first, second))
|
|
cache = DependencyTraversalCache()
|
|
|
|
upstream = composite.upstream_sources_for(
|
|
[a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
Path("tests"),
|
|
dependency_cache=cache,
|
|
)
|
|
downstream = composite.downstream_sources_for(
|
|
[a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
dependency_cache=cache,
|
|
)
|
|
upstream_hit = composite.upstream_sources_for(
|
|
[a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
Path("tests"),
|
|
dependency_cache=cache,
|
|
)
|
|
downstream_hit = composite.downstream_sources_for(
|
|
[a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
dependency_cache=cache,
|
|
)
|
|
|
|
assert [source.location for source in upstream] == ["src/b.py", "src/c.py"]
|
|
assert [source.location for source in downstream] == ["src/b.py", "src/c.py"]
|
|
assert upstream_hit == upstream
|
|
assert downstream_hit == downstream
|
|
assert first.upstream_calls == 1
|
|
assert second.upstream_calls == 1
|
|
assert first.downstream_calls == 1
|
|
assert second.downstream_calls == 1
|
|
assert all(cache_arg is None for cache_arg in [*first.cache_args, *second.cache_args])
|
|
assert cache.upstream[DependencyTraversalQuery(("src/a.py",), False)] == tuple(upstream)
|
|
assert cache.downstream[DependencyTraversalQuery(("src/a.py",), False)] == tuple(downstream)
|
|
|
|
|
|
def test_stable_quicksort_reranks_dependencies_before_dependents() -> None:
|
|
"""Verify cached upstream membership moves report_ui.py before terminal_ui.py."""
|
|
|
|
report = ClassMeta("src/report_ui.py", "report", [], {})
|
|
terminal = ClassMeta("src/terminal_ui.py", "terminal", [], {})
|
|
unrelated = ClassMeta("src/unrelated.py", "unrelated", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[report, terminal, unrelated],
|
|
dependencies={"src/terminal_ui.py": {"src/report_ui.py"}},
|
|
)
|
|
|
|
reranked = strategy.rerank_sources_by_upstream(
|
|
[terminal, report, unrelated],
|
|
[report, terminal, unrelated],
|
|
Path("src"),
|
|
Path("tests"),
|
|
DependencyTraversalCache(),
|
|
)
|
|
|
|
assert [source.location for source in reranked] == [
|
|
"src/report_ui.py",
|
|
"src/terminal_ui.py",
|
|
"src/unrelated.py",
|
|
]
|
|
|
|
|
|
def test_stable_quicksort_preserves_unrelated_cycles_duplicates_and_degenerates() -> None:
|
|
"""Verify stable equal partitions and unique-root handling."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[a, b, c],
|
|
dependencies={
|
|
"src/a.py": {"src/b.py"},
|
|
"src/b.py": {"src/a.py"},
|
|
},
|
|
)
|
|
cache = DependencyTraversalCache()
|
|
|
|
assert strategy.rerank_sources_by_upstream([], [], Path("src"), Path("tests"), cache) == []
|
|
assert strategy.rerank_sources_by_upstream([a], [a], Path("src"), Path("tests"), cache) == [a]
|
|
assert [source.location for source in strategy.rerank_sources_by_upstream(
|
|
[a, b, c, a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
Path("tests"),
|
|
DependencyTraversalCache(),
|
|
)] == ["src/a.py", "src/b.py", "src/c.py"]
|
|
|
|
|
|
def test_rerank_prepares_one_recursive_query_per_unique_root_before_sorting() -> None:
|
|
"""Verify quicksort comparisons do not perform graph traversal."""
|
|
|
|
a = ClassMeta("src/a.py", "a", [], {})
|
|
b = ClassMeta("src/b.py", "b", [], {})
|
|
c = ClassMeta("src/c.py", "c", [], {})
|
|
|
|
class _CountingStrategy(PythonAstDependencyStrategy):
|
|
def __init__(self, maintainer) -> None:
|
|
super().__init__(maintainer)
|
|
self.calls: list[tuple[str, bool]] = []
|
|
|
|
def upstream_sources_for(
|
|
self,
|
|
files,
|
|
all_source_files=None,
|
|
src_root=None,
|
|
test_root=None,
|
|
recursive=False,
|
|
dependency_cache=None,
|
|
selector=DependencySelector.BOTH,
|
|
):
|
|
self.calls.append((files[0].location, recursive))
|
|
return super().upstream_sources_for(
|
|
files,
|
|
all_source_files,
|
|
src_root,
|
|
test_root,
|
|
recursive=recursive,
|
|
dependency_cache=dependency_cache,
|
|
selector=selector,
|
|
)
|
|
|
|
graph = CodeDependencyGraph(source_by_location={item.location: item for item in [a, b, c]})
|
|
strategy = _CountingStrategy(
|
|
PythonAstDependencyGraphMaintainer(_graph=graph, repo_exec=MemoryRepoExec())
|
|
)
|
|
|
|
first = strategy.rerank_sources_by_upstream(
|
|
[a, b, c, a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
Path("tests"),
|
|
DependencyTraversalCache(),
|
|
)
|
|
second = strategy.rerank_sources_by_upstream(
|
|
[a, b, c, a],
|
|
[a, b, c],
|
|
Path("src"),
|
|
Path("tests"),
|
|
DependencyTraversalCache(),
|
|
)
|
|
|
|
assert [source.location for source in first] == ["src/a.py", "src/b.py", "src/c.py"]
|
|
assert [source.location for source in second] == ["src/a.py", "src/b.py", "src/c.py"]
|
|
assert strategy.calls == [
|
|
("src/a.py", True),
|
|
("src/b.py", True),
|
|
("src/c.py", True),
|
|
("src/a.py", True),
|
|
("src/b.py", True),
|
|
("src/c.py", True),
|
|
]
|
|
|
|
|
|
def test_stable_rerank_orders_a_dependency_chain_without_reordering_independent_peers() -> None:
|
|
"""The reranker puts prerequisites first while retaining independent-peer order."""
|
|
|
|
dependency = ClassMeta("src/dependency.py", "dependency", [], {})
|
|
dependent = ClassMeta("src/dependent.py", "dependent", [], {})
|
|
first_peer = ClassMeta("src/first_peer.py", "first peer", [], {})
|
|
second_peer = ClassMeta("src/second_peer.py", "second peer", [], {})
|
|
strategy = _graph_strategy(
|
|
sources=[dependency, dependent, first_peer, second_peer],
|
|
dependencies={dependent.location: {dependency.location}},
|
|
)
|
|
|
|
reranked = strategy.rerank_sources_by_upstream(
|
|
[dependent, first_peer, second_peer, dependency],
|
|
[dependency, dependent, first_peer, second_peer],
|
|
Path("src"),
|
|
Path("tests"),
|
|
DependencyTraversalCache(),
|
|
)
|
|
|
|
locations = [source.location for source in reranked]
|
|
assert locations.index(dependency.location) < locations.index(dependent.location)
|
|
assert locations.index(first_peer.location) < locations.index(second_peer.location)
|