import struct import unittest from witcher_devkit import performance as p class PerformanceTests(unittest.TestCase): def test_nearest_rank_and_tail_gate(self): rows=[dict(interval_ns=i*1000000) for i in range(1,101)] r=p.summarize(rows,[50],60) self.assertEqual((r['median_ms'],r['p95_ms'],r['p99_ms']),(50,95,99)) self.assertFalse(r['tail_estimates_qualified']);self.assertEqual(r['budget_exceeded_frames'],40) def test_spikes_are_strict_and_runs_preserve_duration(self): r=p.summarize([dict(interval_ns=x*1000000) for x in [10,20,21,30,10,40]],[20])['spikes'][0] self.assertEqual(r['frame_count'],3);self.assertEqual(r['episode_count'],2) self.assertEqual(r['total_frame_time_ms'],91);self.assertEqual(r['total_excess_ms'],31);self.assertEqual(r['longest_episode_ms'],51) def trace(self,records,epoch=9,dropped=0): h=[p.MAGIC,1,123,p.CAPACITY,p.RECORD.size,0,len(records),dropped,0,epoch,0,99,0,0,0,0] return p.HEADER.pack(*h)+b''.join(p.RECORD.pack(i,b,b+2,sw,55,epoch,0,1,1) for i,(b,sw) in enumerate(records)) def test_decode_intervals_and_swapchain_boundary(self): h,rows=p.decode(self.trace([(100,1),(150,1),(180,2),(220,2)]),9) self.assertEqual([r['interval_ns'] for r in rows],[None,50,None,40]) def test_corrupt_trace_rejected(self): raw=self.trace([(100,1),(150,1)]) for damaged,epoch in [(raw[:-1],9),(raw,8),(self.trace([(150,1),(100,1)]),9),(b'bad',9)]: with self.subTest(): with self.assertRaises(ValueError):p.decode(damaged,epoch) def test_empty_is_not_zero_timing(self): r=p.summarize([],[],None);self.assertIsNone(r['median_ms']);self.assertIsNone(r['maximum_ms']);self.assertFalse(r['tail_estimates_qualified']) def test_numeric_limits(self): for n in (float('nan'),float('inf'),-1,121): with self.assertRaises(ValueError):p.validate_number(n,.1,120,'seconds') class StopTests(unittest.TestCase): def setUp(self): import tempfile from pathlib import Path from unittest.mock import patch self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup) self.root=Path(self.tmp.name) self.enterContext(patch.object(p.support,'STATE_ROOT',self.root)) self.enterContext(patch.object(p.support,'ARTIFACTS',self.root/'artifacts')) for name,value in [('ROOT',self.root),('ACTIVE',self.root/'active.json'),('LAUNCH',self.root/'launch.json')]: obj=patch.object(p,name,value);obj.start();self.addCleanup(obj.stop) def fixture(self,dropped=0,alive=True): from unittest.mock import patch folder=self.root/'capture';folder.mkdir();buffer=folder/'buffer.bin' buffer.write_bytes(PerformanceTests().trace([(100,1),(200,1)],dropped=dropped)) state=dict(folder=str(folder),buffer=str(buffer),epoch=9,finished=False,label='test',session='session',generation=1,process={'pid':123},seconds=1,thresholds_ms=[25],budget_ms=16.67,graphics={},capability={'clock_alignment_note':'same clock'}) p.write(p.ACTIVE,state);p.write(folder/'capture.json',state);(folder/'markers.jsonl').touch() obj=patch.object(p.support,'process_identity',return_value=state['process'] if alive else None);obj.start();self.addCleanup(obj.stop) return folder def test_stop_exports_prefix_and_is_idempotent(self): folder=self.fixture();r=p.stop();self.assertTrue(r['ok']);self.assertEqual(r['retained_records'],2) self.assertEqual(p.stop(),r);self.assertTrue((folder/'trace.bin').exists());self.assertTrue((folder/'frames.csv').exists()) def test_game_exit_preserves_failed_partial_report(self): folder=self.fixture(alive=False);r=p.stop('game_exit');self.assertFalse(r['ok']);self.assertIn('game_exited_partial_capture',r['errors']);self.assertTrue((folder/'trace.bin').exists()) def test_drops_and_failed_experiment_are_not_success(self): self.fixture(dropped=2);r=p.stop('experiment_failed');self.assertFalse(r['complete']);self.assertEqual(r['dropped_samples'],2);self.assertIn('experiment_failed_partial_capture',r['errors']) def test_restore_refuses_external_changes(self): target=self.root/'user_settings.py';target.write_text('external') p.write(p.LAUNCH,dict(path=str(target),folder=str(self.root),before_exists=False,installed_sha256='wrong',restored=False)) with self.assertRaises(RuntimeError):p.restore() self.assertEqual(target.read_text(),'external') def test_restore_exact_previous_bytes(self): target=self.root/'user_settings.py';target.write_bytes(b'owned') (self.root/'user_settings.before').write_bytes(b'old\r\n') p.write(p.LAUNCH,dict(path=str(target),folder=str(self.root),before_exists=True,installed_sha256=p.digest(target),restored=False)) self.assertTrue(p.restore()['ok']);self.assertEqual(target.read_bytes(),b'old\r\n') def test_markers_limit_and_failed_markers_visible(self): folder=self.fixture();(folder/'marker-errors.jsonl').write_text('{}\n') self.assertIn('marker_recording_failed',p.stop()['errors']) def test_verification_recomputes_binary_csv_and_summaries(self): folder=self.fixture();p.stop();report=folder/'report.json' self.assertTrue(p.verify_report(report)['ok']) original=(folder/'frames.csv').read_text();(folder/'frames.csv').write_text(original.replace('100,102','101,102')) with self.assertRaises(ValueError):p.verify_report(report) def test_comparison_rejects_changed_settings(self): import json folder=self.fixture();report=p.stop();report['window_validity']={'ok':True} report['windows']=[dict(name='gameplay',begin_ns=100,end_ns=200,**p.window_analysis(p.decode((folder/'trace.bin').read_bytes(),9)[1],100,200,[25],16.67))] p.write(folder/'report.json',report) capture=p.read(folder/'capture.json');capture['capability']['build']={'library_sha256':'same'};p.write(folder/'capture.json',capture) p.write(folder/'fixture.json',dict(seconds=20,warmup=3,pause_seconds=0,lab={'data':{'owned':[]}})) import shutil second=self.root/'second';shutil.copytree(folder,second) other=p.read(second/'report.json');other['graphics']={'changed':True};p.write(second/'report.json',other) controls={k:'declared' for k in ('fixture','camera_input_protocol','graphics_frame_cap','capture_settings','warmup_exclusions','observer_cost_assessment')};controls['actor_count']=0 pair={'baseline':str(folder/'report.json'),'treatment':str(second/'report.json')} plan=self.root/'plan.json';p.write(plan,dict(changed_input='none',controls=controls,pairs=[pair,pair])) with self.assertRaisesRegex(ValueError,'controls differ'):p.compare(plan) def test_capture_failure_is_not_reported_as_compile_failure(self): from witcher_devkit import diagnostics self.fixture(alive=False);r=p.stop('game_exit') self.assertEqual(diagnostics.rejected(r).code,'E8002') self.assertEqual(diagnostics.rejected(dict(ok=False,available=False,provider='PathPresent',reason='missing')).code,'E8001') class ReviewCoverageTests(unittest.TestCase): def analyze(self,times,begin,end): _,rows=p.decode(PerformanceTests().trace([(round(t*1e9),1) for t in times]),9) return p.window_analysis(rows,int(begin*1e9),int(end*1e9),[25,50]) def test_review_terminal_crossing_stall_is_visible_and_rejected(self): a=self.analyze([1+i/100 for i in range(2001)]+[42],1,41) self.assertEqual(a['summary']['maximum_ms'],10) self.assertTrue(a['summary']['tail_estimates_qualified']) c=a['coverage'];self.assertEqual(c['completed_fraction'],.5) self.assertEqual(c['known_fraction'],1) self.assertEqual(c['boundary_overlap_ns'],20_000_000_000) self.assertEqual(c['boundary_intervals'][0]['interval_ns'],21_000_000_000) self.assertEqual(c['boundary_intervals'][0]['exceeds_thresholds_ms'],[25,50]) self.assertFalse(c['acceptance_eligible']) def test_start_crossing_stall_and_both_boundaries(self): a=self.analyze([0,21,21.01,21.02],1,21.02) self.assertEqual(a['coverage']['boundary_overlap_ns'],20_000_000_000) self.assertTrue(a['coverage']['boundary_intervals'][0]['crosses_begin']) self.assertFalse(a['coverage']['acceptance_eligible']) c=self.analyze([0,42],1,41)['coverage'] self.assertEqual(c['boundary_overlap_ns'],40_000_000_000) self.assertTrue(c['boundary_intervals'][0]['crosses_begin']) self.assertTrue(c['boundary_intervals'][0]['crosses_end']) def test_no_next_present_is_censored_not_a_fabricated_interval(self): a=self.analyze([1+i/100 for i in range(2001)],1,41);c=a['coverage'] self.assertEqual(a['summary']['maximum_ms'],10) self.assertEqual(c['known_fraction'],.5) self.assertEqual(c['uncovered_trailing_ns'],20_000_000_000) self.assertEqual(c['boundary_intervals'],[]) self.assertTrue(c['censored_gaps'][0]['missing_next_present']) self.assertFalse(c['acceptance_eligible']) def test_leading_and_empty_coverage_are_not_clean(self): c=self.analyze([21,21.01,21.02],1,21.02)['coverage'] self.assertEqual(c['uncovered_leading_ns'],20_000_000_000) self.assertTrue(c['censored_gaps'][0]['missing_previous_present']) c=self.analyze([],1,41)['coverage'] self.assertEqual(c['uncovered_ns'],40_000_000_000) self.assertFalse(c['acceptance_eligible']) def test_normal_boundaries_and_strict_threshold_equality(self): c=self.analyze([0,.025,.05,.075],.005,.07)['coverage'] self.assertTrue(c['acceptance_eligible']);self.assertEqual(c['known_fraction'],1) self.assertEqual(c['boundary_overlap_ns'],40_000_000) def test_swapchain_gap_is_uncovered(self): _,rows=p.decode(PerformanceTests().trace([(0,1),(10,1),(20,2),(30,2)]),9) c=p.window_analysis(rows,0,30,[25])['coverage'] self.assertEqual(c['uncovered_ns'],10);self.assertFalse(c['acceptance_eligible']) class ReviewFinalizationTests(unittest.TestCase): setUp=StopTests.setUp fixture=StopTests.fixture def fill(self,folder): import json line=(json.dumps(dict(monotonic_ns=150,kind='annotation',details={'text':'x'*2000}))+'\n').encode() (folder/'markers.jsonl').write_bytes(line*(p.MARKER_LIMIT//len(line)+1)) def check_finished(self,folder,reason): r=p.read(folder/'report.json') self.assertEqual(r['stop_reason'],reason);self.assertFalse(r['ok']);self.assertFalse(r['complete']) self.assertIn('marker_log_incomplete',r['errors']) self.assertGreater(r['marker_log']['omitted_bytes'],0) self.assertLessEqual(r['marker_log']['parsed_prefix_bytes'],p.MARKER_LIMIT) self.assertLess((folder/'report.json').stat().st_size,3*p.MARKER_LIMIT) self.assertEqual(r['retained_records'],2) self.assertEqual(p.HEADER.unpack_from((folder/'buffer.bin').read_bytes())[5],0) self.assertTrue(p.read(p.ACTIVE)['finished']);self.assertEqual(p.stop(),r) self.assertTrue(p.verify_report(folder/'report.json')['ok']) def test_actual_full_marker_log_manual(self): folder=self.fixture();self.fill(folder);p.stop();self.check_finished(folder,'manual') def test_actual_full_marker_log_deadline_watch(self): folder=self.fixture();self.fill(folder) state=p.read(folder/'capture.json');state['deadline_ns']=0;p.write(folder/'capture.json',state) p.watch(str(folder));self.check_finished(folder,'deadline') def test_oversized_single_marker_rejected_before_write_and_retained(self): folder=self.fixture() with self.assertRaisesRegex(RuntimeError,'marker buffer limit'):p.marker('huge',{'text':'x'*p.MARKER_LIMIT}) self.assertEqual((folder/'markers.jsonl').stat().st_size,0) r=p.stop();self.assertFalse(r['ok']);self.assertTrue(r['marker_log']['overflow_recorded']) class ReviewComparisonTests(unittest.TestCase): setUp=StopTests.setUp fixture=StopTests.fixture def reports(self): import shutil folder=self.fixture();p.stop() self.paths=[] for i in range(4): dest=self.root/('arm'+str(i));shutil.copytree(folder,dest) state=p.read(dest/'capture.json');state.update(folder=str(dest),buffer=str(dest/'buffer.bin'),epoch=9+i,finished=False) state['capability']['build']={'library_sha256':'same'} # Four distinct captures and disjoint windows in the same clock domain. (dest/'buffer.bin').write_bytes(PerformanceTests().trace([(100+i*1000,1),(200+i*1000,1),(300+i*1000,1)],epoch=9+i)) p.write(p.ACTIVE,state);p.write(dest/'capture.json',state) r=p.stop();r['window_validity']={'ok':True} rows=p.decode((dest/'trace.bin').read_bytes(),9+i)[1] r['windows']=[dict(name='gameplay',begin_ns=100+i*1000,end_ns=300+i*1000,**p.window_analysis(rows,100+i*1000,300+i*1000,[25],16.67))] p.write(dest/'report.json',r) p.write(dest/'fixture.json',dict(seconds=20,warmup=3,pause_seconds=0,lab={'data':{'owned':[]}})) self.paths.append(str(dest/'report.json')) controls={k:'declared' for k in ('fixture','camera_input_protocol','graphics_frame_cap','capture_settings','warmup_exclusions','observer_cost_assessment')};controls['actor_count']=0 self.plan=self.root/'plan.json' p.write(self.plan,dict(changed_input='none',controls=controls,pairs=[dict(baseline=self.paths[0],treatment=self.paths[1]),dict(baseline=self.paths[2],treatment=self.paths[3])])) def test_distinct_repeated_trials_accepted(self): self.reports();self.assertTrue(p.compare(self.plan)['ok']) def test_same_path_reuse_rejected(self): self.reports();plan=p.read(self.plan) plan['pairs']=[dict(baseline=self.paths[0],treatment=self.paths[0])]*2;p.write(self.plan,plan) with self.assertRaisesRegex(ValueError,'Reused capture/window'):p.compare(self.plan) def test_copied_report_sidecars_rejected(self): import shutil self.reports();dest=self.root/'copy';shutil.copytree(self.root/'arm0',dest) plan=p.read(self.plan);plan['pairs'][1]['treatment']=str(dest/'report.json');p.write(self.plan,plan) with self.assertRaisesRegex(ValueError,'Reused capture/window'):p.compare(self.plan) def test_same_capture_identity_even_if_raw_changes(self): self.reports();dest=self.root/'arm1';state=p.read(dest/'capture.json');state['epoch']=9;p.write(dest/'capture.json',state) raw=bytearray((dest/'trace.bin').read_bytes());struct.pack_into('