331 lines
16 KiB
Python
331 lines
16 KiB
Python
"""Test-origin qualification precedes repair ownership and method selection."""
|
|
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from band_aider.judgement.pytest_runner_strategy import PytestRunnerStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome, TestRunResult
|
|
from tests.band_aider.test_beta_prompt_refinement_boundary import _MemoryFileManager
|
|
from tests.band_aider.test_pytest_runner_strategy import _strategy
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_runner_can_be_imported_before_language_factory():
|
|
"""Normal test startup must not hide an import cycle at the public runner."""
|
|
result = subprocess.run([sys.executable, "-B", "-c",
|
|
"from band_aider.judgement.pytest_runner_strategy import PytestRunnerStrategy"],
|
|
capture_output=True, text=True, check=False)
|
|
assert result.returncode == 0, result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("code,output,outcome", (
|
|
(0, "1 passed", TestRunOutcome.Success),
|
|
(1, "___ test_value ___\nAssertionError", TestRunOutcome.Failure),
|
|
(1, "___ ERROR at setup of test_value ___\nfixture failed", TestRunOutcome.Error),
|
|
(1, "___ ERROR at teardown of test_value ___\nfixture failed", TestRunOutcome.Error),
|
|
(2, "___ ERROR collecting tests/test_value.py ___\nImportError", TestRunOutcome.Error),
|
|
))
|
|
def test_runner_classifies_before_method_discovery(tmp_path, code, output, outcome):
|
|
runner, repository, manager = _strategy(tmp_path)
|
|
repository.queue_result(subprocess.CompletedProcess(["pytest"], code, output, ""))
|
|
|
|
result = runner._run([manager.test_root / "test_value.py"])
|
|
|
|
assert result.outcome is outcome
|
|
assert result.passed is (outcome is TestRunOutcome.Success)
|
|
|
|
|
|
def test_origin_intersects_actual_frames_before_selecting_first():
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_fixture.py": "def test_other(): pass\n",
|
|
"tests/test_origin.py": "def test_origin(): pass\n",
|
|
"tests/test_inner.py": "def test_inner(): pass\n",
|
|
})
|
|
output: str = (
|
|
"___ test_origin ___\n"
|
|
"original_test_index = {'tests/test_fixture.py': 0}\n"
|
|
'trace = "tests/test_fixture.py:1: in test_other"\n'
|
|
"tests/test_fixture.py:1: in helper\n"
|
|
"tests/test_origin.py:1: in test_origin\n"
|
|
"tests/test_inner.py:1: in test_inner\n"
|
|
"E AssertionError\n"
|
|
)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
reports = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Failure, 1, output, list(output.splitlines()),
|
|
))
|
|
|
|
assert [(report.test.location, report.test_name, report.outcome) for report in reports] == [
|
|
("tests/test_origin.py", "test_origin", TestRunOutcome.Failure),
|
|
]
|
|
assert reports[0].raw_output == output
|
|
|
|
|
|
def test_mixed_reports_keep_error_scope_and_failure_method():
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_broken.py": "this is invalid python !\n",
|
|
"tests/test_value.py": "def test_value(): pass\n",
|
|
})
|
|
output: str = (
|
|
"___ ERROR collecting tests/test_broken.py ___\nSyntaxError\n"
|
|
"___ test_value ___\ntests/test_value.py:1: AssertionError\n"
|
|
)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
reports = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Error, 1, output, list(output.splitlines()),
|
|
))
|
|
|
|
assert [(report.test.location, report.test_name, report.outcome) for report in reports] == [
|
|
("tests/test_broken.py", "", TestRunOutcome.Error),
|
|
("tests/test_value.py", "test_value", TestRunOutcome.Failure),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("case", ("", "[path::with.dots and spaces]"))
|
|
def test_setup_error_summary_identifies_test_file_without_an_executing_test_frame(case):
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": "def test_value(): pass\n",
|
|
"tests/test_b.py": "def test_value(): pass\n",
|
|
"tests/fixtures.py": "def fixture(): raise RuntimeError()\n",
|
|
})
|
|
output: str = (
|
|
f"___ ERROR at setup of test_value{case} ___\ntests/fixtures.py:1: in fixture\nE RuntimeError\n"
|
|
"================ short test summary info ================\n"
|
|
f"ERROR tests/test_b.py::test_value{case} - RuntimeError\n"
|
|
)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
[report] = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Error, 1, output, list(output.splitlines()),
|
|
))
|
|
|
|
assert report.test.location == "tests/test_b.py"
|
|
assert report.outcome is TestRunOutcome.Error
|
|
assert report.test_name == ""
|
|
assert report.raw_output == output
|
|
|
|
|
|
def test_same_named_error_reports_keep_distinct_collected_files():
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": "def test_value(): pass\n",
|
|
"tests/test_b.py": "def test_value(): pass\n",
|
|
})
|
|
output: str = (
|
|
"___ ERROR at setup of test_value ___\nRuntimeError: first fixture\n"
|
|
"___ ERROR at setup of test_value ___\nRuntimeError: second fixture\n"
|
|
"================ short test summary info ================\n"
|
|
"ERROR tests/test_a.py::test_value - RuntimeError\n"
|
|
"ERROR tests/test_b.py::test_value - RuntimeError\n"
|
|
)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
reports = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Error, 1, output, list(output.splitlines()),
|
|
))
|
|
|
|
assert [report.test.location for report in reports] == ["tests/test_a.py", "tests/test_b.py"]
|
|
assert all(report.outcome is TestRunOutcome.Error and report.test_name == "" for report in reports)
|
|
|
|
|
|
def test_unidentified_new_failure_is_not_removed_test_evidence():
|
|
manager = _MemoryFileManager(tests={"tests/test_value.py": "def test_current(): pass\n"})
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
output: str = "___ test_missing ___\ntests/test_value.py:1: AssertionError\n"
|
|
|
|
with pytest.raises(ValueError, match="identify"):
|
|
runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Failure, 1, output, list(output.splitlines()),
|
|
))
|
|
|
|
|
|
@pytest.mark.parametrize("incoming", (TestRunOutcome.Error, TestRunOutcome.Timeout))
|
|
@pytest.mark.parametrize("context", ("AlphaLoop", "BetaLoop"))
|
|
@pytest.mark.parametrize("outcome,code", (
|
|
(TestRunOutcome.Error, 1),
|
|
(TestRunOutcome.Timeout, 124),
|
|
(TestRunOutcome.Failure, 1),
|
|
(TestRunOutcome.Success, 5),
|
|
(TestRunOutcome.Success, 0),
|
|
))
|
|
def test_file_error_requires_successful_execution_even_without_changed_methods(memory_root, context, outcome, code, incoming):
|
|
"""Error and timeout obligations clear only after successful file execution."""
|
|
from tests.band_aider.test_python_compiler import _compiler, _TargetedRunner
|
|
from band_aider.strategies.compiler import TestRunCompileError
|
|
|
|
alpha = context == "AlphaLoop"
|
|
runner = _TargetedRunner([TestRunResult("pytest tests/test_value.py", outcome, code, "fresh evidence")])
|
|
compiler, _, manager = _compiler(
|
|
memory_root, sources={"value.py": "value = 1\n"},
|
|
tests={"test_value.py": "fixture_value = 1\ndef test_value(): pass\n"},
|
|
edit_location="src/value.py" if alpha else "tests/test_value.py",
|
|
replacement="value = 2\n" if alpha else "fixture_value = 2\ndef test_value(): pass\n",
|
|
context_name=context, runner=runner, is_remediation=alpha,
|
|
remediation_originator_location="tests/test_value.py" if alpha else None,
|
|
readonly_locations=("tests/test_value.py",) if alpha else (),
|
|
failure_outcome=incoming,
|
|
failure_error_message="setup failed", failure_raw_output="___ ERROR at setup of test_value ___\nfixture error",
|
|
)
|
|
|
|
errors = compiler.compile([memory_root / ("src/value.py" if alpha else "tests/test_value.py")])
|
|
|
|
assert compiler.edit_request.failure_outcome is incoming
|
|
assert runner.calls == [([manager.test_classes[0]], 1, ())]
|
|
if code == 0:
|
|
assert errors == []
|
|
else:
|
|
assert len(errors) == 1
|
|
assert isinstance(errors[0], TestRunCompileError)
|
|
assert errors[0].test_name == ""
|
|
assert errors[0].raw_output == "fresh evidence"
|
|
|
|
|
|
def test_file_receipt_cannot_clear_named_failure_or_accept_empty_collection():
|
|
from band_aider.context.validation_receipts import ValidationReceipt, ValidationReceipts
|
|
|
|
manager = _MemoryFileManager(tests={"tests/test_value.py": "def test_value(): pass\n"})
|
|
test = manager.test_classes[0]
|
|
result = TestRunResult("pytest tests/test_value.py", TestRunOutcome.Success, 0, "1 passed")
|
|
receipt = ValidationReceipt.from_result(test, result, test_definition="source", generation=1, file_scope=True)
|
|
receipts = ValidationReceipts((receipt,), generation=1)
|
|
|
|
assert receipts.current_for(test.location, "test_value") is None
|
|
assert receipts.current_for(test.location, "") is None
|
|
assert receipts.current_for(test.location, "", file_scope=True) is receipt
|
|
assert ValidationReceipts((receipt,), generation=2).current_for(test.location, "", file_scope=True) is None
|
|
empty = ValidationReceipt.from_result(test, TestRunResult("pytest", TestRunOutcome.Success, 5, "no tests"),
|
|
test_definition="", generation=1, file_scope=True)
|
|
assert not empty.passed
|
|
assert empty.failure is not None
|
|
assert empty.failure.outcome is TestRunOutcome.Error
|
|
|
|
|
|
def test_alpha_error_routing_uses_known_file_without_beta_item():
|
|
from band_aider.context.beta_workflow_judgement import BetaWorkflowJudgementService
|
|
from band_aider.context.beta_workflow_story import TargetedFailureStory
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.working_backlog import ChangeType
|
|
|
|
manager = _MemoryFileManager(tests={"tests/test_value.py": "def test_value(): pass\n"})
|
|
test = manager.test_classes[0]
|
|
output: str = "___ ERROR at teardown of test_value ___\nRuntimeError: cleanup\n"
|
|
failure = FailingTest.from_result(test, TestRunResult("pytest", TestRunOutcome.Error, 1, output, list(output.splitlines())))
|
|
story = TargetedFailureStory(
|
|
testing_item=None, focused_test=test, raw_output=output, work_needed="Repair fixture cleanup.",
|
|
candidate_sources=[], failing=failure, failure_phase="after_alpha", change_type=ChangeType.Update,
|
|
)
|
|
service = BetaWorkflowJudgementService()
|
|
|
|
decision = service.route_collection_failure(file_manager=manager, story=story)
|
|
|
|
assert decision.outcome == "test"
|
|
assert service.last_collection_evidence["candidate_test_location"] == test.location
|
|
assert service.last_collection_evidence["raw_output_present"]
|
|
|
|
|
|
def test_error_recheck_exports_current_file_receipt(memory_root):
|
|
from tests.band_aider.test_python_compiler import _compiler, _TargetedRunner
|
|
from band_aider.context.validation_receipts import ValidationReceipts
|
|
from band_aider.strategies.python.python_compiler import AlphaPythonCompiler
|
|
|
|
runner = _TargetedRunner([TestRunResult("pytest tests/test_value.py", TestRunOutcome.Success, 0, "1 passed")])
|
|
compiler, _, manager = _compiler(
|
|
memory_root, sources={"value.py": "value = 1\n"}, tests={"test_value.py": "def test_value(): pass\n"},
|
|
edit_location="src/value.py", replacement="value = 2\n", context_name="AlphaLoop",
|
|
runner=runner, is_remediation=True, remediation_originator_location="tests/test_value.py",
|
|
readonly_locations=("tests/test_value.py",), failure_outcome=TestRunOutcome.Error,
|
|
failure_error_message="setup failed", failure_raw_output="ERROR at setup", candidate=False,
|
|
)
|
|
|
|
assert isinstance(compiler, AlphaPythonCompiler)
|
|
result = compiler.recheck()
|
|
receipts = ValidationReceipts(compiler.blackboard.receipts, compiler.blackboard.accepted_generation)
|
|
|
|
assert isinstance(result, TestRunResult) and result.passed
|
|
receipt = receipts.current_for(manager.test_classes[0].location, "", file_scope=True)
|
|
assert receipt is not None and receipt.passed
|
|
|
|
|
|
@pytest.mark.parametrize("output,expected", (
|
|
("___ test_value ___\nE Failed: Timeout (>4.0s) from pytest-timeout.\n", TestRunOutcome.Timeout),
|
|
("FAILED tests/test_value.py::test_value - Failed: Timeout (>4.0s) from pytest-timeout.\n", TestRunOutcome.Timeout),
|
|
("++++++++++++++++ Timeout ++++++++++++++++\nThread stack follows", TestRunOutcome.Timeout),
|
|
('value = "E Failed: Timeout (>4.0s) from pytest-timeout."\nE AssertionError', TestRunOutcome.Failure),
|
|
('{"error": "Timeout (>4.0s) from pytest-timeout."}\nE AssertionError', TestRunOutcome.Failure),
|
|
("WARNING app: Timeout while connecting\nE AssertionError", TestRunOutcome.Failure),
|
|
))
|
|
def test_pytest_timeout_is_distinct_from_timeout_words_in_evidence(tmp_path, output, expected):
|
|
"""Only an actual pytest timeout diagnostic changes the execution outcome."""
|
|
runner, repository, manager = _strategy(tmp_path)
|
|
repository.queue_result(subprocess.CompletedProcess(["pytest"], 1, output, ""))
|
|
|
|
result = runner.run_tests_for(manager.test_classes)
|
|
|
|
assert result.outcome is expected
|
|
assert result.output == output
|
|
assert result.outcome.is_error is (expected is TestRunOutcome.Timeout)
|
|
assert "PROCESS DEADLINE SECONDS: 60" in result.run_info
|
|
if expected is TestRunOutcome.Timeout:
|
|
assert "TIMEOUT MECHANISM: pytest-reported timeout" in result.run_info
|
|
|
|
|
|
def test_mixed_timeout_and_assertion_keep_independent_reports_without_timeout_frames():
|
|
"""A timeout uses its collected file; a separate assertion still intersects methods."""
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_slow.py": "def test_slow(): pass\n",
|
|
"tests/test_value.py": "def test_value(): pass\n",
|
|
})
|
|
output = (
|
|
"___ test_slow ___\nE Failed: Timeout (>4.0s) from pytest-timeout.\n"
|
|
"___ test_value ___\ntests/test_value.py:1: in test_value\nE AssertionError\n"
|
|
"================ short test summary info ================\n"
|
|
"FAILED tests/test_slow.py::test_slow - Failed: Timeout (>4.0s) from pytest-timeout.\n"
|
|
"FAILED tests/test_value.py::test_value - AssertionError\n"
|
|
)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
reports = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Timeout, 1, output, run_info="PROCESS DEADLINE SECONDS: 7200",
|
|
))
|
|
|
|
assert [(item.test.location, item.test_name, item.outcome) for item in reports] == [
|
|
("tests/test_slow.py", "", TestRunOutcome.Timeout),
|
|
("tests/test_value.py", "test_value", TestRunOutcome.Failure),
|
|
]
|
|
assert all(item.run_info == "PROCESS DEADLINE SECONDS: 7200" for item in reports)
|
|
|
|
|
|
@pytest.mark.parametrize("partial", (
|
|
"___ test_value ___\ntests/test_value.py:1: in test_value\nE AssertionError\n",
|
|
"FAILED tests/test_value.py::test_value - AssertionError\n",
|
|
))
|
|
def test_process_deadline_is_not_hidden_by_an_earlier_report(partial):
|
|
"""A previously printed assertion does not discharge a later invocation deadline."""
|
|
manager = _MemoryFileManager(tests={"tests/test_value.py": "def test_value(): pass\n"})
|
|
output = "TIMEOUT: pytest exceeded 60 seconds\n" + partial
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
reports = runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest", TestRunOutcome.Timeout, 124, output, run_info="REQUESTED PATHS: tests/test_value.py",
|
|
))
|
|
|
|
assert [item.outcome for item in reports] == [TestRunOutcome.Failure, TestRunOutcome.Timeout]
|
|
assert reports[-1].raw_output == output
|
|
assert reports[-1].test_name == ""
|
|
|
|
|
|
def test_unlocalized_suite_timeout_retains_existing_error_scope_failure():
|
|
"""Absent scope remains an identification error, never an invented first test."""
|
|
manager = _MemoryFileManager(tests={
|
|
"tests/test_a.py": "def test_a(): pass\n", "tests/test_b.py": "def test_b(): pass\n",
|
|
})
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
|
|
with pytest.raises(ValueError, match="PROCESS DEADLINE SECONDS: 7200"):
|
|
runner.failure_reports(manager.test_classes, TestRunResult(
|
|
"pytest tests", TestRunOutcome.Timeout, 124, "TIMEOUT: pytest exceeded 7200 seconds",
|
|
run_info="PROCESS DEADLINE SECONDS: 7200",
|
|
))
|