tests/test_evaluation_runner.py

import json
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from evaluation_runner import (
    AnswerCache,
    EvaluationClient,
    EvaluationError,
    RunnerSettings,
    audit_submission_answer,
    configure_utf8_stdio,
    safe_filename,
)


class AnswerCacheTests(unittest.TestCase):
    def test_round_trip_and_question_hash(self) -> None:
        question = {"task_id": "abc", "question": "What is 2 + 2?"}
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "answers.json"
            cache = AnswerCache(path)
            cache.record(question, "4", "test-agent", None)

            reloaded = AnswerCache(path)
            self.assertEqual(reloaded.get_valid(question), "4")
            self.assertTrue(reloaded.is_validated(question))
            changed = {"task_id": "abc", "question": "What is 3 + 3?"}
            self.assertIsNone(reloaded.get_valid(changed))
            self.assertEqual(json.loads(path.read_text())["version"], 1)

    def test_best_effort_metadata_round_trip(self) -> None:
        question = {"task_id": "abc", "question": "What is the count?"}
        with tempfile.TemporaryDirectory() as directory:
            path = Path(directory) / "answers.json"
            cache = AnswerCache(path)
            cache.record(
                question,
                "41",
                "test-agent",
                None,
                validated=False,
                validation_issues=["Only one weak source was available."],
            )

            reloaded = AnswerCache(path)
            self.assertEqual(reloaded.get_valid(question), "41")
            self.assertFalse(reloaded.is_validated(question))
            entry = reloaded.get_entry(question)
            self.assertEqual(
                entry["validation_issues"],
                ["Only one weak source was available."],
            )


class PathSafetyTests(unittest.TestCase):
    def test_filename_discards_parent_directories(self) -> None:
        self.assertEqual(safe_filename("../../secret.py"), "secret.py")
        self.assertEqual(safe_filename(r"..\..\sheet data.xlsx"), "sheet_data.xlsx")

    def test_filename_rejects_empty_component(self) -> None:
        with self.assertRaises(EvaluationError):
            safe_filename("..")


class SubmissionAuditTests(unittest.TestCase):
    def test_normalizes_single_word_terminal_punctuation(self) -> None:
        self.assertEqual(audit_submission_answer("task", "Extremely."), "Extremely")

    def test_rejects_placeholder_and_punctuation_only_answers(self) -> None:
        with self.assertRaises(EvaluationError):
            audit_submission_answer("task", "N/A")
        with self.assertRaises(EvaluationError):
            audit_submission_answer("task", "},{")


class Utf8ConsoleTests(unittest.TestCase):
    def test_configures_both_output_streams_for_utf8(self) -> None:
        class ReconfigurableStream:
            def __init__(self) -> None:
                self.calls = []

            def reconfigure(self, **kwargs) -> None:
                self.calls.append(kwargs)

        stdout = ReconfigurableStream()
        stderr = ReconfigurableStream()
        with patch("evaluation_runner.sys.stdout", stdout), patch(
            "evaluation_runner.sys.stderr", stderr
        ):
            configure_utf8_stdio()

        expected = [{"encoding": "utf-8", "errors": "replace"}]
        self.assertEqual(stdout.calls, expected)
        self.assertEqual(stderr.calls, expected)


class GaiaFallbackTests(unittest.TestCase):
    def test_official_dataset_file_is_copied_to_private_task_cache(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            downloaded = root / "hub-file.mp3"
            downloaded.write_bytes(b"test audio")
            destination = root / "attachments" / "task.mp3"
            settings = RunnerSettings(
                api_url="https://example.test",
                username="tester",
                space_id="tester/space",
                local_dir=root,
                gaia_repo_id="gaia-benchmark/GAIA",
                gaia_data_dir="2023/validation",
            )
            client = EvaluationClient(settings)

            with patch(
                "huggingface_hub.hf_hub_download", return_value=str(downloaded)
            ) as hub_download:
                result = client._download_gaia_attachment("task.mp3", destination)

            self.assertEqual(result, destination)
            self.assertEqual(destination.read_bytes(), b"test audio")
            self.assertEqual(
                hub_download.call_args.kwargs["filename"],
                "2023/validation/task.mp3",
            )


if __name__ == "__main__":
    unittest.main()