"""Synthetic telemetry and strict comparison arithmetic; no gameplay evidence.""" import copy import csv import json from pathlib import Path import sys import tempfile import unittest sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools')) import ai_pursuit_replay as replay class ReplayTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='err-pursuit-replay-') self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.log = self.root / 'hq.log' def line(self, event, extra, execution=1, session='round-a', role=4): return f'#1 t=1.0s M7 {event} op=1 revision=1 execution={execution} unit=3 role={role} belief=5 {extra} [session={session}]\n' def queued(self, **kwargs): return self.line('OPERATION ORDER QUEUED', 'queued_ms=1000 destination=<1,2,3>', **kwargs) def delivered(self, state='COMPLETED', sequence=1, observed=34000, received=36000, searched=10, contact=0, support=0, until=0, **kwargs): return self.line('EXECUTION DELIVERED', f'sequence={sequence} state={state} observed_ms={observed} received_ms={received} remaining=0 searched={searched} contact={contact} support={support} support_until_ms={until}', **kwargs) def test_rotations_deduplicate_and_separate_sessions(self): self.log.write_text(self.queued() + self.delivered() + self.queued(execution=2) + self.delivered(state='FAILED', searched=0, execution=2)) rotated = self.root / 'old.log' rotated.write_text(self.queued() + self.delivered() + self.queued(session='round-b')) rows = replay.summary([self.log, rotated])['sessions'] self.assertEqual(rows[0]['executions'], 2) self.assertEqual(rows[0]['physical_arrivals_reported'], 1) self.assertEqual(rows[0]['failed_fraction'], .5) self.assertEqual(rows[0]['median_delivery_latency_seconds'], 2) self.assertEqual(rows[0]['median_order_to_physical_arrival_seconds'], 33) self.assertEqual(rows[1]['last_reported_states'], {'UNREPORTED': 1}) def test_blocked_can_recover_and_contact_is_not_physical_arrival(self): self.log.write_text(self.queued() + self.delivered('BLOCKED', observed=2000, received=3000, searched=0) + self.delivered('CONTACT_CHANGED', sequence=2, observed=4000, received=5000, searched=0, contact=1)) row = replay.summary([self.log])['sessions'][0] self.assertEqual(row['physical_arrivals_reported'], 0) self.assertEqual(row['blocked_at_least_once'], 1) self.assertEqual(row['contact_reported_executions'], 1) self.assertEqual(row['failed_fraction'], 0) def test_rotation_orphan_is_explicit(self): self.log.write_text(self.delivered()) row = replay.summary([self.log])['sessions'][0] self.assertEqual(row['reports_missing_original_order'], 1) self.assertIsNone(row['median_order_to_physical_arrival_seconds']) def test_explicit_arrival_flag_handles_contact_transition(self): self.log.write_text(self.queued() + self.delivered('CONTACT_CHANGED', searched=0, contact=1).replace('sequence=1', 'arrived=1 sequence=1')) self.assertEqual(replay.summary([self.log])['sessions'][0]['physical_arrivals_reported'], 1) self.log.write_text(self.delivered('CONTACT_CHANGED', searched=0, contact=1).replace('sequence=1', 'arrived=0 sequence=1')) self.assertEqual(replay.summary([self.log])['sessions'][0]['physical_arrivals_reported'], 0) def test_reject_malformed_conflict_impossible_time_and_oversized_footprint(self): for bad in ('sequence=x', 'observed_ms=nan', 'state=99', 'received_ms=1', 'searched=500', 'support=2'): key = bad.split('=')[0] fields = self.delivered().split() line = ' '.join(bad if item.startswith(key+'=') else item for item in fields) self.log.write_text(self.queued() + line) with self.subTest(bad=bad), self.assertRaises(ValueError): replay.summary([self.log]) self.log.write_text(self.delivered() + self.delivered(received=37000)) with self.assertRaisesRegex(ValueError, 'conflicting duplicate'): replay.summary([self.log]) def test_reject_lineage_changes_and_reversed_sequences(self): for extra in (self.delivered().replace('belief=5', 'belief=6'), self.delivered(sequence=1, observed=34000) + self.delivered(sequence=2, observed=2000)): self.log.write_text(self.queued() + extra) with self.assertRaises(ValueError): replay.summary([self.log]) def test_finite_support_and_numeric_states(self): self.log.write_text(self.queued(role=1) + self.delivered(state='4', role=1, searched=0, contact=1, support=1, until=54000)) self.assertEqual(replay.summary([self.log])['sessions'][0]['actual_support_reported'], 1) self.log.write_text(self.delivered(role=1, searched=0, contact=1, support=1, until=54001)) with self.assertRaisesRegex(ValueError, 'support lease'): replay.summary([self.log]) def test_expired_and_not_applied_are_separate_from_accepted_outcomes(self): expired = '#2 t=200s M0 OPERATION EXPIRED op=1 revision=1 execution=1 unit=3 deadline_ms=120000 released_ms=125000 outcome=unconfirmed [session=round-a]\n' late = '#3 t=201s M8 EXECUTION NOT APPLIED op=1 revision=1 execution=1 unit=3 sequence=2 state=COMPLETED observed_ms=126000 received_ms=127000 [session=round-a]\n' self.log.write_text(self.queued() + self.delivered('TRAVELLING', searched=0) + expired + late) row = replay.summary([self.log])['sessions'][0] self.assertEqual(row['last_reported_states'], {'TRAVELLING': 1}) self.assertEqual(row['unconfirmed_deadline_expiries'], 1) self.assertEqual(row['delivered_packets_not_applied'], 1) self.assertEqual(row['physical_arrivals_reported'], 0) self.assertEqual(row['local_search_footprints_reported'], 0) class ComparisonTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='err-pursuit-comparison-') self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) template = { 'variant': 'baseline', 'repeat_id': '1', 'artifact_id': 'synthetic-baseline', 'addon_ids': ['synthetic'], 'engine_version': 'synthetic', 'hardware': {'cpu': 'synthetic', 'gpu': 'synthetic', 'ram_gib': 16, 'os': 'synthetic', 'driver': 'synthetic'}, 'workload': {'map': 'synthetic', 'route': 'synthetic', 'players': 2, 'ai': 80, 'vehicles': 10, 'entities': 120, 'settings': {'aiDoctrine': 0, 'aiShadow': 0, 'fps_limit': 60}, 'warmup_seconds': 30, 'duration_seconds': 60}, 'capture': {'kind': 'server_tick', 'environment': 'dedicated_server', 'tool': 'synthetic', 'version': '1'}, 'sample_file': 'unset', } self.data = {'schema_version': 1, 'sample_origin': 'synthetic', 'runs': []} for repeat in range(3): for variant, frame in [('baseline', 20), ('successor', 21)]: run = copy.deepcopy(template) run.update(variant=variant, repeat_id=str(repeat), artifact_id=f'synthetic-{variant}', sample_file=f'{variant}-{repeat}.csv') run['workload']['settings']['aiDoctrine'] = int(variant == 'successor') self.data['runs'].append(run) with (self.root / run['sample_file']).open('w', newline='') as stream: writer = csv.writer(stream) writer.writerow(replay.performance.COLUMNS) index = 1 while (index - 1) * frame < 90000: writer.writerow([index, index*frame, frame, 2, 80, 10, 120, 1]) index += 1 def execute(self): path = self.root / 'comparison.json' path.write_text(json.dumps(self.data)) return replay.compare(path) def test_only_doctrine_may_differ_and_synthetic_never_qualifies(self): result = self.execute() self.assertAlmostEqual(result['average_server_fps_regression_percent'], 100/21) self.assertIsNone(result['performance_target_met']) self.assertFalse(result['release_qualified']) def test_measured_flag_only_evaluates_arithmetic_not_release(self): self.data['sample_origin'] = 'measured' # Synthetic test of metadata branch only. result = self.execute() self.assertTrue(result['performance_target_met']) self.assertFalse(result['release_qualified']) def test_mismatched_ai_scenario_settings_or_environment_rejected(self): original = copy.deepcopy(self.data) for key, value in [('ai', 81), ('route', 'different')]: self.data = copy.deepcopy(original) self.data['runs'][1]['workload'][key] = value with self.subTest(key=key), self.assertRaises(ValueError): self.execute() self.data = copy.deepcopy(original) self.data['runs'][1]['workload']['settings']['fps_limit'] = 30 with self.assertRaises(ValueError): self.execute() self.data = copy.deepcopy(original) self.data['runs'][1]['capture']['environment'] = 'workbench_headless' with self.assertRaises(ValueError): self.execute() def test_changed_population_in_raw_sample_rejected(self): path = self.root / self.data['runs'][0]['sample_file'] path.write_text(path.read_text().replace(',80,', ',81,', 1)) with self.assertRaisesRegex(ValueError, 'ai workload changed'): self.execute() def test_partial_or_reused_repeats_rejected(self): self.data['runs'].pop() with self.assertRaisesRegex(ValueError, 'three paired'): self.execute() if __name__ == '__main__': unittest.main()