Spaces:
Running
Running
File size: 6,689 Bytes
a0f7ac6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Exercise the cleanup repair through the pinned real step/compile methods."""
from argparse import Namespace
from pathlib import Path
from unittest.mock import patch
import pytest
from programbench.eval.eval import Evaluator, TestResult as UpstreamTestResult
from programbench.exceptions import EvalStepError
from programbench.submission import score_from_tests, test_results_map as upstream_results_map
from programbench.utils.load_data import get_active_branches, get_ignored_branches, get_ignored_tests
from study import programbench_adapter as adapter
FIGLET_INSTANCE = "cmatsuoka__figlet.202a0a8"
def result(code=0, output="", exception=""):
return {"returncode": code, "output": output, "exception_info": exception}
ENOTEMPTY = result(1, "rm: cannot remove '/workspace/.git/example': Directory not empty\n")
class SubmissionReached(Exception):
pass
class Environment:
def __init__(self, replies):
self.replies = iter(replies)
self.commands = []
self.extracted = False
def execute(self, command, timeout):
self.commands.append((command, timeout))
return next(self.replies)
def copy_in_tar(self, *_):
self.extracted = True
raise SubmissionReached
class RepairEvaluator(Evaluator):
def __init__(self, instance_id=FIGLET_INSTANCE, reference=False):
super().__init__(tests_branches=[], instance_id=instance_id, submission_archive=Path("unused.tar.gz"))
self.reference = reference
def _run_step(self, command, **kwargs):
return adapter.cleanup_step(super()._run_step, command, instance_id=self.instance_id,
reference=self.reference, **kwargs)
def test_retry_preserves_failure_and_proves_empty_before_extraction():
evaluator = RepairEvaluator()
env = Environment([ENOTEMPTY, result(), result(output="0\n")])
with pytest.raises(SubmissionReached):
evaluator._compile_executable(env, evaluator.result.log)
assert env.extracted
assert env.commands == [(adapter.WORKSPACE_WIPE, 300), (adapter.WORKSPACE_WIPE, 300),
(adapter.VERIFY_WORKSPACE_EMPTY, 30)]
assert [entry["step"] for entry in evaluator.result.log] == [
"wipe_workspace_initial_attempt", "wipe_workspace_retry", "verify_workspace_empty"]
assert evaluator.result.log[0]["output"] == ENOTEMPTY["output"]
assert evaluator.result.log[0]["returncode"] == 1
def test_first_success_still_requires_empty_workspace():
evaluator = RepairEvaluator()
env = Environment([result(), result(output="0\n")])
with pytest.raises(SubmissionReached):
evaluator._compile_executable(env, evaluator.result.log)
assert len(env.commands) == 2
@pytest.mark.parametrize("initial", [result(137, ENOTEMPTY["output"]), result(1, "Permission denied"),
result(1, ENOTEMPTY["output"] + "Error response from daemon: unavailable\n"),
result(1, ENOTEMPTY["output"], "transport timeout"), result(0, "", "transport failure"),
result(1, "rm: cannot remove '/elsewhere/file': Directory not empty\n")])
def test_unproven_failures_are_not_retried(initial):
evaluator = RepairEvaluator()
env = Environment([initial])
with pytest.raises(EvalStepError):
evaluator._compile_executable(env, evaluator.result.log)
assert not env.extracted and len(env.commands) == 1
def test_retry_is_bounded_and_a_second_failure_stops_extraction():
evaluator = RepairEvaluator()
env = Environment([ENOTEMPTY, ENOTEMPTY])
with pytest.raises(EvalStepError):
evaluator._compile_executable(env, evaluator.result.log)
assert not env.extracted and len(env.commands) == 2
@pytest.mark.parametrize("verification", [result(1, "1\n"), result(0, "1\n"),
result(0, "garbage"), result(0, "0\n", "transport failure")])
def test_unverified_emptiness_blocks_extraction(verification):
evaluator = RepairEvaluator()
env = Environment([result(), verification])
with pytest.raises(EvalStepError):
evaluator._compile_executable(env, evaluator.result.log)
assert not env.extracted
@pytest.mark.parametrize("instance,reference", [("other.task", False), (FIGLET_INSTANCE, True)])
def test_other_task_and_reference_keep_native_single_step(instance, reference):
evaluator = RepairEvaluator(instance, reference)
env = Environment([ENOTEMPTY])
with pytest.raises(EvalStepError, match="wipe_workspace_failed"):
evaluator._compile_executable(env, evaluator.result.log)
assert not env.extracted and len(env.commands) == 1
assert evaluator.result.log[0]["step"] == "wipe_workspace"
def test_other_commands_do_not_gain_retry():
evaluator = RepairEvaluator()
env = Environment([ENOTEMPTY])
with pytest.raises(EvalStepError):
evaluator._run_step("different cleanup", env=env, log_buf=evaluator.result.log,
step_name="wipe_workspace", timeout=300)
assert len(env.commands) == 1
def test_adapter_wires_real_compile_step_and_emits_separate_repair_metadata(tmp_path):
task = next(item for item in adapter.load_manifest(adapter.DEFAULT_MANIFEST)["tasks"]
if item["instance_id"] == FIGLET_INSTANCE)
instance = {"image_name": "unused", "branches": {"branch": {"tests": ["example"]}}}
(tmp_path / "tests").mkdir()
(tmp_path / "tests/branch.tar.gz").touch()
class TestEvaluator(Evaluator):
def run(self):
env = Environment([ENOTEMPTY, result(), result(output="0\n")])
with pytest.raises(SubmissionReached):
self._compile_executable(env, self.result.log)
self.result.test_results.append(UpstreamTestResult(name="example", branch="branch", status="passed", extra={}))
return self.result
api = {"Evaluator": TestEvaluator, "get_active_branches": get_active_branches,
"get_ignored_branches": get_ignored_branches, "get_ignored_tests": get_ignored_tests,
"get_blob_dir": lambda _: tmp_path, "test_results_map": upstream_results_map,
"score_from_tests": score_from_tests}
args = Namespace(docker="docker", cpus=4, memory="30g", evaluator_wheelhouse=None,
branch_timeout_seconds=14400)
with patch.object(adapter, "check_local_image"):
summary = adapter.evaluate(args, task, instance, api, tmp_path / "result", reference=False,
submission=tmp_path / "unused.tar.gz")
assert summary["workspace_cleanup_repair"] == adapter.CLEANUP_REPAIR
assert summary["evaluator_repair"] == "pause_timeout_during_pytest_report_v1"
assert summary["score"] == 1 and summary["complete"] is True
|