Spaces:
Running
Running
File size: 4,817 Bytes
2b4bd40 57661f5 2b4bd40 57661f5 2b4bd40 57661f5 2b4bd40 57661f5 2b4bd40 | 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 | 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()
|