315 lines
16 KiB
Python
315 lines
16 KiB
Python
"""Final-exam observations become independently routed method obligations."""
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
from dataclasses import replace
|
|
import pickle
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.beta_workflow_judgement import BetaWorkflowJudgementService
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.phase_start_record import phase_start_from_record, phase_start_record
|
|
from band_aider.context.working_backlog import (
|
|
BackLog, BacklogLane, FailedQAItem, QABacklog, StagedInboundItem,
|
|
TestingBacklog as BetaWork,
|
|
)
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.pytest_runner_strategy import PytestRunnerStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult
|
|
from tests.band_aider.test_beta_prompt_refinement_boundary import (
|
|
_BetaProbe, _MemoryFileManager, _QueuedClient, _RecordingEditor, _RoutingCheck,
|
|
_TestRunner,
|
|
)
|
|
from tests.support.completion_clients import ScriptedOpenAICompatibleClient
|
|
|
|
|
|
class _ExamRunner(PytestRunnerStrategy):
|
|
"""Discover actual AST methods and record isolated, scripted executions."""
|
|
|
|
def __init__(self, manager: _MemoryFileManager, results: list[TestRunResult]) -> None:
|
|
"""Bind actual method discovery to a finite script of individual executions."""
|
|
super().__init__(file_manager=manager, repo_exec=manager.repo_exec)
|
|
self.results = iter(results)
|
|
self.calls: list[tuple[str, tuple[str, ...]]] = []
|
|
|
|
def run_tests_for(self, test_files, max_failures=None, test_names=()) -> TestRunResult:
|
|
"""Observe the exact selected method without executing a repository process."""
|
|
assert len(test_files) == 1
|
|
assert len(test_names) == 1
|
|
self.calls.append((test_files[0].location, tuple(test_names)))
|
|
return replace(next(self.results), test_name=test_names[0])
|
|
|
|
|
|
@pytest.fixture
|
|
def exam(monkeypatch):
|
|
"""Bind real discovery, staging and queues around controlled execution/judgment."""
|
|
manager = _MemoryFileManager(
|
|
client=_QueuedClient(),
|
|
sources={"src/value.py": "def value():\n return 1\n"},
|
|
tests={"tests/test_value.py": (
|
|
"from src.value import value\n\n"
|
|
"def test_first():\n assert value() == 2\n\n"
|
|
"def test_second():\n assert value() == 3\n\n"
|
|
"def test_pass():\n assert value() == 1\n"
|
|
)},
|
|
)
|
|
service = Mock(spec=BetaWorkflowJudgementService)
|
|
service.route_targeted_failure.return_value = _RoutingCheck(outcome="test")
|
|
backlog = BackLog(completion={}, judgement_service=service)
|
|
context = BetaContext(
|
|
goal="Repair value", classes=[], file_manager=manager,
|
|
backlog=QABacklog(backlog, manager.client, manager, Mock(spec=CodeParser)),
|
|
)
|
|
context.move_inbound(StagedInboundItem(
|
|
BacklogLane.Testing, BetaWork(None, "last exam", last_exam=True), "exam",
|
|
))
|
|
monkeypatch.setattr(BetaContext, "persist", Mock())
|
|
return manager, service, backlog, context
|
|
|
|
|
|
def test_matching_uses_current_ast_and_preserves_parameter_identity() -> None:
|
|
"""Reported cases intersect real callables without prefix or stale-name matches."""
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": (
|
|
"def test_short(): pass\n"
|
|
"def test_shorter(): pass\n"
|
|
"class TestA:\n def test_shared(self): pass\n"
|
|
"class TestB:\n def test_shared(self): pass\n"
|
|
),
|
|
"tests/test_b.py": "def test_shorter(): pass\n",
|
|
})
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
output = (
|
|
"FAILED tests/test_a.py::test_shorter[param.with spaces] - AssertionError\n"
|
|
r"FAILED tests\test_a.py::test_shorter[path\with::colons] - AssertionError" "\n"
|
|
"FAILED tests/test_a.py::TestA::test_shared - AssertionError\n"
|
|
"FAILED tests/test_a.py::test_removed - AssertionError\n"
|
|
)
|
|
|
|
matches = runner.matching_tests(manager.test_classes, TestRunResult("suite", TestRunOutcome.Failure, 1, output))
|
|
|
|
assert matches == {"tests/test_a.py": (
|
|
"test_shorter[param.with spaces]", r"test_shorter[path\with::colons]", "TestA.test_shared",
|
|
)}
|
|
manager.repo_exec.write_text(manager.test_path("tests/test_a.py"), "def test_new(): pass\n")
|
|
assert runner.matching_tests(manager.test_classes, TestRunResult("suite", TestRunOutcome.Failure, 1, output)) == {}
|
|
|
|
|
|
def test_matching_long_tracebacks_and_ambiguous_bare_names() -> None:
|
|
"""Long-format headers and unqualified names select current AST candidates."""
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": "def test_same(): pass\ndef test_other(): pass\n",
|
|
"tests/test_b.py": "def test_same(): pass\n",
|
|
})
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
result = TestRunResult("suite", TestRunOutcome.Failure, 1, "___ test_same[x.y] ___\nassert False\ntests/test_a.py:1: AssertionError\n")
|
|
|
|
assert runner.matching_tests(manager.test_classes, result) == {
|
|
"tests/test_a.py": ("test_same[x.y]",),
|
|
}
|
|
|
|
|
|
def test_matching_setup_error_alongside_runtime_failure() -> None:
|
|
"""Long-format setup errors remain candidates when ordinary failures also exist."""
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": "def test_setup(): pass\ndef test_runtime(): pass\n",
|
|
})
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
output = (
|
|
"___ ERROR at setup of test_setup[alpha beta] ___\nfixture failed\ntests/test_a.py:1: ValueError\n"
|
|
"___ test_runtime ___\nassert False\ntests/test_a.py:2: AssertionError\n"
|
|
)
|
|
|
|
assert runner.matching_tests(manager.test_classes, TestRunResult("suite", TestRunOutcome.Failure, 1, output)) == {
|
|
"tests/test_a.py": ("test_setup[alpha beta]", "test_runtime"),
|
|
}
|
|
|
|
|
|
def test_final_exam_executes_each_match_and_keeps_fresh_test_obligations(exam) -> None:
|
|
"""Two failures in one file survive staging; a fresh pass needs no repair."""
|
|
manager, service, backlog, context = exam
|
|
first = TestRunResult("first", TestRunOutcome.Failure, 1, "fresh first", ["tests/test_value.py:4", "AssertionError first"])
|
|
second = TestRunResult("second", TestRunOutcome.Failure, 1, "fresh second", ["tests/test_value.py:7", "AssertionError second"])
|
|
runner = _ExamRunner(manager, [first, second, TestRunResult("pass", TestRunOutcome.Success, 0, "passed")])
|
|
editor = _RecordingEditor("must not edit during discovery")
|
|
beta = _BetaProbe(context, runner=runner, editor=editor)
|
|
old = TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first\nFAILED tests/test_value.py::test_second\nFAILED tests/test_value.py::test_pass')
|
|
|
|
beta._report_all_tests_failure(old)
|
|
# Checkpoint recovery keeps each staged identity and its provenance lineage.
|
|
context.outbound_queue = pickle.loads(pickle.dumps(context.outbound_queue))
|
|
assert all(set(staged.lineage) == {"exam"} for staged in context.outbound_queue)
|
|
context.move_outbound_to_backlog()
|
|
|
|
assert runner.calls == [
|
|
("tests/test_value.py", ("test_first",)),
|
|
("tests/test_value.py", ("test_second",)),
|
|
("tests/test_value.py", ("test_pass",)),
|
|
]
|
|
assert editor.requests == []
|
|
assert [item.failure_test_name for item in backlog.testing] == ["test_first", "test_second"]
|
|
assert [item.failure_raw_output for item in backlog.testing] == ["fresh first", "fresh second"]
|
|
assert len(service.route_targeted_failure.call_args_list) == 2
|
|
for call in service.route_targeted_failure.call_args_list:
|
|
story = call.kwargs["story"]
|
|
assert story.failure_phase == "last_exam"
|
|
assert story.beta_edit_diff == story.beta_editor_summary == ""
|
|
assert story.raw_output != old.output
|
|
restored = [phase_start_from_record(phase_start_record(item)) for item in backlog.testing]
|
|
assert all(isinstance(item, BetaWork) for item in restored)
|
|
assert tuple(map(lambda item: item.failure_test_name if isinstance(item, BetaWork) else "", restored)) == (
|
|
"test_first", "test_second",
|
|
)
|
|
|
|
|
|
def test_final_exam_preserves_two_source_repairs_for_one_source(exam, monkeypatch) -> None:
|
|
"""Source ownership cannot stop discovery or collapse another method's repair."""
|
|
manager, service, backlog, context = exam
|
|
service.route_targeted_failure.return_value = _RoutingCheck(outcome="source")
|
|
source = manager.src_classes[0]
|
|
test = manager.test_classes[0]
|
|
first = TestRunResult("first", TestRunOutcome.Failure, 1, "fresh first", ["src/value.py:2"])
|
|
second = TestRunResult("second", TestRunOutcome.Failure, 1, "fresh second", ["src/value.py:2"])
|
|
repairs = [FailedQAItem(
|
|
source, "repair", downstream_classes=[source],
|
|
failed_test=FailingTest.from_result(test, replace(result, test_name=name)),
|
|
failure_test_name=name,
|
|
) for name, result in (("test_first", first), ("test_second", second))]
|
|
report = Mock(side_effect=[[repairs[0]], [repairs[1]]])
|
|
monkeypatch.setattr(QABacklog, "report_remediation", report)
|
|
runner = _ExamRunner(manager, [first, second])
|
|
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(
|
|
TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first\nFAILED tests/test_value.py::test_second'),
|
|
)
|
|
context.outbound_queue = pickle.loads(pickle.dumps(context.outbound_queue))
|
|
context.move_outbound_to_backlog()
|
|
|
|
assert [item.failed_test.test_name for item in backlog.remediation] == ["test_first", "test_second"]
|
|
assert all(item.focused_spotlight.location == source.location for item in backlog.remediation)
|
|
assert len(runner.calls) == 2
|
|
assert [call.kwargs["failed_test"].test_name for call in report.call_args_list] == ["test_first", "test_second"]
|
|
assert all(call.kwargs["failure_phase"] == "last_exam" for call in report.call_args_list)
|
|
|
|
|
|
def test_beta_post_edit_runs_the_whole_file_to_its_first_failure(exam) -> None:
|
|
"""The compiler's named obligation does not narrow Beta post-edit validation."""
|
|
manager, _service, _backlog, context = exam
|
|
runner = _TestRunner([TestRunResult("whole file", TestRunOutcome.Success, 0, "passed")])
|
|
item = BetaWork(manager.test_classes[0], "repair second", failure_test_name="test_second")
|
|
|
|
assert _BetaProbe(context, runner=runner)._run_targeted_tests(item).passed
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ())]
|
|
|
|
|
|
@pytest.mark.parametrize("code", (2, 5))
|
|
def test_named_collection_failure_reaches_collection_routing(exam, code) -> None:
|
|
"""An unresolved selector or collection error is not a passing execution."""
|
|
manager, service, backlog, context = exam
|
|
service.route_collection_failure.return_value = _RoutingCheck(outcome="test")
|
|
runner = _ExamRunner(manager, [TestRunResult("first", TestRunOutcome.Error, code, "no tests ran")])
|
|
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(
|
|
TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first'),
|
|
)
|
|
context.move_outbound_to_backlog()
|
|
|
|
assert len(backlog.testing) == 1
|
|
assert backlog.testing[0].failure_test_name == ""
|
|
assert backlog.testing[0].failure_outcome is TestRunOutcome.Error
|
|
service.route_collection_failure.assert_called_once()
|
|
service.route_targeted_failure.assert_not_called()
|
|
|
|
|
|
def test_duplicate_named_work_is_deduplicated_without_losing_siblings(exam) -> None:
|
|
"""The same method deduplicates; a different method remains executable."""
|
|
manager, _service, backlog, _context = exam
|
|
test = manager.test_classes[0]
|
|
first = BetaWork(test, "first", downstream_classes=manager.src_classes, failure_test_name="test_first")
|
|
second = BetaWork(test, "second", downstream_classes=manager.src_classes, failure_test_name="test_second")
|
|
|
|
assert backlog.queue_testing(first) is first
|
|
assert backlog.queue_testing(second) is second
|
|
assert backlog.queue_testing(replace(first)) is None
|
|
assert backlog.testing == (first, second)
|
|
|
|
|
|
def test_final_exam_invalid_routing_retries_until_one_coherent_answer(exam) -> None:
|
|
"""Repeated invalid output cannot erase the method awaiting owner assignment."""
|
|
manager, _service, backlog, context = exam
|
|
invalid = '{"outcome":"test","reason":"repair assertion","additional":{}}'
|
|
valid = ('{"outcome":"test","reason":"repair assertion",'
|
|
'"additional":{"test_work_needed":"Assert value equals one."}}')
|
|
client = ScriptedOpenAICompatibleClient((invalid,) * 12 + (valid,))
|
|
manager.client = client
|
|
backlog.judgement_service = BetaWorkflowJudgementService()
|
|
runner = _ExamRunner(manager, [TestRunResult(
|
|
"first", TestRunOutcome.Failure, 1, "fresh individual assertion", ["tests/test_value.py:4", "src/value.py:2"],
|
|
)])
|
|
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first'))
|
|
context.move_outbound_to_backlog()
|
|
|
|
assert len(client.prompts) == 13
|
|
assert "additional: missing member test_work_needed." in client.prompts[-1]
|
|
assert "fresh individual assertion" in client.prompts[-1]
|
|
assert [item.failure_test_name for item in backlog.testing] == ["test_first"]
|
|
|
|
|
|
def test_final_exam_environment_failure_escapes_without_a_pass(exam, monkeypatch) -> None:
|
|
"""An unusable test process cannot furnish a successful ownership decision."""
|
|
manager, service, backlog, context = exam
|
|
runner = _ExamRunner(manager, [])
|
|
monkeypatch.setattr(runner, "run_tests_for", Mock(side_effect=OSError("runner unavailable")))
|
|
|
|
with pytest.raises(OSError, match="runner unavailable"):
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first'))
|
|
|
|
service.route_targeted_failure.assert_not_called()
|
|
assert backlog.testing == ()
|
|
|
|
|
|
def test_final_exam_timeout_routes_to_error_judgment_with_invocation_evidence(exam) -> None:
|
|
"""A timed-out individual rerun retains its evidence without demanding a frame."""
|
|
manager, service, backlog, context = exam
|
|
service.route_collection_failure.return_value = _RoutingCheck(outcome="test")
|
|
run_info = 'ARGV: ["python", "-m", "pytest", "tests/test_value.py::test_first"]\nPROCESS DEADLINE SECONDS: 60'
|
|
runner = _ExamRunner(manager, [TestRunResult("first", TestRunOutcome.Timeout, 124,
|
|
"TIMEOUT: pytest exceeded 60 seconds", run_info=run_info)])
|
|
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(TestRunResult("suite", TestRunOutcome.Failure, 1, 'FAILED tests/test_value.py::test_first'))
|
|
context.move_outbound_to_backlog()
|
|
|
|
service.route_targeted_failure.assert_not_called()
|
|
service.route_collection_failure.assert_called_once()
|
|
[item] = backlog.testing
|
|
assert item.failure_outcome is TestRunOutcome.Timeout
|
|
assert item.failure_run_info == run_info
|
|
assert item.failure_test_name == ""
|
|
assert service.route_collection_failure.call_args.kwargs["story"].failing.run_info == run_info
|
|
|
|
|
|
def test_final_exam_collection_error_routes_raw_file_without_method_execution(exam):
|
|
manager, service, backlog, context = exam
|
|
service.route_collection_failure.return_value = _RoutingCheck(outcome="test")
|
|
runner = _ExamRunner(manager, [])
|
|
output = "___ ERROR collecting tests/test_value.py ___\nImportError: missing value\n"
|
|
|
|
_BetaProbe(context, runner=runner)._report_all_tests_failure(
|
|
TestRunResult("suite", TestRunOutcome.Error, 2, output, list(output.splitlines())),
|
|
)
|
|
context.move_outbound_to_backlog()
|
|
|
|
assert runner.calls == []
|
|
service.route_collection_failure.assert_called_once()
|
|
service.route_targeted_failure.assert_not_called()
|
|
[item] = backlog.testing
|
|
assert item.failure_test_name == ""
|
|
assert item.failure_outcome is TestRunOutcome.Error
|
|
assert item.failure_raw_output == output
|
|
restored = phase_start_from_record(phase_start_record(item))
|
|
assert isinstance(restored, BetaWork)
|
|
assert restored.failure_outcome is TestRunOutcome.Error
|
|
assert restored.failure_raw_output == output
|