84 lines
4.2 KiB
Python
84 lines
4.2 KiB
Python
"""Meaningful scorer and frozen-provenance checks; no inference."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
spec = importlib.util.spec_from_file_location('mixed_eval', Path(__file__).with_name('evaluate.py'))
|
|
evaluate = importlib.util.module_from_spec(spec); spec.loader.exec_module(evaluate)
|
|
|
|
|
|
class EvaluationTests(unittest.TestCase):
|
|
def test_existing_report_is_preserved_and_new_report_is_reserved(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'nested' / 'report.json'
|
|
evaluate.reserve_report(path)
|
|
evaluate.write_report(path, {'complete': False, 'query': 'Straße'})
|
|
previous = path.read_bytes()
|
|
with self.assertRaises(FileExistsError):
|
|
evaluate.reserve_report(path)
|
|
self.assertEqual(path.read_bytes(), previous)
|
|
evaluate.write_report(path, {'complete': True})
|
|
self.assertIn('true', path.read_text(encoding='utf-8'))
|
|
self.assertEqual(list(path.parent.glob('*.partial')), [])
|
|
|
|
def test_failed_checkpoint_publication_keeps_previous_report(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / 'report.json'
|
|
evaluate.reserve_report(path)
|
|
evaluate.write_report(path, {'complete': False, 'rows': [1]})
|
|
previous = path.read_bytes()
|
|
with patch.object(evaluate.os, 'replace', side_effect=OSError('disk failure')):
|
|
with self.assertRaises(OSError):
|
|
evaluate.write_report(path, {'complete': True})
|
|
self.assertEqual(path.read_bytes(), previous)
|
|
self.assertEqual(list(path.parent.glob('*.partial')), [])
|
|
|
|
def test_frozen_real_assets_verify(self):
|
|
fixture, _ = evaluate.load_fixture()
|
|
self.assertEqual(len(fixture['assets']), 32)
|
|
self.assertEqual(len(fixture['queries']), 91)
|
|
self.assertTrue(any(a.get('source_repository') for a in fixture['assets']))
|
|
|
|
def test_multiple_relevant_files_and_duplicate_hits(self):
|
|
query = {'expected': ['a', 'b'], 'hard_negatives': ['x']}
|
|
result = evaluate.score(query, [{'asset': x} for x in ['x', 'a', 'a', 'c', 'b']])
|
|
self.assertEqual(result['first_relevant_rank'], 2)
|
|
self.assertEqual(result['recall_at_5'], 1)
|
|
self.assertEqual(result['reciprocal_rank'], .5)
|
|
self.assertTrue(result['hard_negative_in_top5'])
|
|
|
|
def test_no_answer_is_not_counted_as_positive_recall(self):
|
|
result = evaluate.score({'expected': []}, [{'asset': 'unrelated'}])
|
|
self.assertTrue(result['no_answer_returned_any'])
|
|
self.assertIsNone(result['recall_at_5'])
|
|
self.assertIsNone(result['top1'])
|
|
self.assertIsNone(result['reciprocal_rank'])
|
|
|
|
def test_temporal_labels_file_relevance_only(self):
|
|
result = evaluate.score({'expected': [{'asset': 'speech', 'start_s': 20, 'end_s': 25}]}, [{'asset': 'speech', 'source': {'number': 0}}])
|
|
self.assertTrue(result['top1']) # Deliberately no invented timestamp success metric.
|
|
|
|
def test_label_tampering_rejected_before_assets(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory)
|
|
(path / 'fixture.json').write_text('{"assets": []}')
|
|
(path / 'labels.lock.json').write_text('{"fixture_sha256": "wrong"}')
|
|
with self.assertRaisesRegex(ValueError, 'Frozen labels changed'):
|
|
evaluate.load_fixture(path)
|
|
|
|
def test_summary_separates_scope_and_language(self):
|
|
rows = []
|
|
for scope, lang, expected, found in [('all_files', 'en', ['a'], ['b']), ('code', 'en', ['a'], ['a']), ('all_files', 'de', [], ['b'])]:
|
|
query = {'expected': expected, 'language': lang, 'category': 'code'}
|
|
rows.append({'mode': 'hybrid', 'scope': scope, 'query': query, 'score': evaluate.score(query, [{'asset': a} for a in found])})
|
|
groups = evaluate.summarize(rows)
|
|
self.assertEqual(len(groups), 3)
|
|
self.assertEqual(next(g for g in groups if g['scope'] == 'code')['top1'], 1)
|
|
self.assertEqual(next(g for g in groups if g['language'] == 'de')['no_answer_returned_any'], 1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|