306 lines
10 KiB
Python
306 lines
10 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import pytest
|
|
|
|
from band_aider.judgement.model2vec_code_parser import (
|
|
MODEL2VEC_SIMILARITY_THRESHOLD,
|
|
Model2VecSimilarityCheck,
|
|
RequestChunker,
|
|
)
|
|
|
|
|
|
def _check(
|
|
*,
|
|
request_text: str = "repair widget mode",
|
|
searchable_beats: list[str] | None = None,
|
|
) -> Model2VecSimilarityCheck:
|
|
"""Create one check without invoking optional-package or filesystem loading."""
|
|
|
|
return Model2VecSimilarityCheck(
|
|
changes=[],
|
|
request_text=request_text,
|
|
searchable_beats=(
|
|
["Widget.set_mode"]
|
|
if searchable_beats is None
|
|
else searchable_beats
|
|
),
|
|
cache_dir=Path("/testbed/model2vec"),
|
|
)
|
|
|
|
|
|
def _unit_vector_for(score: float) -> list[float]:
|
|
"""Return a unit vector whose cosine with [1, 0] is the supplied score."""
|
|
|
|
return [score, math.sqrt(1 - score * score)]
|
|
|
|
|
|
def test_model2vec_similarity_threshold_values() -> None:
|
|
"""Cosine scores compare inclusively against the explicit policy threshold."""
|
|
|
|
check = _check()
|
|
below = check._cosine_similarity((1.0, 0.0), _unit_vector_for(0.71))
|
|
equal = check._cosine_similarity(
|
|
(1.0, 0.0),
|
|
_unit_vector_for(MODEL2VEC_SIMILARITY_THRESHOLD),
|
|
)
|
|
above = check._cosine_similarity((1.0, 0.0), _unit_vector_for(0.73))
|
|
|
|
assert MODEL2VEC_SIMILARITY_THRESHOLD == 0.72
|
|
assert below < MODEL2VEC_SIMILARITY_THRESHOLD
|
|
assert equal >= MODEL2VEC_SIMILARITY_THRESHOLD
|
|
assert above > MODEL2VEC_SIMILARITY_THRESHOLD
|
|
|
|
|
|
def test_model2vec_similarity_retains_maximum_valid_score() -> None:
|
|
"""The deterministic vector operation preserves the largest candidate score."""
|
|
|
|
check = _check(searchable_beats=["Widget.helper", "Widget.set_mode"])
|
|
scores = [
|
|
check._cosine_similarity((1.0, 0.0), _unit_vector_for(0.4)),
|
|
check._cosine_similarity((1.0, 0.0), _unit_vector_for(0.9)),
|
|
]
|
|
|
|
assert max(scores) == 0.9
|
|
assert check.searchable_beats is not None
|
|
|
|
|
|
def test_model2vec_similarity_empty_inputs_return_false() -> None:
|
|
"""Verify empty request text and empty beats fail closed."""
|
|
|
|
assert _check(request_text="")() is False
|
|
assert _check(searchable_beats=[] )() is False
|
|
|
|
|
|
def test_model2vec_similarity_candidate_scoring_rejects_blank_request_before_model_loading() -> None:
|
|
"""Blank candidate scoring stops before consulting an optional model runtime."""
|
|
|
|
check = _check()
|
|
|
|
assert check.score_candidates("", ["Widget.set_mode"]) == []
|
|
|
|
|
|
def test_model2vec_similarity_chunk_matching_rejects_empty_request_before_model_loading() -> None:
|
|
"""Empty request chunks stop before consulting an optional model runtime."""
|
|
|
|
check = _check()
|
|
|
|
assert check.match_request_chunks([], ["Widget.set_mode"]) is False
|
|
assert check.reason == "empty_request_chunks"
|
|
|
|
|
|
def test_model2vec_similarity_chunk_matching_rejects_empty_beats_before_model_loading() -> None:
|
|
"""Empty candidate beats stop before consulting an optional model runtime."""
|
|
|
|
check = _check()
|
|
|
|
assert check.match_request_chunks(["repair widget mode"], []) is False
|
|
assert check.reason == "empty_searchable_beats"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_model2vec_similarity_package_unavailable_returns_false(
|
|
pytestconfig: pytest.Config,
|
|
) -> None:
|
|
"""Verify an isolated runtime without Model2Vec rejects semantic matching."""
|
|
|
|
child_program = (
|
|
"import json\n"
|
|
"from pathlib import Path\n"
|
|
"from band_aider.judgement.model2vec_code_parser import Model2VecSimilarityCheck\n"
|
|
"check = Model2VecSimilarityCheck(\n"
|
|
" changes=[],\n"
|
|
" request_text='repair widget mode',\n"
|
|
" searchable_beats=['Widget.set_mode'],\n"
|
|
" cache_dir=Path('/testbed/model2vec'),\n"
|
|
")\n"
|
|
"result = check()\n"
|
|
"print(json.dumps({'result': result, 'reason': check.reason, 'changes': len(check.changes)}, sort_keys=True))\n"
|
|
)
|
|
project_root = pytestconfig.rootpath.parent
|
|
|
|
completed = subprocess.run(
|
|
(sys.executable, "-E", "-S", "-c", child_program),
|
|
cwd=project_root,
|
|
env={},
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=4,
|
|
)
|
|
|
|
assert (completed.returncode, completed.stderr) == (0, "")
|
|
assert json.loads(completed.stdout) == {
|
|
"changes": 1,
|
|
"reason": "model2vec_package_unavailable",
|
|
"result": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_model2vec_similarity_cached_model_unavailable_raises_explicitly(
|
|
pytestconfig: pytest.Config,
|
|
) -> None:
|
|
"""The isolated local-only model loader fails without a download."""
|
|
|
|
child_program = (
|
|
"import json\n"
|
|
"from pathlib import Path\n"
|
|
"from band_aider.judgement.model2vec_code_parser import Model2VecSimilarityCheck\n"
|
|
"check = Model2VecSimilarityCheck(\n"
|
|
" changes=[],\n"
|
|
" request_text='repair widget mode',\n"
|
|
" searchable_beats=['Widget.set_mode'],\n"
|
|
" cache_dir=Path('/testbed/model2vec'),\n"
|
|
")\n"
|
|
"try:\n"
|
|
" check._load_model()\n"
|
|
"except FileNotFoundError as error:\n"
|
|
" print(json.dumps({'error': str(error)}, sort_keys=True))\n"
|
|
"else:\n"
|
|
" raise AssertionError('expected cached-model failure')\n"
|
|
)
|
|
project_root = pytestconfig.rootpath.parent
|
|
|
|
completed = subprocess.run(
|
|
(sys.executable, "-E", "-S", "-c", child_program),
|
|
cwd=project_root,
|
|
env={},
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=4,
|
|
)
|
|
|
|
assert (completed.returncode, completed.stderr) == (0, "")
|
|
assert json.loads(completed.stdout) == {
|
|
"error": "Model2Vec model minishlab/potion-retrieval-32M is not cached under /testbed/model2vec.",
|
|
}
|
|
|
|
|
|
def test_model2vec_similarity_loading_and_encoding_failures_return_false(caplog) -> None:
|
|
"""Unexpected model-path failures are recorded by the public fail-closed boundary."""
|
|
|
|
class _BrokenCheck(Model2VecSimilarityCheck):
|
|
"""Raise at the owned semantic step without creating an optional model client."""
|
|
|
|
def _matches(self, request_text: str, beats: tuple[str, ...]) -> bool:
|
|
"""Represent a deterministic model loading failure."""
|
|
|
|
del request_text, beats
|
|
raise RuntimeError("load failed")
|
|
|
|
check = _BrokenCheck(
|
|
changes=[],
|
|
request_text="repair widget",
|
|
searchable_beats=["Widget"],
|
|
cache_dir=Path("/testbed/model2vec"),
|
|
)
|
|
|
|
assert check() is False
|
|
assert check.error.startswith("RuntimeError:")
|
|
assert "Model2VecSimilarityCheck failed closed" in caplog.text
|
|
|
|
|
|
def test_model2vec_similarity_invalid_vectors_raise_normalized_errors() -> None:
|
|
"""Malformed request and candidate vectors cannot enter cosine scoring."""
|
|
|
|
check = _check()
|
|
for vector in ([], [0], ["bad", 0], [math.nan, 0], [math.inf, 0]):
|
|
with pytest.raises((TypeError, ValueError)):
|
|
check._validated_vector(vector, context="candidate")
|
|
with pytest.raises(ValueError, match="dimension mismatch"):
|
|
check._cosine_similarity((1.0, 0.0), (1.0, 0.0, 0.0))
|
|
|
|
|
|
def test_model2vec_similarity_candidate_vector_count_is_explicit() -> None:
|
|
"""Candidate encoding must retain one vector for each candidate beat."""
|
|
|
|
check = _check(searchable_beats=["bad", "good"])
|
|
|
|
assert check._candidate_vectors(((0.0, 0.0), (1.0, 0.0)), 2) == (
|
|
(0.0, 0.0),
|
|
(1.0, 0.0),
|
|
)
|
|
with pytest.raises(ValueError, match="did not match beats"):
|
|
check._candidate_vectors(((1.0, 0.0),), 2)
|
|
|
|
|
|
def test_model2vec_similarity_exceptions_log_and_record_false(caplog) -> None:
|
|
"""Verify unexpected semantic failures are logged and recorded as false."""
|
|
|
|
class _BrokenCheck(Model2VecSimilarityCheck):
|
|
"""Raise from the owned semantic matching operation."""
|
|
|
|
def _matches(self, request_text: str, beats: tuple[str, ...]) -> bool:
|
|
"""Raise a deterministic semantic error."""
|
|
|
|
del request_text, beats
|
|
raise ValueError("broken")
|
|
|
|
check = _BrokenCheck(
|
|
changes=[],
|
|
request_text="x",
|
|
searchable_beats=["y"],
|
|
cache_dir=Path("/testbed/model2vec"),
|
|
)
|
|
|
|
assert check() is False
|
|
assert check.changes == [(check, False)]
|
|
assert "Model2VecSimilarityCheck failed closed" in caplog.text
|
|
|
|
|
|
def test_request_chunker_preserves_short_prompt_and_splits_headings() -> None:
|
|
"""Verify short prompts remain intact and structured prompts retain headings."""
|
|
|
|
chunker = RequestChunker(token_budget=100)
|
|
assert chunker.chunk("repair band_aider/context/class_meta.py") == (
|
|
"repair band_aider/context/class_meta.py",
|
|
)
|
|
chunks = chunker.chunk(
|
|
"## Memento\n\nUpdate rollback.\n\n## ClassMeta API\n\nUpdate ClassMeta.validate_commit."
|
|
)
|
|
assert chunks == (
|
|
"## Memento\n\nUpdate rollback.",
|
|
"## ClassMeta API\n\nUpdate ClassMeta.validate_commit.",
|
|
)
|
|
|
|
|
|
def test_request_chunker_repeats_heading_for_oversized_sections() -> None:
|
|
"""Verify oversized sections split at paragraph boundaries with heading context."""
|
|
|
|
chunks = RequestChunker(token_budget=26).chunk(
|
|
"## ClassMeta API\n\nFirst paragraph names ClassMeta.\n\nSecond paragraph names validate_commit."
|
|
)
|
|
assert len(chunks) == 2
|
|
assert all(chunk.startswith("## ClassMeta API\n\n") for chunk in chunks)
|
|
assert "ClassMeta" in chunks[0]
|
|
assert "validate_commit" in chunks[1]
|
|
|
|
|
|
def test_request_chunker_reaches_minimum_greedily_without_backtracking() -> None:
|
|
"""Verify the first chunk closes at the first boundary reaching the minimum."""
|
|
|
|
chunker = RequestChunker(token_budget=26)
|
|
chunks = chunker.chunk(
|
|
"## API\n\nFirst paragraph is deliberately substantial.\n\nSecond paragraph is also substantial.\n\nThird paragraph is separate."
|
|
)
|
|
assert chunks[0].endswith("First paragraph is deliberately substantial.")
|
|
assert chunks[1].endswith("Second paragraph is also substantial.")
|
|
|
|
|
|
def test_request_chunker_deduplicates_repeated_request_chunks() -> None:
|
|
"""Request chunk preparation retains the first occurrence of repeated evidence."""
|
|
|
|
assert RequestChunker()._deduplicate(("memento", "classmeta", "memento")) == (
|
|
"memento",
|
|
"classmeta",
|
|
)
|