Spaces:
Sleeping
Sleeping
File size: 7,551 Bytes
9001c13 4787daf 9001c13 0bfffbc 9001c13 6aa62ca 9001c13 4787daf 9001c13 | 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 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 | from __future__ import annotations
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
import requests
from lilith_agent.scoring_client import ScoringApiClient
class ScoringApiClientTests(unittest.TestCase):
def test_get_questions_returns_api_payload_on_success(self) -> None:
response = Mock()
response.json.return_value = [{"task_id": "t1", "question": "Q"}]
response.raise_for_status.return_value = None
session = Mock()
session.get.return_value = response
client = ScoringApiClient(api_url="https://example.com", session=session)
payload = client.get_questions()
self.assertEqual(payload, [{"task_id": "t1", "question": "Q"}])
self.assertIsNone(client.last_warning)
session.get.assert_called_once_with("https://example.com/questions", timeout=30)
def test_get_questions_falls_back_to_dataset_on_429(self) -> None:
response = Mock(status_code=429, headers={})
error = requests.HTTPError("Too Many Requests", response=response)
response.raise_for_status.side_effect = error
session = Mock()
session.get.return_value = response
dataset_client = Mock()
dataset_client.get_questions.return_value = [{"task_id": "fallback", "question": "Q2"}]
client = ScoringApiClient(
api_url="https://example.com",
session=session,
dataset_client=dataset_client,
)
payload = client.get_questions()
self.assertEqual(payload, [{"task_id": "fallback", "question": "Q2"}])
self.assertIn("429", client.last_warning or "")
dataset_client.get_questions.assert_called_once_with()
def test_fallback_prints_hf_visible_warning(self) -> None:
response = Mock(status_code=503, headers={})
error = requests.HTTPError("Service Unavailable", response=response)
response.raise_for_status.side_effect = error
session = Mock()
session.get.return_value = response
dataset_client = Mock()
dataset_client.get_questions.return_value = []
client = ScoringApiClient(
api_url="https://example.com",
session=session,
dataset_client=dataset_client,
)
with patch("builtins.print") as printed:
client.get_questions()
printed.assert_any_call(
"Scoring API unavailable while trying to fetch questions (status=503); falling back to GAIA dataset.",
flush=True,
)
def test_download_file_falls_back_to_dataset_on_429(self) -> None:
response = Mock(status_code=429, headers={})
error = requests.HTTPError("Too Many Requests", response=response)
response.raise_for_status.side_effect = error
session = Mock()
session.get.return_value = response
dataset_client = Mock()
with tempfile.TemporaryDirectory() as tmpdir:
fallback_path = Path(tmpdir) / "from-dataset.txt"
fallback_path.write_text("ok")
dataset_client.download_file.return_value = fallback_path
client = ScoringApiClient(
api_url="https://example.com",
session=session,
dataset_client=dataset_client,
)
payload = client.download_file("task-123", tmpdir)
self.assertEqual(payload, fallback_path)
self.assertIn("429", client.last_warning or "")
dataset_client.download_file.assert_called_once_with("task-123", Path(tmpdir))
def test_download_file_falls_back_to_dataset_on_404(self) -> None:
response = Mock(status_code=404, headers={})
error = requests.HTTPError("Not Found", response=response)
response.raise_for_status.side_effect = error
session = Mock()
session.get.return_value = response
dataset_client = Mock()
with tempfile.TemporaryDirectory() as tmpdir:
fallback_path = Path(tmpdir) / "from-dataset.png"
fallback_path.write_bytes(b"image-bytes")
dataset_client.download_file.return_value = fallback_path
client = ScoringApiClient(
api_url="https://example.com",
session=session,
dataset_client=dataset_client,
)
with patch("builtins.print") as printed:
payload = client.download_file("task-404", tmpdir)
self.assertEqual(payload, fallback_path)
self.assertIn("404", client.last_warning or "")
dataset_client.download_file.assert_called_once_with("task-404", Path(tmpdir))
printed.assert_any_call(
"Scoring API unavailable while trying to download file for task-404 (status=404); falling back to GAIA dataset.",
flush=True,
)
def test_download_file_falls_back_when_api_returns_tiny_json_payload(self) -> None:
response = Mock(status_code=200, headers={"content-type": "application/json"})
response.content = b'{"detail":"file not found"}'
response.raise_for_status.return_value = None
session = Mock()
session.get.return_value = response
dataset_client = Mock()
with tempfile.TemporaryDirectory() as tmpdir:
fallback_path = Path(tmpdir) / "from-dataset.png"
fallback_path.write_bytes(b"image-bytes")
dataset_client.download_file.return_value = fallback_path
client = ScoringApiClient(
api_url="https://example.com",
session=session,
dataset_client=dataset_client,
)
with patch("builtins.print") as printed:
payload = client.download_file("task-json", tmpdir)
self.assertEqual(payload, fallback_path)
dataset_client.download_file.assert_called_once_with("task-json", Path(tmpdir))
printed.assert_any_call(
"[scoring] invalid file payload task=task-json status=200 bytes=27 content_type='application/json'; trying dataset fallback",
flush=True,
)
def test_default_dataset_client_is_built_lazily(self) -> None:
session = Mock()
response = Mock(status_code=429, headers={})
error = requests.HTTPError("Too Many Requests", response=response)
response.raise_for_status.side_effect = error
session.get.return_value = response
dataset_instance = Mock()
dataset_instance.get_questions.return_value = []
with patch.object(ScoringApiClient, "_get_dataset_client", return_value=dataset_instance) as getter:
client = ScoringApiClient(api_url="https://example.com", session=session)
client.get_questions()
getter.assert_called_once_with()
def test_default_dataset_client_uses_validation_split(self) -> None:
dataset_instance = Mock()
with patch.dict(os.environ, {}, clear=True), patch(
"lilith_agent.gaia_dataset.GaiaDatasetClient",
return_value=dataset_instance,
) as dataset_client:
client = ScoringApiClient(api_url="https://example.com", session=Mock())
payload = client._get_dataset_client()
self.assertIs(payload, dataset_instance)
dataset_client.assert_called_once_with(
config="2023_all",
split="validation",
level=None,
token=None,
)
if __name__ == "__main__":
unittest.main()
|