Spaces:
Running
Running
File size: 3,228 Bytes
ed28aa2 | 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 | import json
import os
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
from benchmarks.shadow_telemetry import validate_and_record_shadow
class ShadowTelemetryTests(unittest.TestCase):
def test_disabled_mode_does_not_write_or_validate(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "shadow.jsonl"
with patch.dict(os.environ, {"BENCHMARK_SHADOW_MODE": "0", "BENCHMARK_SHADOW_LOG_PATH": str(path)}, clear=False):
result = validate_and_record_shadow(goal="MMLU domanda A/B/C/D", answer="ANSWER: C")
self.assertIsNone(result)
self.assertFalse(path.exists())
def test_mmlu_failure_event_is_recorded_without_raw_answer(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "shadow.jsonl"
raw = "RISPOSTA SEGRETA: il testo non deve essere salvato"
with patch.dict(os.environ, {"BENCHMARK_SHADOW_MODE": "1", "BENCHMARK_SHADOW_LOG_PATH": str(path)}, clear=False):
result = validate_and_record_shadow(
goal="MMLU domanda di informatica A/B/C/D",
answer=raw,
metadata={"provider": "mock", "model": "test", "secret": "must-drop"},
)
self.assertIsNotNone(result)
self.assertEqual(result.failure_code, "answer_missing")
event = json.loads(path.read_text(encoding="utf-8"))
self.assertEqual(event["category"], "mmlu")
self.assertEqual(event["failure_code"], "answer_missing")
self.assertEqual(event["metadata"], {"provider": "mock", "model": "test"})
self.assertNotIn(raw, path.read_text(encoding="utf-8"))
def test_coding_failure_event_is_recorded(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "shadow.jsonl"
with patch.dict(os.environ, {"BENCHMARK_SHADOW_MODE": "true", "BENCHMARK_SHADOW_LOG_PATH": str(path)}, clear=False):
result = validate_and_record_shadow(
goal="code_correct: implementa in TypeScript",
answer="Non posso includere il codice.",
metadata={"attempt": 1, "latency_ms": 123.4},
)
self.assertIsNotNone(result)
self.assertEqual(result.failure_code, "code_missing")
event = json.loads(path.read_text(encoding="utf-8"))
self.assertEqual(event["validator"], "coding_v1")
self.assertEqual(event["metadata"]["attempt"], 1)
self.assertEqual(event["metadata"]["latency_ms"], 123.4)
def test_unsupported_category_is_ignored(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "shadow.jsonl"
with patch.dict(os.environ, {"BENCHMARK_SHADOW_MODE": "1", "BENCHMARK_SHADOW_LOG_PATH": str(path)}, clear=False):
result = validate_and_record_shadow(goal="generic task", answer="output")
self.assertIsNone(result)
self.assertFalse(path.exists())
if __name__ == "__main__":
unittest.main()
|