425 lines
14 KiB
Python
425 lines
14 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
|
|
from collections.abc import Callable
|
|
from band_aider.judgement.source_discovery_strategy import CallTraceNode
|
|
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, TextIO
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.lexical_anchor_check import LexicalAnchorCheck
|
|
from band_aider.judgement.model2vec_code_parser import (
|
|
LexicalCheckFactory,
|
|
Model2VecSimilarityCheck,
|
|
NarrowParser,
|
|
SemanticCheckFactory,
|
|
)
|
|
from band_aider.judgement.source_discovery_strategy import (
|
|
SearchableBeatValue,
|
|
SourceDiscoveryStrategy,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _InnerParser(CodeParser):
|
|
"""_InnerParser records forwarded parse calls."""
|
|
|
|
def __init__(self, repo_exec: MemoryRepoExec | None = None) -> None:
|
|
self.calls: list[tuple[str, str]] = []
|
|
self.repo_exec = repo_exec or MemoryRepoExec()
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Parse one memory-backed file through the configured repository boundary."""
|
|
|
|
source = self.repo_exec.read_text(Path(file_name))
|
|
return self.parse_stream(file_name, StringIO(source), prompt, content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Record one forwarded parse call."""
|
|
|
|
self.calls.append((file_name, prompt))
|
|
stream.read()
|
|
return ClassMeta(
|
|
location=file_name,
|
|
summary="inner",
|
|
relevance=["inner"],
|
|
relevant_symbols={"Symbol": "forwarded"},
|
|
content_type=content_type,
|
|
)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Parse multiple files from disk."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _Strategy(SourceDiscoveryStrategy):
|
|
"""_Strategy returns configured searchable beats."""
|
|
|
|
def visit_call_sites(self, target: CallTraceNode, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, *, target_args: dict[str, CallTraceNode] | None = None, pristine: Mapping[str, str] | None = None) -> None:
|
|
"""Supply no imported calls in this stationary discovery double."""
|
|
del target, already_seen, predicate
|
|
|
|
def get_call_tree(self, location: str, name: str | None = None, *, line: int | None = None, depth: int = 1, already_seen: set[str] | None = None, predicate: Callable[[CallTraceNode], bool] | None = None, pristine: Mapping[str, str] | None = None) -> CallTraceNode:
|
|
"""Reject origins because this double contains no callable definitions."""
|
|
raise LookupError(location)
|
|
|
|
def __init__(self, beats: list[SearchableBeatValue] | BaseException) -> None:
|
|
self.beats = beats
|
|
self.calls: list[tuple[str, str]] = []
|
|
|
|
def get_sources(
|
|
self,
|
|
recursive: bool = True,
|
|
type: Any | None = None,
|
|
) -> list[Path]:
|
|
"""Return no source files for parser tests."""
|
|
|
|
del recursive, type
|
|
return []
|
|
|
|
def get_tests(
|
|
self,
|
|
recursive: bool = True,
|
|
type: Any | None = None,
|
|
) -> list[Path]:
|
|
"""Return no test files for parser tests."""
|
|
|
|
del recursive, type
|
|
return []
|
|
|
|
def searchable_beats(
|
|
self,
|
|
file_name: str,
|
|
source: str,
|
|
) -> list[SearchableBeatValue]:
|
|
"""Return configured beats or raise a configured error."""
|
|
|
|
self.calls.append((file_name, source))
|
|
if isinstance(self.beats, BaseException):
|
|
raise self.beats
|
|
return self.beats
|
|
|
|
|
|
class _LexicalCheck(LexicalAnchorCheck):
|
|
"""_LexicalCheck returns one configured lexical decision."""
|
|
|
|
def __init__(self, result: bool, *, error: str = "") -> None:
|
|
super().__init__(changes=[], anchors=(), candidate_beats=())
|
|
self.result = result
|
|
self.error = error
|
|
self.reason = "configured"
|
|
self.best_anchor = "cli.py"
|
|
self.best_candidate = "cli.py"
|
|
self.edit_distance = 0
|
|
self.allowed_edit_distance = 0
|
|
self.comparison_count = 1
|
|
self.calls = 0
|
|
|
|
def __call__(
|
|
self,
|
|
anchors: Iterable[str] | None = None,
|
|
candidate_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Return the configured lexical result."""
|
|
|
|
del anchors, candidate_beats
|
|
self.calls += 1
|
|
return self.result
|
|
|
|
|
|
class _SemanticCheck(Model2VecSimilarityCheck):
|
|
"""_SemanticCheck returns one configured semantic decision."""
|
|
|
|
def __init__(self, result: bool, *, error: str = "") -> None:
|
|
super().__init__(changes=[], request_text="", searchable_beats=())
|
|
self.result = result
|
|
self.error = error
|
|
self.reason = "configured"
|
|
self.best_candidate = "cli.py"
|
|
self.score = 0.5
|
|
self.best_request_chunk = ""
|
|
self.best_request_chunk_index = None
|
|
self.calls = 0
|
|
|
|
def __call__(
|
|
self,
|
|
request_text: str | None = None,
|
|
searchable_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Return the configured semantic result."""
|
|
|
|
del request_text, searchable_beats
|
|
self.calls += 1
|
|
return self.result
|
|
|
|
def match_request_chunks(
|
|
self,
|
|
request_chunks: Iterable[str],
|
|
searchable_beats: Iterable[str] | None = None,
|
|
) -> bool:
|
|
"""Return the configured chunk-aware semantic decision."""
|
|
|
|
del request_chunks, searchable_beats
|
|
self.calls += 1
|
|
return self.result
|
|
|
|
|
|
class _UnavailableSemanticCheck(_SemanticCheck):
|
|
"""_UnavailableSemanticCheck models an unavailable local semantic package."""
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(False)
|
|
self.reason = "model2vec_package_unavailable"
|
|
|
|
|
|
def _parser(
|
|
strategy: SourceDiscoveryStrategy,
|
|
lexical_factory: LexicalCheckFactory | None = None,
|
|
semantic_factory: SemanticCheckFactory | None = None,
|
|
) -> tuple[NarrowParser, _InnerParser]:
|
|
"""Return a narrow parser with fake collaborators."""
|
|
|
|
inner = _InnerParser()
|
|
return (
|
|
NarrowParser(
|
|
inner=inner,
|
|
source_discovery_strategy=strategy,
|
|
anchors=("cli.py",),
|
|
lexical_check_factory=lexical_factory,
|
|
semantic_check_factory=semantic_factory,
|
|
),
|
|
inner,
|
|
)
|
|
|
|
|
|
def test_narrow_parser_lexical_true_forwards_and_short_circuits_semantic() -> None:
|
|
"""Verify lexical success forwards without creating semantic checks."""
|
|
|
|
lexical = _LexicalCheck(True)
|
|
parser, inner = _parser(
|
|
_Strategy(["cli.py"]),
|
|
lambda anchors, beats: lexical,
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO("source"), "fix cli.py")
|
|
|
|
assert result.summary == "inner"
|
|
assert lexical.calls == 1
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_narrow_parser_lexical_false_semantic_true_forwards() -> None:
|
|
"""Verify narrow policy forwards when semantic succeeds after lexical fails."""
|
|
|
|
parser, inner = _parser(
|
|
_Strategy(["Widget.set_mode"]),
|
|
lambda anchors, beats: _LexicalCheck(False),
|
|
lambda request, beats: _SemanticCheck(True),
|
|
)
|
|
|
|
result = parser.parse_stream("pkg/widget.py", StringIO("source"), "fix widget")
|
|
|
|
assert result.summary == "inner"
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_narrow_parser_lexical_false_semantic_false_skips() -> None:
|
|
"""Verify narrow policy skips when both checks reject."""
|
|
|
|
parser, inner = _parser(
|
|
_Strategy(["Widget.helper"]),
|
|
lambda anchors, beats: _LexicalCheck(False),
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("pkg/widget.py", StringIO("source"), "fix widget")
|
|
|
|
assert result.relevance == []
|
|
assert result.relevant_symbols == {}
|
|
assert "Search prefilter skipped advisory parse" in result.summary
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_narrow_parser_empty_beats_and_extraction_errors_skip(caplog) -> None:
|
|
"""Verify empty beats and beat extraction errors fail closed."""
|
|
|
|
empty_parser, empty_inner = _parser(
|
|
_Strategy([]),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
lambda request, beats: _SemanticCheck(True),
|
|
)
|
|
error_parser, error_inner = _parser(
|
|
_Strategy(RuntimeError("extract failed")),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
lambda request, beats: _SemanticCheck(True),
|
|
)
|
|
|
|
assert empty_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert error_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert empty_inner.calls == []
|
|
assert error_inner.calls == []
|
|
assert "beat_extraction=error" in caplog.text
|
|
|
|
|
|
def test_narrow_parser_empty_source_lexical_false_skips_before_semantic() -> None:
|
|
"""Verify narrow parser lets empty files do lexical matching but skips semantic."""
|
|
|
|
strategy = _Strategy(["cli.py"])
|
|
lexical = _LexicalCheck(False)
|
|
parser, inner = _parser(
|
|
strategy,
|
|
lambda anchors, beats: lexical,
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/__init__.py", StringIO(" \n\t"), "fix cli.py")
|
|
|
|
assert result.relevance == []
|
|
assert result.relevant_symbols == {}
|
|
assert strategy.calls == [("band_aider/__init__.py", " \n\t")]
|
|
assert lexical.calls == 1
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_narrow_parser_empty_source_lexical_true_forwards() -> None:
|
|
"""Verify narrow parser can forward an empty file when lexical filename matching succeeds."""
|
|
|
|
lexical = _LexicalCheck(True)
|
|
parser, inner = _parser(
|
|
_Strategy(["band_aider/cli.py", "cli.py", "cli"]),
|
|
lambda anchors, beats: lexical,
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO(""), "populate cli.py")
|
|
|
|
assert result.summary == "inner"
|
|
assert lexical.calls == 1
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_narrow_parser_matcher_errors_do_not_forward() -> None:
|
|
"""Verify lexical and semantic errors behave as false decisions."""
|
|
|
|
lexical_error_parser, lexical_inner = _parser(
|
|
_Strategy(["Widget"]),
|
|
lambda anchors, beats: _LexicalCheck(False, error="ValueError: lexical"),
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
semantic_error_parser, semantic_inner = _parser(
|
|
_Strategy(["Widget"]),
|
|
lambda anchors, beats: _LexicalCheck(False),
|
|
lambda request, beats: _SemanticCheck(False, error="RuntimeError: semantic"),
|
|
)
|
|
unavailable_parser = NarrowParser(
|
|
inner=_InnerParser(),
|
|
source_discovery_strategy=_Strategy(["Widget"]),
|
|
anchors=("Widget",),
|
|
lexical_check_factory=lambda anchors, beats: _LexicalCheck(False),
|
|
semantic_check_factory=lambda request, beats: _UnavailableSemanticCheck(),
|
|
)
|
|
|
|
assert lexical_error_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert semantic_error_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert unavailable_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert lexical_inner.calls == []
|
|
assert semantic_inner.calls == []
|
|
|
|
|
|
def test_narrow_parser_creates_fresh_checks_per_file() -> None:
|
|
"""Verify narrow parser factories are invoked once per evaluated file."""
|
|
|
|
lexical_checks: list[_LexicalCheck] = []
|
|
semantic_checks: list[_SemanticCheck] = []
|
|
|
|
def lexical_factory(anchors, beats):
|
|
"""Create one fresh false lexical check."""
|
|
|
|
check = _LexicalCheck(False)
|
|
lexical_checks.append(check)
|
|
return check
|
|
|
|
def semantic_factory(request, beats):
|
|
"""Create one fresh false semantic check."""
|
|
|
|
check = _SemanticCheck(False)
|
|
semantic_checks.append(check)
|
|
return check
|
|
|
|
parser, inner = _parser(_Strategy(["anything"]), lexical_factory, semantic_factory)
|
|
first = inner.repo_exec.root / "first.txt"
|
|
second = inner.repo_exec.root / "second.txt"
|
|
inner.repo_exec.write_text(first, "first")
|
|
inner.repo_exec.write_text(second, "second")
|
|
|
|
parser.parse_files([str(first), str(second)], "fix")
|
|
|
|
assert len(lexical_checks) == 2
|
|
assert len(semantic_checks) == 2
|
|
assert lexical_checks[0] is not lexical_checks[1]
|
|
|
|
|
|
def test_narrow_parser_delegates_an_accepted_non_python_filename() -> None:
|
|
"""Lexical evidence, rather than a filename suffix, owns parser delegation."""
|
|
|
|
lexical = _LexicalCheck(True)
|
|
parser, inner = _parser(
|
|
_Strategy(["README.md"]),
|
|
lambda anchors, beats: lexical,
|
|
lambda request, beats: _SemanticCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("README.md", StringIO("# fixed documentation\n"), "fix docs")
|
|
|
|
assert result.summary == "inner"
|
|
assert lexical.calls == 1
|
|
assert inner.calls == [("README.md", "fix docs")]
|
|
|
|
|
|
def test_model2vec_check_preserves_the_injected_memory_cache_root() -> None:
|
|
"""Verify semantic checks retain the explicitly supplied local cache location."""
|
|
|
|
cache_root = MemoryRepoExec().root / "cache"
|
|
check = Model2VecSimilarityCheck(
|
|
changes=[],
|
|
request_text="fix widget",
|
|
searchable_beats=("Widget",),
|
|
cache_dir=cache_root,
|
|
)
|
|
|
|
assert check.cache_dir == cache_root
|
|
|
|
|
|
def test_model2vec_check_is_stationary_without_a_model_call() -> None:
|
|
"""Verify the test semantic double stays local when a model is unavailable."""
|
|
|
|
check = _UnavailableSemanticCheck()
|
|
|
|
assert check.match_request_chunks(("fix widget",), ("Widget",)) is False
|
|
assert check.reason == "model2vec_package_unavailable"
|