tests/test_shadow_telemetry.py

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()