#!/usr/bin/env python3 # MobileGL - tools/trace_replay/test_run_trace_case.py # Copyright (c) 2025-2026 MobileGL-Dev # Licensed under the GNU Lesser General Public License v3.0: # https://www.gnu.org/licenses/gpl-3.0.txt # https://www.gnu.org/licenses/lgpl-3.0.txt # SPDX-License-Identifier: LGPL-3.0-only # End of Source File Header """Canned-log tests of run_trace_case.cmake's split and verify assertions. Each case runs the script exactly as ctest does (`cmake -D... -P run_trace_case.cmake`) with a stub replay that copies canned library logs into the output directory. A bad log must turn the case red with the script's own sentence and leave no split-evidence.txt; a good one must pass and leave evidence that scripts/ci/retrace_split_evidence.py accepts. No build, GPU or trace needed. """ import contextlib import io import os from pathlib import Path import re import shutil import subprocess import sys import tarfile import tempfile import unittest HERE = Path(__file__).resolve().parent SCRIPT = HERE / 'run_trace_case.cmake' CHECKER = HERE.parents[1] / 'scripts' / 'ci' / 'retrace_split_evidence.py' VERIFY_REASON = HERE.parents[1] / 'scripts' / 'ci' / 'retrace_verify_reason.py' sys.path.insert(0, str(CHECKER.parent)) import retrace_split_evidence as checker # noqa: E402 # The stub replay. It refuses to run when the wire counters disagree with what the leg asked for, # so removing the script's MOBILEGL_PIPE_STATS export turns every wire case red, and exporting it # on a leg that did not ask turns that leg red too. STUB = r'''#!/usr/bin/env python3 import json, os, shutil, sys args = sys.argv[1:] out = args[args.index('--output') + 1] split = os.environ.get('MOBILEGL_TRANSPORT', '') not in ('', 'monolith') wire = os.environ.get('MGTRACE_WIRE_EVIDENCE', '') not in ('', '0') stats = os.environ.get('MOBILEGL_PIPE_STATS', '') if wire and split and (stats != '1' or os.environ.get('MOBILEGL_PIPE_STATS_PERIOD') != '1'): sys.stderr.write('stub: MOBILEGL_PIPE_STATS did not reach the replay\n') sys.exit(3) if not wire and stats: sys.stderr.write('stub: MOBILEGL_PIPE_STATS reached a leg that did not ask for it\n') sys.exit(4) canned = os.environ.get('STUB_CANNED_DIR', '') if canned: for name in os.listdir(canned): shutil.copy(os.path.join(canned, name), os.path.join(out, name)) with open(os.path.join(out, 'retrace.log'), 'w') as log: log.write('stub replay\n') if os.environ.get('STUB_ABORT'): sys.exit(134) with open(os.path.join(out, 'result.json'), 'w') as result: json.dump({'passed': True, 'ssim': 1.0}, result) ''' PREFIX = '[12:00:00] [Linux mobilegl_trace_/INFO]: ' MARKER = { 'inproc': 'Config: MOBILEGL_TRANSPORT=inproc - the MGPipe record stream crosses a real ring to an apply thread', 'spawn': 'Config: MOBILEGL_TRANSPORT=spawn - the MGPipe record stream crosses a real ring to an apply ' 'thread in ANOTHER PROCESS', } SPAWN_ARMED = 'MG_Remote client: spawn ARMED - the server role runs in pid 4242, reached at /tmp/x' TCP_ARMED = 'MG_Remote client: control=tcp data=stream server=127.0.0.1:1 pid=4243 dial=connect' SENT = ['P65LinkMetrics kind=frame-sent frame=1 records=40 by_op=2=4,57=1,59=30,67=1', 'P65LinkMetrics kind=frame-sent frame=2 records=31 by_op=57=1,59=28,67=1'] STATS_ON = 'MGPipe stats: counters ON (MOBILEGL_PIPE_STATS), summary every 1 frames' STATS = ['MGPipe stats: frames=1 window=1 draws=30 draws/f=30.00 acc=12 acc/draw=0.40 bytes/f[a=1]', 'MGPipe stats: frames=2 window=1 draws=28 draws/f=28.00 acc=10 acc/draw=0.36 bytes/f[a=1]'] SERVER_START = 'MG_Remote server: apply thread started' def quiet_check(argv): """The checker's exit status, with its ::error:: lines kept off a green run's annotations.""" with contextlib.redirect_stdout(io.StringIO()): return checker.main(argv) def lines(*rows): return ''.join(PREFIX + row + '\n' for row in rows) def good_logs(transport): client = [MARKER['spawn' if transport == 'tcp' else transport]] if transport == 'spawn': client.append(SPAWN_ARMED) if transport == 'tcp': client.append(TCP_ARMED) client += SENT server = [SERVER_START, STATS_ON] + ([] if transport == 'tcp' else STATS) return {'mobilegl.client.log': lines(*client), 'mobilegl.server.log': lines(*server)} def squash(text): return re.sub(r'\s+', ' ', text) @unittest.skipIf(os.name == 'nt', 'the stub replay is a POSIX executable script') class RunTraceCaseTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='run-trace-case-') self.root = Path(self.temp.name) payload = self.root / 't.trace' payload.write_text('not a real trace\n') self.archive = self.root / 't.tar.gz' with tarfile.open(self.archive, 'w:gz') as tar: tar.add(payload, arcname='t.trace') self.stub = self.root / 'stub_replay' self.stub.write_text(STUB) self.stub.chmod(0o755) self.out = self.root / 'case' def tearDown(self): self.temp.cleanup() def run_case(self, env, logs, tcp=False, squashed=True): canned = self.root / 'canned' shutil.rmtree(canned, ignore_errors=True) canned.mkdir() for name, text in logs.items(): (canned / name).write_text(text) environment = {key: value for key, value in os.environ.items() if not key.startswith(('MOBILEGL_', 'MGTRACE_'))} environment.update(env) environment['STUB_CANNED_DIR'] = str(canned) command = ['cmake', f'-DTRACE_REPLAY_EXE={self.stub}', f'-DMOBILEGL_LIBRARY={self.root}/libMobileGL.so', f'-DTRACE_ARCHIVE={self.archive}', '-DTRACE_FILE=t.trace', f'-DTRACE_GOLDEN={self.root}/golden.png', f'-DTRACE_OUTPUT_DIR={self.out}', f'-DTRACE_ARTIFACT_DIR={self.root}/artifacts', '-DTRACE_BACKEND=DirectGLES', '-DTRACE_CASE_NAME=Canned', '-DTRACE_TARGET_CALL=1', '-DTRACE_WIDTH=1', '-DTRACE_HEIGHT=1'] if tcp: command += ['-DTRACE_IPC_CONTROL=tcp://127.0.0.1:1', '-DTRACE_IPC_TOKEN=x'] command += ['-P', str(SCRIPT)] result = subprocess.run(command, env=environment, capture_output=True, text=True) output = result.stdout + result.stderr return result.returncode, squash(output) if squashed else output def split_env(self, transport, wire=False): env = {'MOBILEGL_TRANSPORT': 'spawn' if transport == 'tcp' else transport, 'MOBILEGL_IPC_SERVER_PATH': str(self.root / 'libMobileGLServer.so')} if wire: env['MGTRACE_WIRE_EVIDENCE'] = '1' return env @property def evidence(self): return self.out / 'output' / 'split-evidence.txt' def expect_red(self, env, logs, sentence, tcp=False): rc, output = self.run_case(env, logs, tcp=tcp) self.assertNotEqual(rc, 0, output) self.assertIn(squash(sentence), output) self.assertFalse(self.evidence.exists(), 'a red leg must leave no evidence file') def expect_green(self, env, logs, transport, tcp=False, wire=False): rc, output = self.run_case(env, logs, tcp=tcp) self.assertEqual(rc, 0, output) self.assertEqual(quiet_check([str(self.evidence), '--transport', transport, '--case', 'Canned', '--backend', 'DirectGLES'] + (['--require-wire'] if wire else [])), 0) return output # Red: each assertion of the transport block, fed the log that must trip it. def test_r1_a_library_without_a_transport_writes_no_client_log(self): self.expect_red(self.split_env('inproc'), {'mobilegl.log': lines('Config: loaded')}, 'so there is no evidence the transport ever resolved') def test_r2_the_env_dump_line_is_not_the_marker(self): logs = {'mobilegl.client.log': lines('Config: Accepted env variable: MOBILEGL_TRANSPORT=inproc'), 'mobilegl.server.log': lines(SERVER_START)} self.expect_red(self.split_env('inproc'), logs, 'never reported resolving it') def test_r3_spawn_without_a_separate_server_pid(self): logs = good_logs('spawn') logs['mobilegl.client.log'] = lines(MARKER['spawn'], *SENT) self.expect_red(self.split_env('spawn'), logs, 'spawn ARMED - the server role runs in pid') def test_r4_tcp_without_endpoint_and_pid(self): logs = good_logs('tcp') logs['mobilegl.client.log'] = lines(MARKER['spawn'], *SENT) self.expect_red(self.split_env('tcp'), logs, 'TCP session has no endpoint and server pid arm proof', tcp=True) def test_r5_no_server_log(self): logs = good_logs('inproc') del logs['mobilegl.server.log'] self.expect_red(self.split_env('inproc'), logs, 'but the run wrote no') def test_r6_a_server_fatal(self): logs = good_logs('inproc') logs['mobilegl.server.log'] += lines('MGPipe: Fatal{UnmigratedVerb, "X"}') self.expect_red(self.split_env('inproc'), logs, 'MGPipe Fatal(s) under MOBILEGL_TRANSPORT') def test_r7_a_floor_line(self): logs = good_logs('inproc') logs['mobilegl.server.log'] += lines('MGWIRE-FLOOR unsound-serial-complete serial=3') self.expect_red(self.split_env('inproc'), logs, 'MGWIRE-FLOOR unsound-serial-complete line(s)') def test_r8_no_frame_sent_lines(self): logs = good_logs('inproc') logs['mobilegl.client.log'] = lines(MARKER['inproc']) self.expect_red(self.split_env('inproc', wire=True), logs, 'committed no DrawVbo record to the wire') def test_r9_frames_sent_without_a_draw(self): logs = good_logs('inproc') logs['mobilegl.client.log'] = lines(MARKER['inproc'], 'P65LinkMetrics kind=frame-sent frame=1 ' 'records=2 by_op=57=1,67=1,159=4,590=2') self.expect_red(self.split_env('inproc', wire=True), logs, 'committed no DrawVbo record to the wire') def test_r10_the_server_drew_nothing(self): logs = good_logs('inproc') logs['mobilegl.server.log'] = lines(SERVER_START, STATS_ON, 'MGPipe stats: frames=1 window=1 draws=0 draws/f=0.00 acc=0') self.expect_red(self.split_env('inproc', wire=True), logs, 'backend executed no draw') def test_r11_an_unknown_transport(self): env = self.split_env('inproc') env['MOBILEGL_TRANSPORT'] = 'unix:/x' self.expect_red(env, good_logs('inproc'), 'this runner has no marker for it') def test_r12_wire_evidence_without_a_split_transport(self): self.expect_red({'MGTRACE_WIRE_EVIDENCE': '1', 'MOBILEGL_TRANSPORT': 'monolith'}, {}, 'a run with no wire has no wire volume to count') def test_r13_the_stub_refuses_counters_on_a_leg_that_did_not_ask(self): env = self.split_env('inproc') env['MOBILEGL_PIPE_STATS'] = '1' self.expect_red(env, good_logs('inproc'), 'stub: MOBILEGL_PIPE_STATS reached a leg that did not ask for it') # Green: the same logs as a real leg writes, on every arm. def test_g1_inproc_with_wire(self): output = self.expect_green(self.split_env('inproc', wire=True), good_logs('inproc'), 'inproc', wire=True) self.assertIn('wire_draw_records=58, server_draws=58', output) def test_g1b_inproc_marker_only(self): self.expect_green(self.split_env('inproc'), good_logs('inproc'), 'inproc') values = checker.parse(self.evidence.read_text()) self.assertEqual((values['wire'], values['wire_draw_records']), ('OFF', '-')) self.assertNotEqual(quiet_check([str(self.evidence), '--transport', 'inproc', '--require-wire']), 0) def test_g2_spawn_with_wire(self): self.expect_green(self.split_env('spawn', wire=True), good_logs('spawn'), 'spawn', wire=True) def test_g3_tcp_with_wire(self): self.expect_green(self.split_env('tcp', wire=True), good_logs('tcp'), 'spawn', tcp=True, wire=True) self.assertEqual(checker.parse(self.evidence.read_text())['server_draws'], 'n/a') def test_g4_monolith_leaves_no_evidence(self): rc, output = self.run_case({'MOBILEGL_TRANSPORT': 'monolith'}, {'mobilegl.log': lines('Config: loaded')}) self.assertEqual(rc, 0, output) self.assertFalse(self.evidence.exists()) self.assertNotEqual(quiet_check([str(self.evidence), '--transport', 'inproc']), 0, 'a leg whose transport block was skipped must fail the checker') # MGTRACE_EXPECT_IPC: the client's IPC line must carry every token, whole. IPC = 'Config: IPC ring=8MiB run-ahead={ra} present-credit=1 strict={st} audit=0 role-split-state={rs} x' def ipc_logs(self, ra='1', st='0', rs='1', line=True): logs = good_logs('spawn') if line: logs['mobilegl.client.log'] += lines(self.IPC.format(ra=ra, st=st, rs=rs)) return logs def ipc_env(self, expect='strict=0 role-split-state=1 run-ahead=1'): env = self.split_env('spawn') env.update({'MGTRACE_EXPECT_IPC': expect, 'MGTRACE_EXPECT_IPC_REASON': 'restore the inproc rows'}) return env def test_i1_the_expected_defaults_are_met(self): output = self.expect_green(self.ipc_env(), self.ipc_logs(), 'spawn') self.assertIn('IPC expectation met', output) self.assertEqual(checker.parse(self.evidence.read_text())['ipc_expected'], 'strict=0 role-split-state=1 run-ahead=1') def test_i2_a_changed_default_names_the_reason(self): self.expect_red(self.ipc_env(), self.ipc_logs(rs='0'), 'lacks role-split-state=1') self.expect_red(self.ipc_env(), self.ipc_logs(rs='0'), 'restore the inproc rows') def test_i3_no_ipc_line(self): self.expect_red(self.ipc_env(), self.ipc_logs(line=False), "carries no 'Config: IPC' line") def test_i4_a_token_is_matched_whole(self): self.expect_red(self.ipc_env(), self.ipc_logs(ra='10'), 'lacks run-ahead=1') def test_i5_an_expectation_on_a_monolith_run(self): self.expect_red({'MOBILEGL_TRANSPORT': 'monolith', 'MGTRACE_EXPECT_IPC': 'strict=1'}, {}, 'a monolith run logs no IPC configuration to check') # The verify block, which shares the script. def test_v1_verify_without_the_armed_line(self): self.expect_red({'MOBILEGL_PIPE_VERIFY': '1'}, {'mobilegl.log': lines('Config: loaded')}, 'never reported "MGPipe: verify armed"') def test_v2_verify_with_a_divergence(self): logs = {'mobilegl.log': lines('MGPipe: verify armed', 'MGPipe: Fatal{PipeVerifyDiffer, "GetRenderStateParameters@Clear"}')} self.expect_red({'MOBILEGL_PIPE_VERIFY': '1'}, logs, 'MGPipe Fatal(s) under MOBILEGL_PIPE_VERIFY') def test_v4_an_aborted_verify_run_names_its_divergence(self): # The comparator aborts the replay before result.json, so the script dumps the library log; # the retrace-verify control requires that dump to carry the divergence it caused, through # the same scripts/ci/retrace_verify_reason.py the workflow runs. Another Fatal is not it, # and neither is the comparator's divergence on a field the control did not corrupt. fatal = 'MGPipe: Fatal{PipeVerifyDiffer, "GetRenderStateParameters@Clear"} field=3 where=entry' other = 'MGPipe: Fatal{UnmigratedPipeInput, "GetPixelStoreParameters@ReadPixels"}' other_field = 'MGPipe: Fatal{PipeVerifyDiffer, "GetPixelStoreParameters@ReadPixels"} field=1 where=entry' env = {'MOBILEGL_PIPE_VERIFY': '1', 'STUB_ABORT': '1'} rc, output = self.run_case(env, {'mobilegl.log': lines('MGPipe: verify armed', fatal)}) self.assertNotEqual(rc, 0, output) transcript = self.root / 'control.txt' for log, expected in ((lines('MGPipe: verify armed', fatal), 0), (lines('MGPipe: verify armed'), 1), (lines('MGPipe: verify armed', other), 1), (lines('MGPipe: verify armed', other_field), 1)): _, raw = self.run_case(env, {'mobilegl.log': log}, squashed=False) transcript.write_text(''.join('1: ' + row + '\n' for row in raw.splitlines())) check = subprocess.run([sys.executable, str(VERIFY_REASON), str(transcript)], capture_output=True, text=True) self.assertEqual(check.returncode, expected, raw + check.stdout) def test_v3_verify_green(self): rc, output = self.run_case({'MOBILEGL_PIPE_VERIFY': '1'}, {'mobilegl.log': lines('MGPipe: verify armed')}) self.assertEqual(rc, 0, output) self.assertIn('armed, zero divergences', output) class EvidenceCheckerTests(unittest.TestCase): GOOD = ('case=Canned\nbackend=DirectGLES\ntransport=spawn\ncontrol=-\nmarker=m\n' 'second=spawn ARMED - the server role runs in pid \nserver_log=1\nfatal_lines=0\nfloor_lines=0\n' 'wire=ON\ndraw_opcode=59\nwire_draw_records=58\nserver_draws=58\n') def check(self, text, *args): with tempfile.TemporaryDirectory() as directory: path = Path(directory) / 'split-evidence.txt' if text is not None: path.write_text(text) return quiet_check([str(path)] + list(args)) def test_good_file_passes(self): self.assertEqual(self.check(self.GOOD, '--transport', 'spawn', '--require-wire'), 0) def test_missing_file_fails(self): self.assertEqual(self.check(None, '--transport', 'spawn'), 1) def test_transport_mismatch_fails(self): self.assertEqual(self.check(self.GOOD.replace('transport=spawn', 'transport=inproc'), '--transport', 'spawn'), 1) def test_case_mismatch_fails(self): self.assertEqual(self.check(self.GOOD, '--transport', 'spawn', '--case', 'OpenRA'), 1) def test_zero_wire_records_fail(self): self.assertEqual(self.check(self.GOOD.replace('wire_draw_records=58', 'wire_draw_records=0'), '--transport', 'spawn'), 1) def test_zero_server_draws_fail(self): self.assertEqual(self.check(self.GOOD.replace('server_draws=58', 'server_draws=0'), '--transport', 'spawn'), 1) def test_spawn_without_second_sentence_fails(self): self.assertEqual(self.check(self.GOOD.replace('second=spawn ARMED - the server role runs in pid ', 'second=-'), '--transport', 'spawn'), 1) def test_a_fatal_count_fails(self): self.assertEqual(self.check(self.GOOD.replace('fatal_lines=0', 'fatal_lines=2'), '--transport', 'spawn'), 1) if __name__ == '__main__': unittest.main()