"""Real-format parser regressions; mutations are synthetic malformed fixtures.""" import hashlib import importlib.util import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[1] TOOL = ROOT/'tools/parse_native_stats.py' spec = importlib.util.spec_from_file_location('parse_native_stats', TOOL) stats = importlib.util.module_from_spec(spec) spec.loader.exec_module(stats) # Exact preserved source line, not a newly measured workload. SOURCE = '08:32:34.010 DEFAULT : FPS: 29.8, frame time (avg: 33.6 ms, min: 3.3 ms, max: 89.1 ms, mean: 33.3 ms, median: 33.3 ms), Mem: 996027 kB, Player: 0, AI: 29, AIChar: 22, Veh: 0 (2), Proj (S: 0, M: 0, G: 0 | 0), Streaming(Dynam: 698, Static: 57591, Disabled: 30 | New: 0, Del: 0, Bump: 16)' VANILLA = '08:40:57.674 DEFAULT : FPS: 29.9, frame time (avg: 33.5 ms, min: 10.4 ms, max: 89.1 ms, mean: 33.3 ms, median: 33.4 ms), Mem: 988737 kB, Player: 0, AI: 0, AIChar: 0, Veh: 0 (0), Proj (S: 0, M: 0, G: 0 | 0), Streaming(Dynam: 509, Static: 57594, Disabled: 21 | New: 0, Del: 0, Bump: 16)' # Exact packed Linux Chat 13:24:56 log line 278; no new benchmark. FREEZES = '13:25:16.186 DEFAULT : FPS: 29.7, frame time (avg: 33.7 ms, min: 3.6 ms, max: 94.8 ms, mean: 32.9 ms, median: 33.3 ms), Mem: 996591 kB, Player: 0, AI: 32, AIChar: 23, Veh: 0 (2), Proj (S: 0, M: 0, G: 0 | 0), Freezes(cnt: 1, max: 90.6 ms, len: 1), Streaming(Dynam: 708, Static: 57591, Disabled: 30 | New: 0, Del: 0, Bump: 16)' class NativeStatsTests(unittest.TestCase): def parse(self, text): return stats.parse_bytes(text.encode(), 'synthetic-test-input') def test_exact_native_fields_raw_line_and_source_hash(self): raw = ('unrelated log\r\n' + SOURCE + '\r\n').encode() result = stats.parse_bytes(raw, 'source-console.log') self.assertEqual(result['source'], {'path': 'source-console.log', 'sha256': hashlib.sha256(raw).hexdigest(), 'bytes': len(raw)}) self.assertEqual(result['record_count'], 1) row = result['records'][0] self.assertEqual(row['line_number'], 2) self.assertEqual(row['raw_line'], SOURCE) self.assertEqual(row['timestamp'], '08:32:34.010') self.assertEqual(row['fps'], 29.8) self.assertEqual(row['frame_time_summary_ms'], {'avg': 33.6, 'min': 3.3, 'max': 89.1, 'mean': 33.3, 'median': 33.3}) self.assertEqual(row['memory_kB'], 996027) self.assertEqual(row['native_populations'], {'Player': 0, 'AI': 29, 'AIChar': 22, 'Veh': {'value': 0, 'parenthesized': 2}}) self.assertEqual(row['native_projectiles'], {'S': 0, 'M': 0, 'G': 0, 'trailing': 0}) self.assertEqual(row['native_streaming'], {'Dynam': 698, 'Static': 57591, 'Disabled': 30, 'New': 0, 'Del': 0, 'Bump': 16}) def test_zero_populations_remain_observed_zero_and_missing_times_null(self): result = self.parse(VANILLA) row = result['records'][0] self.assertEqual(row['native_populations']['AI'], 0) self.assertEqual(row['native_populations']['Player'], 0) self.assertIsNone(row['ai_only_duration_ms']) self.assertIsNone(row['window_duration_ms']) self.assertEqual(result['performance_acceptance'], 'not_evaluated') self.assertIn('percentiles', result['unavailable']) self.assertIn('raw_frame_samples', result['unavailable']) self.assertNotIn('p99', row['frame_time_summary_ms']) self.assertNotIn('frames', row) def test_unrelated_errors_do_not_become_successful_runtime_qualification(self): result = self.parse('08:32:31.000 RESOURCES (E): Failed to load resource\n' + SOURCE) self.assertEqual(result['record_count'], 1) self.assertEqual(result['performance_acceptance'], 'not_evaluated') self.assertTrue(any('does not check engine/script errors' in entry for entry in result['limitations'])) def test_any_malformed_native_record_refuses_whole_export(self): for bad in (SOURCE[:-1], SOURCE.replace('FPS:', 'FPS'), SOURCE.replace('AIChar: 22, ', ''), SOURCE.replace('Mem: 996027 kB', 'Mem: 996027 MB'), SOURCE+' unknown: 1', SOURCE.replace('AI: 29', 'AI: -1'), SOURCE.replace('Player: 0', 'Player: 0.5'), SOURCE.replace('08:32:34.010', '25:32:34.010'), SOURCE.replace('08:32:34.010', '08:61:34.010'), SOURCE.replace('08:32:34.010', '08:32:60.010'), SOURCE.replace('08:32:34.010', '08:32:34'), SOURCE.replace('AI: 29', 'AI: '+str(1 << 63)), SOURCE.replace('min: 3.3', 'min: 99.0'), SOURCE.replace('avg: 33.6', 'avg: 99.0')): with self.subTest(bad=bad), self.assertRaises(stats.StatsError): self.parse(SOURCE+'\n'+bad+'\n'+VANILLA) def test_nonfinite_negative_scientific_and_oversized_decimals_refused(self): for field in ('FPS: 29.8', 'avg: 33.6', 'min: 3.3', 'max: 89.1', 'mean: 33.3', 'median: 33.3'): for invalid in ('nan', 'NaN', 'inf', '-inf', '-1', '1e400', '9'*400): with self.subTest(field=field, invalid=invalid), self.assertRaises(stats.StatsError): self.parse(SOURCE.replace(field, field.split(':')[0]+': '+invalid)) def test_observed_optional_freezes_preserves_values_without_filling_absence(self): raw = (FREEZES+'\n'+SOURCE).encode() parsed = stats.parse_bytes(raw) first, second = parsed['records'] self.assertEqual(first['native_freezes'], {'cnt': 1, 'max_ms': 90.6, 'len': 1}) self.assertEqual(first['raw_line'], FREEZES) self.assertEqual(first['native_populations']['AIChar'], 23) self.assertEqual(first['native_streaming']['Dynam'], 708) self.assertEqual(parsed['source']['sha256'], hashlib.sha256(raw).hexdigest()) self.assertIsNone(second['native_freezes']) self.assertIsNone(first['window_duration_ms']) self.assertIsNone(first['ai_only_duration_ms']) self.assertNotIn('p99', first['frame_time_summary_ms']) # Explicit zero differs from an absent native block; this mutation is synthetic. zeros = FREEZES.replace('Freezes(cnt: 1, max: 90.6 ms, len: 1)', 'Freezes(cnt: 0, max: 0.0 ms, len: 0)') self.assertEqual(self.parse(zeros)['records'][0]['native_freezes'], {'cnt': 0, 'max_ms': 0.0, 'len': 0}) def test_optional_freezes_is_still_strict_and_cannot_hide_malformed_records(self): block = 'Freezes(cnt: 1, max: 90.6 ms, len: 1)' invalid_blocks = ['Freezes(cnt: 1, max: 90.6 ms)', 'Freezes(cnt: 1, max: 90.6 ms, len: 1, extra: 1)', 'Freezes(cnt: 1, max: 90.6 s, len: 1)', block+', '+block] for field, original in [('cnt', '1'), ('max', '90.6'), ('len', '1')]: for value in ('-1', 'NaN', 'inf', '1e400', '9'*400): invalid_blocks.append(block.replace(field+': '+original, field+': '+value)) for field in ('cnt', 'len'): invalid_blocks.append(block.replace(field+': 1', field+': 1.5')) invalid_blocks.append(block.replace(field+': 1', field+': '+str(1 << 63))) for invalid in invalid_blocks: with self.subTest(block=invalid), self.assertRaises(stats.StatsError): self.parse(SOURCE+'\n'+FREEZES.replace(block, invalid)+'\n'+VANILLA) def test_clock_rollover_kept_without_inventing_date_interval_or_windows(self): a = SOURCE.replace('08:32:34.010', '23:59:59.999') b = SOURCE.replace('08:32:34.010', '00:00:05.005') result = self.parse(a+'\n'+b) self.assertEqual([r['timestamp'] for r in result['records']], ['23:59:59.999', '00:00:05.005']) self.assertTrue(all(r['window_duration_ms'] is None for r in result['records'])) self.assertTrue(all('elapsed_ms' not in r for r in result['records'])) def test_bounds_encoding_and_empty_input(self): for raw in (b'', b'No telemetry\n', b'\xff'): with self.assertRaises(stats.StatsError): stats.parse_bytes(raw) with patch.object(stats, 'MAX_INPUT_BYTES', 20), self.assertRaises(stats.StatsError): self.parse(SOURCE) with patch.object(stats, 'MAX_LINE_BYTES', 20), self.assertRaises(stats.StatsError): self.parse(SOURCE) with patch.object(stats, 'MAX_RECORDS', 1), self.assertRaises(stats.StatsError): self.parse(SOURCE+'\n'+VANILLA) self.assertEqual(stats.parse_bytes(b'\xef\xbb\xbf'+SOURCE.encode())['record_count'], 1) def test_cli_read_only_success_and_failure_no_partial_json(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp)/'console.log'; path.write_text(SOURCE+'\n') before = path.read_bytes() process = subprocess.run([sys.executable, str(TOOL), str(path)], capture_output=True, text=True) self.assertEqual(process.returncode, 0, process.stderr) self.assertEqual(json.loads(process.stdout)['record_count'], 1) self.assertEqual(path.read_bytes(), before) path.write_text(SOURCE+'\n'+SOURCE[:-1]) process = subprocess.run([sys.executable, str(TOOL), str(path)], capture_output=True, text=True) self.assertEqual(process.returncode, 1) self.assertEqual(process.stdout, '') self.assertIn('line 2', process.stderr) self.assertEqual(list(Path(tmp).iterdir()), [path]) def test_both_preserved_actual_archives_parse_without_new_measurement(self): for directory, stamp, first in [('dedicated-2026-09-19-083213', '08-32-13', SOURCE), ('dedicated-2026-09-19-084040-vanilla', '08-40-40', VANILLA)]: path = ROOT/'docs/evidence'/directory/'profile/logs'/('logs_2026-09-19_'+stamp)/'console.log' self.assertTrue(path.is_file(), f'missing preserved evidence: {path}') parsed = stats.parse_file(path) self.assertGreater(parsed['record_count'], 1) self.assertEqual(parsed['records'][0]['raw_line'], first) self.assertEqual(parsed['performance_acceptance'], 'not_evaluated') if __name__ == '__main__': unittest.main()