#!/usr/bin/env python3 """Synthetic comparator regression only; no measured performance evidence.""" import copy import csv import importlib.util import json from pathlib import Path import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location('compare_performance', ROOT / 'tools/compare_performance.py') comparison = importlib.util.module_from_spec(spec) spec.loader.exec_module(comparison) class PerformanceComparisonTests(unittest.TestCase): def setUp(self): self.directory = tempfile.TemporaryDirectory(prefix='erri-synthetic-performance-') self.addCleanup(self.directory.cleanup) self.root = Path(self.directory.name) template = { 'variant': 'baseline', 'repeat_id': 'r1', 'artifact_id': 'synthetic-baseline', 'addon_ids': ['synthetic-base'], 'engine_version': 'synthetic-engine', 'hardware': {'cpu': 'synthetic-cpu', 'gpu': 'synthetic-gpu', 'ram_gib': 32, 'os': 'synthetic-os', 'driver': 'synthetic-driver'}, 'workload': {'map': 'synthetic-map', 'route': 'synthetic-route', 'players': 12, 'ai': 36, 'vehicles': 4, 'entities': 100, 'settings': {'resolution': [1920, 1080], 'fps_limit': 0, 'vsync': False}, 'warmup_seconds': 30, 'duration_seconds': 60}, 'capture': {'kind': 'client_frame', 'environment': 'client', 'tool': 'synthetic-generator', 'version': '1'}, 'sample_file': 'unused.csv', } self.data = {'schema_version': 1, 'sample_origin': 'synthetic', 'runs': []} for repeat in range(1, 4): for variant, frame in [('baseline', 20), ('successor', 10)]: run = copy.deepcopy(template) run.update(variant=variant, repeat_id=f'r{repeat}', artifact_id=f'synthetic-{variant}', sample_file=f'{variant}-{repeat}.csv') run['addon_ids'] = [f'synthetic-{variant}'] self.data['runs'].append(run) with (self.root / run['sample_file']).open('w', newline='') as stream: writer = csv.writer(stream) writer.writerow(comparison.COLUMNS) for index in range(1, 90000 // frame + 1): writer.writerow([index, index * frame, frame, 12, 36, 4, 100, 2]) def execute(self): manifest = self.root / 'synthetic.json' manifest.write_text(json.dumps(self.data)) return comparison.compare(manifest) def change_csv(self, transform): path = self.root / self.data['runs'][0]['sample_file'] rows = path.read_text().splitlines() path.write_text('\n'.join(transform(rows)) + '\n') def replace_frames(self, frames): path = self.root / self.data['runs'][0]['sample_file'] with path.open('w', newline='') as stream: writer = csv.writer(stream) writer.writerow(comparison.COLUMNS) elapsed = 0 for index, frame in enumerate(frames, 1): elapsed += frame writer.writerow([index, elapsed, frame, 12, 36, 4, 100, 2]) def test_paired_arithmetic_is_explicitly_synthetic_and_not_acceptance(self): report = self.execute() self.assertEqual(report['sample_origin'], 'synthetic') self.assertEqual(report['acceptance'], 'not_evaluated') self.assertEqual(report['aggregate']['baseline']['median_fps'], 50) self.assertEqual(report['aggregate']['successor']['p99_frame_ms'], 10) self.assertEqual(report['median_paired_deltas']['median_frame_reduction_percent'], 50) self.assertEqual(report['runs'][0]['samples_measured'], 3000) self.assertEqual(len(report['manifest_sha256']), 64) def test_percentiles_and_warmup_exclusion(self): self.assertEqual(comparison.percentile(list(range(1, 101)), .99), 99) self.change_csv(lambda rows: [rows[0]] + [row.rsplit(',', 1)[0] + ',999' if i < 1500 else row for i, row in enumerate(rows[1:])]) self.assertEqual(self.execute()['aggregate']['baseline']['median_ai_simulation_ms'], 2) def test_rejects_mismatched_conditions(self): original = copy.deepcopy(self.data) mutations = [('hardware', 'cpu', 'other'), ('workload', 'ai', 35), ('workload', 'route', 'other'), ('workload', 'settings', {'fps_limit': 30}), ('workload', 'warmup_seconds', 31), ('capture', 'environment', 'workbench_headless')] for group, key, value in mutations: with self.subTest(group=group, key=key): self.data = copy.deepcopy(original) self.data['runs'][1][group][key] = value with self.assertRaisesRegex(ValueError, 'mismatched'): self.execute() def test_rejects_missing_pair_and_duplicate_repeat(self): self.data['runs'][-1]['repeat_id'] = 'unpaired' with self.assertRaisesRegex(ValueError, 'Every repeat'): self.execute() self.data['runs'][-1]['repeat_id'] = 'r1' with self.assertRaisesRegex(ValueError, 'Duplicate repeat'): self.execute() def test_rejects_reused_samples(self): self.data['runs'][1]['sample_file'] = self.data['runs'][0]['sample_file'] with self.assertRaisesRegex(ValueError, 'reused'): self.execute() def test_rejects_incomplete_capture(self): self.change_csv(lambda rows: rows[:-1]) with self.assertRaisesRegex(ValueError, 'incomplete recording'): self.execute() def test_rejects_final_stall_hiding_most_of_measurement(self): # Before the coverage check this 1.2-second sample was accepted as a # 60-second measurement with 100 FPS, dropping its 60-second stall. self.replace_frames([20] * 1500 + [10] * 120 + [60000]) with self.assertRaisesRegex(ValueError, 'only 2.000%.*boundary stalls'): self.execute() def test_rejects_warmup_crossing_stall_hiding_measurement(self): # The long frame starts in warm-up and consumes 58.8 measured seconds. self.replace_frames([20] * 1499 + [58820] + [10] * 120) with self.assertRaisesRegex(ValueError, 'only 2.000%.*boundary stalls'): self.execute() def test_short_boundary_crossings_are_visible_with_high_coverage(self): self.replace_frames([17] * 5295) run = self.execute()['runs'][0] self.assertGreaterEqual(run['measurement_coverage_fraction'], .99) self.assertLess(run['measurement_coverage_fraction'], 1) self.assertEqual(len(run['boundary_frames_excluded']), 2) overlap = sum(frame['measurement_overlap_ms'] for frame in run['boundary_frames_excluded']) self.assertAlmostEqual(run['measured_frame_seconds'] * 1000 + overlap, 60000) def test_rejects_coverage_just_below_threshold(self): self.replace_frames([20] * 1500 + [10] * 5939 + [620]) with self.assertRaisesRegex(ValueError, 'only 98.983%'): self.execute() def test_rejects_missing_or_duplicate_frames(self): self.change_csv(lambda rows: rows[:100] + rows[101:]) with self.assertRaisesRegex(ValueError, 'frame index'): self.execute() def test_rejects_population_drift(self): self.change_csv(lambda rows: rows[:1] + [rows[1].replace(',12,36,', ',12,35,')] + rows[2:]) with self.assertRaisesRegex(ValueError, 'ai workload changed'): self.execute() def test_rejects_nonfinite_sample(self): self.change_csv(lambda rows: rows[:1] + [rows[1].rsplit(',', 1)[0] + ',nan'] + rows[2:]) with self.assertRaisesRegex(ValueError, 'invalid frame/time/load'): self.execute() def test_rejects_extra_columns_and_timing_gaps(self): original = (self.root / self.data['runs'][0]['sample_file']).read_text() self.change_csv(lambda rows: [rows[0], rows[1] + ',extra'] + rows[2:]) with self.assertRaisesRegex(ValueError, 'columns'): self.execute() (self.root / self.data['runs'][0]['sample_file']).write_text(original) self.change_csv(lambda rows: [rows[0], rows[1].replace('1,20,20,', '1,30,20,')] + rows[2:]) with self.assertRaisesRegex(ValueError, 'elapsed delta'): self.execute() def test_rejects_incomplete_metadata_and_infinite_json(self): del self.data['runs'][0]['capture']['version'] with self.assertRaisesRegex(ValueError, 'expected fields'): self.execute() self.data['runs'][0]['capture']['version'] = '1' self.data['runs'][0]['workload']['settings']['bad'] = float('inf') with self.assertRaisesRegex(ValueError, 'Non-finite'): self.execute() if __name__ == '__main__': unittest.main()