475 lines
20 KiB
Python
475 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import cast
|
|
import pytest
|
|
from unittest.mock import patch
|
|
from pathlib import Path
|
|
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult
|
|
from band_aider.context.targeted_test_collection_failure_owner_check import (
|
|
TargetedCollectionAlphaOrBetaRoutingDecision,
|
|
)
|
|
from band_aider.context.working_backlog import (
|
|
ChangeType,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
TestingBacklog as BacklogItem,
|
|
)
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from tests.support.completion_clients import (
|
|
ScriptedCompletionClient,
|
|
ScriptedOpenAICompatibleClient,
|
|
StaticOpenAICompatibleClient,
|
|
)
|
|
from tests.support.memory_path import MemoryPath
|
|
|
|
|
|
class _CollectionFileManager(FileManager):
|
|
"""Serve collection-routing evidence from one nominal in-memory repository."""
|
|
|
|
def __init__(
|
|
self,
|
|
repo_root: MemoryPath,
|
|
client: ScriptedOpenAICompatibleClient | StaticOpenAICompatibleClient,
|
|
) -> None:
|
|
"""Bind source and test paths, prompt formatting, and client state to memory."""
|
|
|
|
self._pristine_sources = {}
|
|
|
|
self.repo_root = Path(str(repo_root))
|
|
self.src_root = self.repo_root / "src"
|
|
self.test_root = self.repo_root / "tests"
|
|
self.repo_exec = repo_root.repository
|
|
self.client = client
|
|
self.prompt = "Keep the public contract described by the original task."
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.sources: dict[str, ClassMeta] = {}
|
|
self.tests: dict[str, ClassMeta] = {}
|
|
|
|
def repo_path(self, location: str | Path) -> Path:
|
|
"""Resolve one repository path lexically inside the in-memory root."""
|
|
|
|
path = Path(location)
|
|
return path if path.is_absolute() else self.repo_root / path
|
|
|
|
def repo_location(self, location: str | Path) -> str:
|
|
"""Return one normalized repository-relative location."""
|
|
|
|
return self.repo_path(location).relative_to(self.repo_root).as_posix()
|
|
|
|
def source_path(self, location: str | Path) -> Path:
|
|
"""Return the source path owned by this isolated repository."""
|
|
|
|
return self.repo_path(location)
|
|
|
|
def test_path(self, location: str | Path) -> Path:
|
|
"""Return the test path owned by this isolated repository."""
|
|
|
|
return self.repo_path(location)
|
|
|
|
def git_diff_for(self, location: str | Path) -> str:
|
|
"""Return no Git evidence because routing fixtures use exact source only."""
|
|
|
|
del location
|
|
return ""
|
|
|
|
def current_src(self, location: str) -> ClassMeta | None:
|
|
"""Return the explicitly registered current source metadata."""
|
|
|
|
return self.sources.get(location)
|
|
|
|
def current_test(self, location: str) -> ClassMeta | None:
|
|
"""Return the explicitly registered current test metadata."""
|
|
|
|
return self.tests.get(location)
|
|
|
|
|
|
def _source(location: str, content_type: ContentType = ContentType.Source) -> ClassMeta:
|
|
"""Return explicit source or test metadata for one routing fixture location."""
|
|
|
|
return ClassMeta(location, f"summary for {location}", [], {}, content_type=content_type)
|
|
|
|
|
|
def _write(tmp_path: MemoryPath, metadata: ClassMeta, content: str) -> None:
|
|
"""Store exact collection evidence in the fixture's in-memory repository."""
|
|
|
|
path = tmp_path / metadata.location
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(content)
|
|
|
|
|
|
def _routing_check(
|
|
tmp_path: MemoryPath,
|
|
responses: tuple[dict[str, object], ...],
|
|
trace: tuple[str, ...] = ("tests/test_example.py:2", "src/example.py:2"),
|
|
test_content: str = "def test_good():\n assert False\n\ndef test_other():\n assert False\n",
|
|
run_info: str = "",
|
|
) -> tuple[TargetedCollectionAlphaOrBetaRoutingDecision, ScriptedCompletionClient]:
|
|
"""Build one collection routing decision with a finite stationary response script."""
|
|
|
|
test = _source("tests/test_example.py", ContentType.Test)
|
|
source = _source("src/example.py")
|
|
_write(tmp_path, test, test_content)
|
|
_write(tmp_path, source, "def current_source():\n return 'current'\n")
|
|
manager = _CollectionFileManager(
|
|
tmp_path,
|
|
StaticOpenAICompatibleClient('{"cdc": ""}'),
|
|
)
|
|
manager.sources[source.location] = source
|
|
manager.tests[test.location] = test
|
|
manager.src_classes = [source]
|
|
manager.test_classes = [test]
|
|
item = BacklogItem(
|
|
focused_spotlight=test,
|
|
summary="test task",
|
|
original_cot=manager.prompt,
|
|
change_type=ChangeType.Update,
|
|
)
|
|
failure = FailingTest(
|
|
test=test,
|
|
error_message="collection failed",
|
|
stack_trace=list(trace),
|
|
raw_output="raw",
|
|
run_info=run_info,
|
|
)
|
|
client = ScriptedCompletionClient(tuple(map(json.dumps, responses)))
|
|
decision = TargetedCollectionAlphaOrBetaRoutingDecision(
|
|
client=client,
|
|
file_manager=manager,
|
|
request=AiderEditRequest(
|
|
authoritative_prompt="",
|
|
editable_files=[Path(test.location)],
|
|
readonly_files=[Path(source.location)],
|
|
),
|
|
story=TargetedFailureStory(
|
|
testing_item=item,
|
|
raw_output=failure.raw_output,
|
|
work_needed="repair the collection failure",
|
|
candidate_sources=[source, test],
|
|
failing=failure,
|
|
original_cot=item.original_cot,
|
|
failure_phase="collection",
|
|
),
|
|
)
|
|
return decision, client
|
|
|
|
|
|
def test_collection_routing_passes_static_calls_without_retained_receipts(tmp_path: MemoryPath) -> None:
|
|
"""The renderer receives admitted call evidence while fresh failure ownership stays local."""
|
|
with patch.object(ContextualFileRenderer, "build", wraps=ContextualFileRenderer.build) as presenter:
|
|
decision, _client = _routing_check(tmp_path, (), trace=("tests/test_example.py:3: in test_good",),
|
|
test_content="from example import current_source\ndef test_good():\n current_source()\n assert False\n")
|
|
|
|
assert "current_source" in presenter.call_args.kwargs["call_trace"]
|
|
assert "def test_good():" in decision.prompt
|
|
assert "VALIDATION RECEIPTS" not in decision.prompt
|
|
|
|
|
|
def test_timeout_routing_checks_unit_isolation_and_external_dependencies(tmp_path: MemoryPath) -> None:
|
|
"""The routing request checks resource isolation before attributing a repeating call."""
|
|
|
|
invocation = "Run test_client alone; unit fixture contacts a remote service and waits for localhost:8123."
|
|
decision, _client = _routing_check(tmp_path, (), run_info=invocation)
|
|
|
|
assert invocation in decision.prompt
|
|
assert "A unit test must run in isolation" in decision.prompt
|
|
assert "internet access, remote services, local services, external processes, and open ports" in decision.prompt
|
|
assert "external to the unit even when hosted on the same machine" in decision.prompt
|
|
assert "Identifying a loop or slow call is necessary but insufficient" in decision.prompt
|
|
assert "A dependency-related timeout alone does not select Alpha or Beta" in decision.prompt
|
|
assert "Do not treat a genuine integration test as a unit test" in decision.prompt
|
|
|
|
|
|
def _empty_import_error() -> dict[str, str]:
|
|
"""Return the complete empty import-error evidence shape required by routing."""
|
|
|
|
return {
|
|
"missing_symbol": "",
|
|
"imported_from_module": "",
|
|
"importer_location": "",
|
|
"import_statement": "",
|
|
"preferred_repair_location": "",
|
|
"repair_policy": "",
|
|
}
|
|
|
|
|
|
def _test_response(assignments: list[dict[str, object]]) -> dict[str, object]:
|
|
"""Return one otherwise-valid test-owned collection-routing response."""
|
|
|
|
return {
|
|
"outcome": "test",
|
|
"reason": "The collection setup remains test-owned.",
|
|
"additional": {
|
|
"failure_phase": "collection",
|
|
"failed_test": "tests/test_example.py",
|
|
"error_message": "collection failed",
|
|
"relevant_stack_trace": ["trace"],
|
|
"test_work_needed": "Repair the collection setup.",
|
|
"source_clues": [],
|
|
"assignments": assignments,
|
|
"import_chain": [],
|
|
"import_error": _empty_import_error(),
|
|
},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("line", [1, 3])
|
|
def test_collection_expands_only_an_identified_callable(tmp_path: MemoryPath, line: int) -> None:
|
|
"""Module-level collection failures do not authorize a whole-module scan."""
|
|
target = _source("tests/test_example.py", ContentType.Test)
|
|
source = _source("src/example.py")
|
|
_write(tmp_path, target, "from example import run\ndef test_example():\n run()\n")
|
|
_write(tmp_path, source, "def run():\n return False\n")
|
|
manager = _CollectionFileManager(tmp_path, StaticOpenAICompatibleClient('{"cdc": ""}'))
|
|
manager.src_classes = [source]
|
|
manager.test_classes = [target]
|
|
manager.sources[source.location] = source
|
|
manager.tests[target.location] = target
|
|
# This case tests fresh discovery without an earlier backlog source admission.
|
|
item = BacklogItem(focused_spotlight=target, summary="repair", change_type=ChangeType.Update)
|
|
failure = FailingTest(test=target, error_message="collection failed",
|
|
stack_trace=[f"tests/test_example.py:{line}: failure"], raw_output="exact failure")
|
|
|
|
decision = TargetedCollectionAlphaOrBetaRoutingDecision(client=ScriptedCompletionClient(()),
|
|
file_manager=manager, request=AiderEditRequest(authoritative_prompt="", editable_files=[Path(target.location)], readonly_files=[Path(source.location)]),
|
|
story=TargetedFailureStory(testing_item=item, raw_output=failure.raw_output, work_needed="repair",
|
|
candidate_sources=[target, source], failing=failure, failure_phase="collection"))
|
|
|
|
assert (source.location in decision.candidate_source_locations) is (line == 3)
|
|
assert ('"caller_name": "test_example"' in decision.prompt) is (line == 3)
|
|
assert failure.stack_trace == [f"tests/test_example.py:{line}: failure"]
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", ("source", "test"))
|
|
def test_collection_routing_preserves_selected_owner_with_plural_assignments(
|
|
tmp_path: MemoryPath,
|
|
outcome: str,
|
|
) -> None:
|
|
"""Concurrent assignment evidence cannot override either selected causal owner."""
|
|
|
|
response = _test_response([
|
|
{
|
|
"outcome": "test",
|
|
"reason": "test_good violates the original contract.",
|
|
"relevant_stack_trace": ["test_good trace"],
|
|
"test_work_needed": "Repair the assertion.",
|
|
"bad_test": {
|
|
"test_location": "tests/test_example.py",
|
|
"search_block": "def test_good():\n assert False",
|
|
},
|
|
"source_clues": [],
|
|
},
|
|
{
|
|
"outcome": "source",
|
|
"reason": "The source import is stale.",
|
|
"relevant_stack_trace": ["source trace"],
|
|
"test_work_needed": "",
|
|
"bad_test": {"test_location": "", "search_block": ""},
|
|
"source_clues": [{"location": "src/example.py", "reason": "stale import"}],
|
|
},
|
|
])
|
|
response["outcome"] = outcome
|
|
additional = cast(dict[str, object], response["additional"])
|
|
additional["test_work_needed"] = "Repair the assertion." if outcome == "test" else ""
|
|
decision, _client = _routing_check(tmp_path, (response,))
|
|
|
|
assert "ROUTING QUESTION: Is the observed divergence in SOURCE behavior" in decision.prompt
|
|
assert "If source ownership is not established, return test" in decision.prompt
|
|
assert "AUTHORITATIVE TASK > COMPATIBLE GUIDANCE > APPLICABLE PRE-EXISTING PRECEDENTS" in decision.prompt
|
|
assert "IF COLLECTION FAILURE" in decision.prompt
|
|
assert "IF TIMEOUT" in decision.prompt
|
|
assert "IF ANOTHER EXECUTION ERROR" in decision.prompt
|
|
assert "Identify the loop or call responsible for the timeout" in decision.prompt
|
|
assert "correctly rejects invalid test input" in decision.prompt
|
|
assert "production invalidates otherwise valid test input" in decision.prompt
|
|
assert "Do not prefer Alpha or Beta" in decision.prompt
|
|
assert "before any test body executes" not in decision.prompt
|
|
assert decision() is (outcome == "source")
|
|
assert decision.outcome == outcome
|
|
assert len(decision.assignments) == 2
|
|
assert decision.assignments[0].outcome == "test"
|
|
assert decision.assignments[0].source_clues[0]["location"] == "tests/test_example.py"
|
|
assert decision.assignments[1].outcome == "source"
|
|
assert decision.additional["assignments"][0]["reason"] == "test_good violates the original contract."
|
|
|
|
|
|
def test_collection_routing_retries_owner_without_supporting_assignment(tmp_path: MemoryPath) -> None:
|
|
"""An overall source verdict requires a source assignment when assignments are supplied."""
|
|
|
|
test_assignment = {
|
|
"outcome": "test",
|
|
"reason": "The test still requires the retired contract.",
|
|
"relevant_stack_trace": [],
|
|
"test_work_needed": "Migrate the obsolete expectation.",
|
|
"bad_test": {
|
|
"test_location": "tests/test_example.py",
|
|
"search_block": "def test_good():\n assert False",
|
|
},
|
|
"source_clues": [],
|
|
}
|
|
rejected = _test_response([test_assignment])
|
|
rejected["outcome"] = "source"
|
|
cast(dict[str, object], rejected["additional"])["test_work_needed"] = ""
|
|
decision, client = _routing_check(tmp_path, (rejected, _test_response([test_assignment])))
|
|
|
|
assert decision() is False
|
|
assert decision.outcome == "test"
|
|
assert len(client.prompts) == 2
|
|
assert "outcome: must match an attributed assignment" in client.prompts[1]
|
|
|
|
|
|
def test_collection_routing_retries_nonbinary_output_with_original_prompt(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""A nonbinary response receives feedback while the canonical prompt remains stable."""
|
|
|
|
invalid = _test_response([])
|
|
invalid["outcome"] = "indeterminate"
|
|
invalid["reason"] = "The response did not establish a source owner."
|
|
decision, client = _routing_check(tmp_path, (invalid, _test_response([])))
|
|
first_prompt = decision.prompt
|
|
|
|
assert decision() is False
|
|
assert decision.outcome == "test"
|
|
assert decision.assignments == ()
|
|
assert decision.additional["import_error"]["repair_policy"] == ""
|
|
assert decision.prompt == first_prompt
|
|
assert client.prompts[0] == first_prompt
|
|
assert client.prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert "outcome: must be source or test" in client.prompts[1]
|
|
|
|
|
|
def test_collection_routing_retries_invalid_excerpt_without_rewriting_original_prompt(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Retry feedback identifies an invalid test excerpt while preserving full evidence."""
|
|
|
|
invalid = _test_response([
|
|
{
|
|
"outcome": "test",
|
|
"reason": "The first test is invalid.",
|
|
"bad_test": {"test_location": "tests/test_example.py", "search_block": "def test_good():\n assert False"},
|
|
"relevant_stack_trace": ["good trace"],
|
|
"test_work_needed": "Repair the first test.",
|
|
"source_clues": [],
|
|
},
|
|
{
|
|
"outcome": "test",
|
|
"reason": "The second attribution needs correction.",
|
|
"bad_test": {"test_location": "tests/test_example.py", "search_block": "not present"},
|
|
"relevant_stack_trace": ["other trace"],
|
|
"test_work_needed": "Repair the second test.",
|
|
"source_clues": [],
|
|
},
|
|
])
|
|
corrected = _test_response([
|
|
{
|
|
"outcome": "test",
|
|
"reason": "test_other is invalid.",
|
|
"bad_test": {"test_location": "tests/test_example.py", "search_block": "def test_other():\n assert False"},
|
|
"relevant_stack_trace": ["other trace"],
|
|
"test_work_needed": "Repair the second test.",
|
|
"source_clues": [],
|
|
},
|
|
])
|
|
decision, client = _routing_check(tmp_path, (invalid, corrected))
|
|
first_prompt = decision.prompt
|
|
|
|
assert decision() is False
|
|
assert decision.prompt == first_prompt
|
|
assert len(client.prompts) == 2
|
|
assert client.prompts[0] == first_prompt
|
|
assert client.prompts[1].startswith(f"{first_prompt}\n\n== BEGIN RESPONSE VALIDATION ERROR ==\n")
|
|
assert first_prompt in client.prompts[1]
|
|
assert "not present" in client.prompts[1]
|
|
assert any(assignment.reason == "test_other is invalid." for assignment in decision.assignments)
|
|
|
|
|
|
def test_remediation_guidance_places_source_assignment_before_complete_diagnostics(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Source-owned remediation returns stationary grounded guidance from actual evidence."""
|
|
|
|
source = _source("src/example.py")
|
|
test = _source("tests/test_example.py", ContentType.Test)
|
|
_write(tmp_path, source, "def stale_import():\n return 'old'\n")
|
|
_write(tmp_path, test, "def test_example():\n assert False\n")
|
|
client = ScriptedOpenAICompatibleClient((json.dumps({
|
|
"work_needed": "1. Repair the stale source import.\n2. Verify the targeted failure.",
|
|
"reason": ["def stale_import():", "return 'old'"],
|
|
}),))
|
|
manager = _CollectionFileManager(tmp_path, client)
|
|
manager.sources[source.location] = source
|
|
manager.tests[test.location] = test
|
|
manager.src_classes = [source]
|
|
manager.test_classes = [test]
|
|
backlog = object.__new__(ContributorBacklog)
|
|
backlog._file_manager = manager
|
|
item = FailedQAItem(
|
|
summary="Fix the stale source import.",
|
|
focused_spotlight=source,
|
|
original_cot=manager.prompt,
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
error_message="complete error output",
|
|
stack_trace=["complete traceback"],
|
|
raw_output="complete raw output",
|
|
),
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="The current Alpha source task.",
|
|
editable_files=[Path(source.location)],
|
|
readonly_files=[],
|
|
)
|
|
|
|
guidance = backlog.remediation_work_needed_for(source, [item], request=request)
|
|
|
|
assert guidance == "1. Repair the stale source import.\n2. Verify the targeted failure."
|
|
assert "Assigned source remediation:" in client.prompts[0]
|
|
assert "Fix the stale source import." in client.prompts[0]
|
|
assert "complete error output" in client.prompts[0]
|
|
|
|
|
|
def test_collection_without_trace_source_routes_to_test_without_completion(tmp_path: MemoryPath) -> None:
|
|
"""Without prior admission or trace/call support, routing cannot authorize Alpha."""
|
|
|
|
decision, client = _routing_check(tmp_path, (), trace=("tests/test_example.py:2",))
|
|
|
|
assert decision() is False
|
|
assert decision.outcome == "test"
|
|
assert decision.additional["candidate_sources"] == ["tests/test_example.py"]
|
|
assert decision.assignments[0].outcome == "test"
|
|
assert decision.assignments[0].test_work_needed
|
|
assert client.prompts == []
|
|
assert decision.prompt == ""
|
|
|
|
|
|
def test_collection_rejects_unadmitted_repair_location(tmp_path: MemoryPath) -> None:
|
|
"""A plausible invented repair path is corrected before it can become remediation."""
|
|
|
|
invalid = _test_response([])
|
|
additional = invalid["additional"]
|
|
assert isinstance(additional, dict)
|
|
additional["source_clues"] = [{"location": "src/untraced.py", "reason": "looks relevant"}]
|
|
decision, client = _routing_check(tmp_path, (invalid, _test_response([])))
|
|
|
|
assert decision() is False
|
|
assert len(client.prompts) == 2
|
|
assert "repair locations must belong to the admitted candidates" in client.prompts[1]
|
|
|
|
|
|
def test_error_routing_prompt_preserves_execution_information(tmp_path: MemoryPath) -> None:
|
|
"""Run details reach the owner judgment outside stack-trace citation evidence."""
|
|
details = 'ARGV: ["python", "-m", "pytest", "tests/test_example.py"]\nTIMEOUT SECONDS: 60'
|
|
decision, _client = _routing_check(tmp_path, (), run_info=details)
|
|
|
|
assert "RUN INFORMATION" in decision.prompt
|
|
assert details in decision.prompt
|
|
assert "A timeout may have no traceback" in decision.prompt
|