122 lines
5.4 KiB
Python
122 lines
5.4 KiB
Python
"""Replay Beta's missing-selector cycle through execution, routing and retry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.working_backlog import TestingBacklog
|
|
from band_aider.judgement.pytest_runner_strategy import PytestRunnerStrategy
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
from band_aider.loop.exceptions import BetaRequeueRequested
|
|
from tests.band_aider import test_beta_prompt_refinement_boundary as boundary
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
@pytest.fixture
|
|
def logged_selector_error() -> TestingBacklog:
|
|
"""Restore the stale identity and unmodified error from target-2 provenance."""
|
|
fixture = Path(__file__).parents[1] / "fixtures" / "beta_target2_selector.json"
|
|
record = json.loads(fixture.read_text(encoding="utf-8"))
|
|
return TestingBacklog(
|
|
summary="FAILED CANDIDATE",
|
|
focused_spotlight=ClassMeta(
|
|
record["test_location"], "recorded test file", [], {},
|
|
content_type=ContentType.Test,
|
|
),
|
|
failure_error_message=record["error_message"],
|
|
failure_raw_output=record["raw_output"],
|
|
failure_stack_trace=record["stack_trace"],
|
|
failure_test_name=record["test_name"],
|
|
)
|
|
|
|
|
|
def test_logged_missing_selector_retries_file_without_discharging_error(
|
|
logged_selector_error: TestingBacklog,
|
|
) -> None:
|
|
"""A removed callable must not survive a fresh collection-error requeue."""
|
|
item = logged_selector_error
|
|
assert item.focused_spotlight is not None
|
|
location = item.focused_spotlight.location
|
|
current_source = (
|
|
"def test_stabilization_remediation_claimed_change_does_not_retest_passing_origin():\n"
|
|
" assert True\n"
|
|
)
|
|
client = boundary._QueuedClient()
|
|
manager = boundary._MemoryFileManager(client=client, tests={location: current_source})
|
|
repository = cast(MemoryRepoExec, manager.repo_exec)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
beta = boundary._BetaProbe(boundary._context(manager), runner=runner)
|
|
raw_output = item.failure_raw_output
|
|
obsolete_name = item.failure_test_name
|
|
repository.queue_result(subprocess.CompletedProcess(["pytest"], 4, raw_output, ""))
|
|
|
|
result = beta._run_targeted_tests(item)
|
|
assert result.outcome is TestRunOutcome.Error
|
|
assert result.test_name == ""
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(item, "Keep the valid renamed test.", result, boundary._applied(item))
|
|
retry = raised.value.testing_item
|
|
|
|
assert raised.value.stop is False
|
|
assert retry.focused_spotlight is item.focused_spotlight
|
|
assert retry.failure_outcome is TestRunOutcome.Error
|
|
assert retry.failure_raw_output == raw_output
|
|
assert retry.failure_test_name == ""
|
|
assert retry.failure_run_info == result.run_info
|
|
assert obsolete_name not in retry.failure_run_info
|
|
repository.queue_result(subprocess.CompletedProcess(["pytest"], 0, "1 passed", ""))
|
|
|
|
assert beta._run_targeted_tests(retry).passed
|
|
first, second = repository.commands
|
|
assert first[-1] == f"/testbed/{location}"
|
|
assert second[-1] == f"/testbed/{location}"
|
|
assert "--maxfail=1" in second
|
|
assert manager.repo_exec.read_text(manager.test_path(location)) == current_source
|
|
assert client.prompts == []
|
|
|
|
|
|
def test_trace_free_timeout_retries_file_and_preserves_invocation(
|
|
logged_selector_error: TestingBacklog,
|
|
) -> None:
|
|
"""A process deadline uses the same file route without inventing a frame."""
|
|
item = logged_selector_error
|
|
assert item.focused_spotlight is not None
|
|
location = item.focused_spotlight.location
|
|
item.failure_test_name = "test_final_cot_runs_direct_remediation_outside_forward_topology"
|
|
client = boundary._QueuedClient()
|
|
manager = boundary._MemoryFileManager(client=client, tests={
|
|
location: "def test_final_cot_runs_direct_remediation_outside_forward_topology(): pass\n",
|
|
})
|
|
repository = cast(MemoryRepoExec, manager.repo_exec)
|
|
runner = PytestRunnerStrategy(manager, manager.repo_exec)
|
|
beta = boundary._BetaProbe(boundary._context(manager), runner=runner)
|
|
repository.queue_result(subprocess.TimeoutExpired(
|
|
["pytest"], runner.TARGETED_TIMEOUT_SECONDS, output="collected 1 item\n",
|
|
))
|
|
|
|
result = beta._run_targeted_tests(item)
|
|
assert result.outcome is TestRunOutcome.Timeout
|
|
with pytest.raises(BetaRequeueRequested) as raised:
|
|
beta._raise_for_targeted_test_failure(item, "Diagnose the timeout.", result, boundary._applied(item))
|
|
retry = raised.value.testing_item
|
|
|
|
assert retry.failure_outcome is TestRunOutcome.Timeout
|
|
assert retry.failure_test_name == ""
|
|
assert retry.failure_raw_output == result.output
|
|
assert "collected 1 item" in retry.failure_raw_output
|
|
assert retry.failure_run_info == result.run_info
|
|
assert "PROCESS DEADLINE SECONDS: 60" in retry.failure_run_info
|
|
assert '"--maxfail=1"' in retry.failure_run_info
|
|
assert location in retry.failure_run_info
|
|
assert "CWD: /testbed" in retry.failure_run_info
|
|
repository.queue_result(subprocess.CompletedProcess(["pytest"], 0, "1 passed", ""))
|
|
|
|
assert beta._run_targeted_tests(retry).passed
|
|
assert repository.commands[-1][-1] == f"/testbed/{location}"
|
|
assert client.prompts == []
|