source/tests/test_programbench_cleanup.py

"""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