tests/unit/cli/test_task_annotate.py - master - egorbu/harbor_fix

tests/unit/cli/test_task_annotate.py
208 строк7 KB
Alex Shaw
Refactor Harbor task handling and viewer UI (#2023)
21 июн 2026, 06:43
21 июн 2026, 06:43c6f4898
import json
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

from harbor.cli.annotator.annotator import (
    ANNOTATE_TASK_TEMPLATE_DIR,
    AnnotateOutput,
    _build_file_tree,
    _extract_annotate_result,
    _write_results,
    assemble_annotate_task,
)
from harbor.models.task.task import Task


def _make_task_dir(tmp_path: Path, name: str = "task") -> Path:
    task_dir = tmp_path / name
    task_dir.mkdir()
    (task_dir / "instruction.md").write_text("Do the thing.")
    (task_dir / "task.toml").write_text(
        'schema_version = "1.3"\n\n'
        "[task]\n"
        'name = "harbor/task"\n'
        'description = "Old description."\n'
    )
    env = task_dir / "environment"
    env.mkdir()
    (env / "Dockerfile").write_text("FROM ubuntu:22.04\n")
    tests = task_dir / "tests"
    tests.mkdir()
    (tests / "test.sh").write_text("#!/bin/bash\nexit 0\n")
    return task_dir


def _assemble(tmp_path: Path) -> Path:
    return assemble_annotate_task(
        task_dir=_make_task_dir(tmp_path),
        output_schema=AnnotateOutput.model_json_schema(),
        dest=tmp_path / "work" / "annotate-task",
    )


class TestBuildFileTree:
    @pytest.mark.unit
    def test_returns_file_listing(self, tmp_path):
        task_dir = _make_task_dir(tmp_path)
        tree = _build_file_tree(task_dir)
        assert "instruction.md" in tree
        assert "task.toml" in tree
        assert "tests/test.sh" in tree

    @pytest.mark.unit
    def test_empty_dir_returns_no_files(self, tmp_path):
        empty_dir = tmp_path / "empty"
        empty_dir.mkdir()
        assert _build_file_tree(empty_dir) == "No files found"


class TestAssembleAnnotateTask:
    @pytest.mark.unit
    def test_wrapper_is_valid_task_dir(self, tmp_path):
        assert Task.is_valid_dir(_assemble(tmp_path))

    @pytest.mark.unit
    def test_layout(self, tmp_path):
        wrapper = _assemble(tmp_path)
        assert (wrapper / "task.toml").is_file()
        assert (wrapper / "instruction.md").is_file()
        assert (wrapper / "environment" / "task" / "instruction.md").is_file()
        assert (wrapper / "tests" / "test.sh").is_file()
        assert (wrapper / "tests" / "validate.py").is_file()

    @pytest.mark.unit
    def test_instruction_contains_tree_path_and_contract(self, tmp_path):
        wrapper = _assemble(tmp_path)
        instruction = (wrapper / "instruction.md").read_text()
        assert "/app/task" in instruction
        assert "tests/test.sh" in instruction
        assert "annotate-result.json" in instruction
        assert "<output_schema>" in instruction

    @pytest.mark.unit
    def test_git_dir_excluded_from_copy(self, tmp_path):
        task_dir = _make_task_dir(tmp_path)
        (task_dir / ".git").mkdir()
        (task_dir / ".git" / "HEAD").write_text("ref: refs/heads/main")
        wrapper = assemble_annotate_task(
            task_dir=task_dir,
            output_schema=AnnotateOutput.model_json_schema(),
            dest=tmp_path / "work" / "annotate-task",
        )
        assert not (wrapper / "environment" / "task" / ".git").exists()


def _run_validate(tmp_path: Path, result: object) -> tuple[int, str]:
    import shutil

    work = tmp_path / "validate-run"
    work.mkdir()
    shutil.copy(
        ANNOTATE_TASK_TEMPLATE_DIR / "tests" / "validate.py", work / "validate.py"
    )
    result_path = work / "annotate-result.json"
    if result is not None:
        result_path.write_text(
            result if isinstance(result, str) else json.dumps(result)
        )
    proc = subprocess.run(
        [sys.executable, str(work / "validate.py"), str(result_path)],
        capture_output=True,
        text=True,
    )
    return proc.returncode, proc.stdout


class TestValidateScript:
    @pytest.mark.unit
    def test_valid_result_passes(self, tmp_path):
        code, _ = _run_validate(
            tmp_path, {"readme": "# Task\n", "description": "Evaluates X."}
        )
        assert code == 0

    @pytest.mark.unit
    def test_missing_file_fails(self, tmp_path):
        code, out = _run_validate(tmp_path, None)
        assert code == 1
        assert "missing result file" in out

    @pytest.mark.unit
    def test_invalid_json_fails(self, tmp_path):
        code, out = _run_validate(tmp_path, "{not json")
        assert code == 1
        assert "invalid JSON" in out

    @pytest.mark.unit
    def test_empty_field_fails(self, tmp_path):
        code, out = _run_validate(tmp_path, {"readme": "", "description": "x"})
        assert code == 1
        assert "readme must be a non-empty string" in out

    @pytest.mark.unit
    def test_extra_key_fails(self, tmp_path):
        code, out = _run_validate(
            tmp_path,
            {"readme": "# Task", "description": "x", "extra": "nope"},
        )
        assert code == 1
        assert "unexpected key: extra" in out


def _fake_trial_result(reward: float | None = 1, exception: bool = False):
    return SimpleNamespace(
        exception_info=(
            SimpleNamespace(exception_type="RuntimeError", exception_message="boom")
            if exception
            else None
        ),
        verifier_result=(
            SimpleNamespace(rewards={"reward": reward}) if reward is not None else None
        ),
    )


class TestExtractAnnotateResult:
    @pytest.mark.unit
    def test_success(self, tmp_path):
        trial_dir = tmp_path / "trial"
        (trial_dir / "artifacts").mkdir(parents=True)
        (trial_dir / "artifacts" / "annotate-result.json").write_text(
            json.dumps({"readme": "# Task", "description": "Evaluates X."})
        )
        result = _extract_annotate_result(_fake_trial_result(), trial_dir)
        assert result.readme == "# Task"
        assert result.description == "Evaluates X."

    @pytest.mark.unit
    def test_trial_exception_raises(self, tmp_path):
        with pytest.raises(RuntimeError, match="boom"):
            _extract_annotate_result(
                _fake_trial_result(exception=True), tmp_path / "trial"
            )

    @pytest.mark.unit
    def test_zero_reward_raises_with_verifier_output(self, tmp_path):
        trial_dir = tmp_path / "trial"
        (trial_dir / "verifier").mkdir(parents=True)
        (trial_dir / "verifier" / "test-stdout.txt").write_text("missing readme")
        with pytest.raises(ValueError, match="missing readme"):
            _extract_annotate_result(_fake_trial_result(reward=0), trial_dir)


class TestWriteResults:
    @pytest.mark.unit
    def test_writes_readme_and_description(self, tmp_path):
        task_dir = _make_task_dir(tmp_path)
        _write_results(
            task_dir,
            AnnotateOutput(readme="# New README\n", description="New description."),
        )
        assert (task_dir / "README.md").read_text() == "# New README\n"
        assert (
            'description = "New description."' in (task_dir / "task.toml").read_text()
        )