392 lines
12 KiB
Python
392 lines
12 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.source_discovery_strategy import (
|
|
SearchableBeatValue,
|
|
SourceDiscoveryStrategy,
|
|
)
|
|
from band_aider.judgement.strict_parser import LexicalCheckFactory, StrictParser
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _InnerParser(CodeParser):
|
|
"""_InnerParser records forwarded strict 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 strict forwarded parse call."""
|
|
|
|
self.calls.append((file_name, prompt))
|
|
stream.read()
|
|
return ClassMeta(
|
|
file_name,
|
|
"inner",
|
|
["inner"],
|
|
{"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 strict 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 = 0
|
|
|
|
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."""
|
|
|
|
del file_name, source
|
|
self.calls += 1
|
|
if isinstance(self.beats, BaseException):
|
|
raise self.beats
|
|
return self.beats
|
|
|
|
|
|
class _LexicalCheck(LexicalAnchorCheck):
|
|
"""_LexicalCheck returns one configured strict 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 strict lexical result."""
|
|
|
|
del anchors, candidate_beats
|
|
self.calls += 1
|
|
return self.result
|
|
|
|
|
|
def _parser(
|
|
anchors: Iterable[str],
|
|
strategy: SourceDiscoveryStrategy,
|
|
lexical_factory: LexicalCheckFactory | None = None,
|
|
) -> tuple[StrictParser, _InnerParser]:
|
|
"""Return a strict parser with fake collaborators."""
|
|
|
|
inner = _InnerParser()
|
|
return (
|
|
StrictParser(
|
|
inner=inner,
|
|
anchors=anchors,
|
|
source_discovery_strategy=strategy,
|
|
lexical_check_factory=lexical_factory,
|
|
),
|
|
inner,
|
|
)
|
|
|
|
|
|
def test_strict_parser_lexical_true_forwards() -> None:
|
|
"""Verify strict policy forwards only on lexical success."""
|
|
|
|
parser, inner = _parser(
|
|
("cli.py",),
|
|
_Strategy(["cli.py"]),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO("source"), "fix cli.py")
|
|
|
|
assert result.summary == "inner"
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_strict_parser_lexical_false_skips() -> None:
|
|
"""Verify strict policy skips on lexical rejection."""
|
|
|
|
parser, inner = _parser(
|
|
("cli.py",),
|
|
_Strategy(["README.md"]),
|
|
lambda anchors, beats: _LexicalCheck(False),
|
|
)
|
|
|
|
result = parser.parse_stream("README.md", StringIO("source"), "fix cli.py")
|
|
|
|
assert result.relevance == []
|
|
assert result.relevant_symbols == {}
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_real_lexical_does_not_forward_shared_extension_only() -> None:
|
|
"""Verify strict parser does not forward unrelated Python files for a cli.py anchor."""
|
|
|
|
inner = _InnerParser()
|
|
parser = StrictParser(
|
|
inner=inner,
|
|
anchors=("cli.py",),
|
|
source_discovery_strategy=_Strategy(["band_aider/__init__.py", "__init__.py", "__init__"]),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/__init__.py", StringIO(""), "fix cli.py")
|
|
|
|
assert result.relevance == []
|
|
assert result.relevant_symbols == {}
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_real_lexical_forwards_matching_filename() -> None:
|
|
"""Verify strict parser still forwards a real filename match."""
|
|
|
|
inner = _InnerParser()
|
|
parser = StrictParser(
|
|
inner=inner,
|
|
anchors=("cli.py",),
|
|
source_discovery_strategy=_Strategy(["band_aider/cli.py", "cli.py", "cli"]),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO("source"), "fix cli.py")
|
|
|
|
assert result.summary == "inner"
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_strict_parser_empty_source_lexical_false_skips() -> None:
|
|
"""Verify strict parser skips an empty file when lexical filename matching fails."""
|
|
|
|
inner = _InnerParser()
|
|
strategy = _Strategy(["band_aider/__init__.py", "__init__.py", "__init__"])
|
|
parser = StrictParser(
|
|
inner=inner,
|
|
anchors=("cli.py",),
|
|
source_discovery_strategy=strategy,
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/__init__.py", StringIO(" \n"), "fix cli.py")
|
|
|
|
assert result.relevance == []
|
|
assert result.relevant_symbols == {}
|
|
assert strategy.calls == 1
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_empty_source_lexical_true_forwards() -> None:
|
|
"""Verify strict parser can forward an empty file when filename matching succeeds."""
|
|
|
|
inner = _InnerParser()
|
|
strategy = _Strategy(["band_aider/cli.py", "cli.py", "cli"])
|
|
parser = StrictParser(
|
|
inner=inner,
|
|
anchors=("cli.py",),
|
|
source_discovery_strategy=strategy,
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO(""), "populate cli.py")
|
|
|
|
assert result.summary == "inner"
|
|
assert strategy.calls == 1
|
|
assert len(inner.calls) == 1
|
|
|
|
|
|
def test_strict_parser_empty_anchors_empty_beats_and_extraction_error_skip(caplog) -> None:
|
|
"""Verify strict empty and extraction failure cases fail closed."""
|
|
|
|
empty_anchor_parser, empty_anchor_inner = _parser(
|
|
(),
|
|
_Strategy(["cli.py"]),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
)
|
|
empty_beats_parser, empty_beats_inner = _parser(
|
|
("cli.py",),
|
|
_Strategy([]),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
)
|
|
error_parser, error_inner = _parser(
|
|
("cli.py",),
|
|
_Strategy(RuntimeError("extract failed")),
|
|
lambda anchors, beats: _LexicalCheck(True),
|
|
)
|
|
|
|
assert empty_anchor_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert empty_beats_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert error_parser.parse_stream("x.py", StringIO("source"), "fix").relevance == []
|
|
assert empty_anchor_inner.calls == []
|
|
assert empty_beats_inner.calls == []
|
|
assert error_inner.calls == []
|
|
assert "beat_extraction=error" in caplog.text
|
|
|
|
|
|
def test_strict_parser_lexical_error_skips() -> None:
|
|
"""Verify strict lexical errors are false and do not forward."""
|
|
|
|
parser, inner = _parser(
|
|
("cli.py",),
|
|
_Strategy(["cli.py"]),
|
|
lambda anchors, beats: _LexicalCheck(False, error="ValueError: lexical"),
|
|
)
|
|
|
|
result = parser.parse_stream("band_aider/cli.py", StringIO("source"), "fix cli.py")
|
|
|
|
assert result.relevance == []
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_does_not_substitute_prompt_or_expand_anchors() -> None:
|
|
"""Verify strict factories receive exactly classifier anchors and not the full prompt."""
|
|
|
|
seen: list[tuple[str, ...]] = []
|
|
|
|
def lexical_factory(anchors, beats):
|
|
"""Capture strict anchors exactly as supplied."""
|
|
|
|
seen.append(tuple(anchors))
|
|
return _LexicalCheck(False)
|
|
|
|
parser, inner = _parser(("cli.py",), _Strategy(["other"]), lexical_factory)
|
|
|
|
parser.parse_stream("other.txt", StringIO("source"), "full prompt mentions many terms")
|
|
|
|
assert seen == [("cli.py",)]
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_zero_matching_files_remain_zero() -> None:
|
|
"""Verify strict parse_files performs no broad retry when no files match."""
|
|
|
|
parser, inner = _parser(
|
|
("cli.py",),
|
|
_Strategy(["README.md"]),
|
|
lambda anchors, beats: _LexicalCheck(False),
|
|
)
|
|
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")
|
|
|
|
results = parser.parse_files([str(first), str(second)], "fix cli.py")
|
|
|
|
assert len(results) == 2
|
|
assert all(result.relevance == [] and result.relevant_symbols == {} for result in results)
|
|
assert inner.calls == []
|
|
|
|
|
|
def test_strict_parser_creates_fresh_lexical_checks_per_file() -> None:
|
|
"""Verify strict parser creates a fresh lexical check for each candidate file."""
|
|
|
|
checks: list[_LexicalCheck] = []
|
|
|
|
def lexical_factory(anchors, beats):
|
|
"""Create one fresh false lexical check."""
|
|
|
|
check = _LexicalCheck(False)
|
|
checks.append(check)
|
|
return check
|
|
|
|
parser, _inner = _parser(("cli.py",), _Strategy(["README.md"]), lexical_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 cli.py")
|
|
|
|
assert len(checks) == 2
|
|
assert checks[0] is not checks[1]
|
|
|
|
|
|
def test_strict_parser_delegates_an_accepted_markdown_filename_lexically() -> None:
|
|
"""A matching non-Python filename is decided by lexical evidence alone."""
|
|
|
|
lexical = _LexicalCheck(True)
|
|
parser, inner = _parser(
|
|
("README.md",),
|
|
_Strategy(["README.md"]),
|
|
lambda anchors, beats: lexical,
|
|
)
|
|
|
|
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")]
|