test/native_journey.py
Ref: Size: 21.1 KiB History
#!/usr/bin/env python3
"""One ordinary native-client journey through panes, transport, fonts, and restore.
The focused acceptance scripts keep destructive races and platform-specific
oracles isolated. This journey owns the shared happy path so those scripts can
start from a lean fixture instead of each re-proving the same window setup.
"""
import os
import shlex
import socket
import subprocess
import sys
import time
sys.dont_write_bytecode = True
from native_fonts import exercise as exercise_fonts, fallback_oracle, font_match
from native_lifecycle import start_workspace
from native_picker import sessions, workflow as picker_workflow
from native_resize import (by_id, clamps, drag_and_keys, modal_leakage,
pixels_and_batch, pty_sizes)
from native_selection import selection_basics
from native_tiling import DEADLINE, colour_counts, eventually, require
from native_wheel import WheelRig, shell_history
def release_binary(rig):
version = subprocess.check_output([rig.muxg, '--version'], text=True)
require('ReleaseSafe' in version or 'ReleaseFast' in version,
'native journey requires a ReleaseSafe or ReleaseFast muxg')
def font_family():
requested = os.environ.get('MUXG_TEST_FONT_FAMILY')
candidates = [requested] if requested else ['CommitMono Nerd Font Mono', 'monospace']
for family in candidates:
match = font_match(family)
exact = family == 'monospace' or match['family'].casefold() == family.casefold()
if exact and fallback_oracle(family):
return family
if requested:
require(font_match(requested)['family'].casefold() == requested.casefold(),
'MUXG_TEST_FONT_FAMILY is not an installed exact family')
return requested
return 'monospace'
def stable_layout(state):
return (state['tab'], [(p['id'], p['label']) for p in state['panes']],
[(d['id'], d['direction']) for d in state['dividers']])
def basic_workspace(rig):
refs = start_workspace(rig, quic=True, picker_checks=True)
panes = list(refs)
left, right, lower = panes
state = rig.state()
require(len(state['dividers']) == 2 and
{d['direction'] for d in state['dividers']} == {'beside', 'stacked'},
'v/b did not produce one below and one beside split')
left_rect, right_rect, lower_rect = [by_id(state)[pane]['outer'] for pane in panes]
require(right_rect['x'] >= left_rect['x'] + left_rect['w'],
'b pane is not beside the original pane')
require(lower_rect['y'] >= right_rect['y'] + right_rect['h'],
'v pane is not below the focused pane')
require(abs(left_rect['w'] - right_rect['w']) <= 1 and
abs(right_rect['h'] - lower_rect['h']) <= 1,
'new splits did not divide the available space equally')
pty_sizes(rig, refs, [(p['id'], p['generation']) for p in state['panes']])
rig.assert_cli_untouched()
rig.focus(right)
rig.chord('h')
rig.wait_state(lambda s: s['focus'] == left)
rig.focus(right)
rig.chord('j')
rig.wait_state(lambda s: s['focus'] == lower)
rig.chord('k')
rig.wait_state(lambda s: s['focus'] == right)
for pane_id, (sock, name) in refs.items():
marker = 'JOURNEY-' + name.upper()
rig.mark(pane_id, sock, name, marker)
for other_sock, other_name in refs.values():
if (other_sock, other_name) != (sock, name):
require(marker not in rig.dump(other_sock, other_name),
'focused input leaked into another pane')
first_sock = refs[left][0]
for flags in (('--next-target', 'box'), ('--next-session', 'right')):
result = subprocess.run([rig.muxg, '--sock', first_sock, *flags], env=rig.env,
capture_output=True, timeout=3)
require(result.returncode == 2, 'retired staged-target arguments were accepted')
rig.focus(right)
rig.shell("stty -echo -icanon -isig min 1 time 0; printf 'RAW-%s\\n' READY; "
"od -An -tx1 -N2; stty sane; printf 'RAW-%s\\n' DONE")
rig.wait_marker(*refs[right], 'RAW-READY')
rig.send('key:prefix', 'key:prefix', 'text:A')
rig.wait_marker(*refs[right], 'RAW-DONE')
require('1c 41' in rig.dump(*refs[right]),
'double prefix did not send one 0x1c before ordinary input')
rig.ok('one window creates below/beside local and QUIC panes, then routes focus and input independently')
return refs
def flood_while_responsive(rig, refs):
left, right, _ = list(refs)
stop = rig.root / 'journey-flood-stop'
progress = rig.root / 'journey-flood-progress'
done = rig.root / 'journey-flood-done'
producer = rig.root / 'journey-flood.sh'
producer.write_text(
'i=0\n'
f'while [ "$i" -lt 4096 ] && [ ! -e {shlex.quote(str(stop))} ]; do\n'
" head -c 1024 /dev/zero | tr '\\000' X\n"
f' i=$((i+1)); echo "$i" > {shlex.quote(str(progress))}; sleep .01\n'
'done\n'
f'touch {shlex.quote(str(done))}\n')
rig.focus(left)
rig.shell('sh ' + shlex.quote(str(producer)))
eventually(lambda: progress.exists() and progress.stat().st_size > 0, 'journey flood did not start')
require(not done.exists(), 'journey flood ended before it could be sampled')
first_count, _ = rig.frames()
first_progress = progress.read_text()
rig.focus(right)
rig.chord('f')
rig.wait_state(lambda s: len(s['panes']) == 1 and s['focus'] == right)
started = time.monotonic()
rig.shell("printf '\\033[38;2;0;0;255mJOURNEY-%s\\033[0m\\n' RESPONSIVE")
rig.wait_marker(*refs[right], 'JOURNEY-RESPONSIVE')
state = rig.wait_state(lambda s: 'JOURNEY-RESPONSIVE' in by_id(s)[right]['painted_text'])
latency = time.monotonic() - started
eventually(lambda: progress.read_text() != first_progress, 'flood output stopped advancing')
time.sleep(.3)
second_count, p99 = rig.frames()
require(not done.exists(), 'journey flood ended during concurrent rendering')
require(colour_counts(rig.pixels(), by_id(state)[right]['content'], 'blue')[0] > 30,
'responsive remote marker did not reach its framebuffer region')
stop.touch()
eventually(done.exists, 'journey flood did not stop')
require(second_count > first_count, 'no frame completed while another pane flooded')
require(latency < DEADLINE and p99 < 20000,
f'pane isolation exceeded budget: {latency}s / {p99}us')
rig.chord('f')
rig.wait_state(lambda s: len(s['panes']) == 3)
rig.ok(f'hidden local flood leaves fullscreen QUIC pane responsive ({latency * 1000:.0f} ms, {p99} us frame p99)')
def fullscreen_pane(rig, refs):
left, right, lower = refs
rig.focus(lower)
before = rig.state()
sizes = {p['id']: (p['rows'], p['cols']) for p in before['panes']}
original_geometry = [(p['id'], p['generation'], p['outer']) for p in before['panes']]
original_dividers = before['dividers']
for key, pane_id in (('f', lower), ('k', right), ('h', left)):
rig.chord(key)
state = rig.wait_state(lambda s: len(s['panes']) == 1 and s['focus'] == pane_id)
pane = state['panes'][0]
require(not state['dividers'] and pane['outer'] ==
dict(x=0, y=0, w=state['width'], h=state['height']),
'fullscreen did not give the focused pane the whole window')
require(pane['generation'] == by_id(before)[pane_id]['generation'],
'fullscreen reattached a pane')
rig.kernel_sizes()
sizes[pane_id] = (pane['rows'], pane['cols'])
for hidden_id, (sock, name) in refs.items():
status = rig.status(sock, name)
require((status['rows'], status['cols']) == sizes[hidden_id],
'fullscreen resized a hidden pane or lost the visible PTY size')
if pane_id == lower:
rig.shell("printf '\\033[0m\\033[2J\\033[H\\033[38;2;0;0;255mFULLSCREEN-%s\\033[0m\\n' INPUT")
rig.wait_marker(*refs[lower], 'FULLSCREEN-INPUT')
eventually(lambda: colour_counts(rig.last_pixels(), pane['content'], 'blue')[0] > 30 and
sum(colour_counts(rig.last_pixels(), pane['content'], 'green')) == 0,
'fullscreen left hidden-pane pixels or failed to paint its input')
require(all('FULLSCREEN-INPUT' not in rig.dump(*refs[p]) for p in (left, right)),
'fullscreen input leaked into a hidden pane')
new_size = (800, 500)
rig.send(f'resize:{new_size[0]}x{new_size[1]}')
rig.wait_state(lambda s: (s['width'], s['height']) == new_size)
rig.kernel_sizes()
rig.chord('f')
restored_new = rig.wait_state(lambda s: len(s['panes']) == 3 and
(s['width'], s['height']) == new_size)
rig.kernel_sizes()
require([(p['id'], p['generation']) for p in restored_new['panes']] ==
[(p['id'], p['generation']) for p in before['panes']] and
[(d['id'], d['direction']) for d in restored_new['dividers']] ==
[(d['id'], d['direction']) for d in original_dividers],
'restoring fullscreen at the new size changed pane or divider identities')
for old, current in zip(original_dividers, restored_new['dividers']):
axis = 'x' if old['direction'] == 'beside' else 'y'
extent = 'w' if old['direction'] == 'beside' else 'h'
old_ratio = (old['rect'][axis] - old['parent'][axis]) / old['parent'][extent]
new_ratio = (current['rect'][axis] - current['parent'][axis]) / current['parent'][extent]
require(abs(old_ratio - new_ratio) < 0.02,
'restoring fullscreen at the new size changed split proportions')
rig.send(f"resize:{before['logical_width']}x{before['logical_height']}")
restored = rig.wait_state(lambda s: len(s['panes']) == 3 and
(s['width'], s['height']) == (before['width'], before['height']))
require([(p['id'], p['generation'], p['outer']) for p in restored['panes']] ==
original_geometry and restored['dividers'] == original_dividers,
'fullscreen changed the saved split geometry after exact original restoration')
pty_sizes(rig, refs, [(p['id'], p['generation']) for p in before['panes']])
rig.chord('f')
rig.chord('r')
rig.wait_state(lambda s: s['resize_mode'] and len(s['panes']) == 3)
rig.key('escape')
rig.ok('fullscreen follows focus, preserves hidden PTYs, and restores weighted splits after window resizing')
def quick_panes(rig, refs):
left, right, lower = list(refs)
# Returning to the remote after a local split must inherit the remote again;
# detached automatic sessions remain alive and must not be joined by accident.
for origin, direction in ((right, 'stacked'), (lower, 'beside'), (right, 'stacked')):
rig.focus(origin)
before = rig.state()
sock, _ = refs[origin]
names = sessions(sock)
other_sock = refs[left if origin == right else right][0]
other_names = sessions(other_sock)
rig.chord('f')
rig.split(direction)
state = rig.wait_state(lambda s: len(s['panes']) == 4 and
all(p['phase'] == 'attached' for p in s['panes']))
require(state['picker'] is None and not state['command_mode'],
'quick split required a picker or stayed in command mode')
pane = by_id(state)[state['focus']]
require(pane['id'] not in refs, 'quick split did not focus its new pane')
name = pane['label'].rsplit('#', 1)[1]
require(name.isdecimal() and name not in names and sessions(sock) == names | {name},
'quick split did not create a fresh numeric session on the focused daemon')
require(sessions(other_sock) == other_names, 'quick split created on the wrong daemon')
source = by_id(state)[origin]
require(pane['label'].rsplit('#', 1)[0] == source['label'].rsplit('#', 1)[0],
'quick split lost its exact connection target')
axis, extent = ('y', 'h') if direction == 'stacked' else ('x', 'w')
require(pane['outer'][axis] >= source['outer'][axis] + source['outer'][extent],
'quick split inserted on the wrong side')
rig.mark(pane['id'], sock, name, 'QUICK-' + str(pane['id']))
rig.chord('f')
rig.chord('d')
state = rig.wait_state(lambda s: len(s['panes']) == 3)
require(stable_layout(state) == stable_layout(before), 'detaching quick pane changed existing layout')
require(name in sessions(sock), 'detaching a quick pane ended its session')
# A split that cannot fit must fail before creating a remote shell.
rig.focus(right)
names = sessions(refs[right][0])
before_fullscreen = rig.state()
original = [(p['id'], p['generation'], p['label']) for p in before_fullscreen['panes']]
original_sessions = {pane_id: sessions(sock) for pane_id, (sock, _) in refs.items()}
rig.chord('f')
fullscreen = rig.wait_state(lambda s: len(s['panes']) == 1 and s['focus'] == right)
require([(p['id'], p['generation'], p['label']) for p in fullscreen['panes']] ==
[next(item for item in original if item[0] == right)],
'fullscreen did not retain the focused identity before undersized insertion')
rig.send('resize:8x8')
rig.wait_state(lambda s: (s['width'], s['height']) == (8, 8))
rig.split('stacked')
state = rig.wait_state(lambda s: bool(s['notice']))
require(state['picker'] is None and len(state['panes']) == 1 and state['focus'] == right and
[(p['id'], p['generation'], p['label']) for p in state['panes']] ==
[next(item for item in original if item[0] == right)] and state['pending'] is None,
'fullscreen undersized quick split changed the visible workspace')
require(sessions(refs[right][0]) == names and
{pane_id: sessions(sock) for pane_id, (sock, _) in refs.items()} == original_sessions,
'fullscreen undersized quick split created or changed a session')
rig.chord('f')
restored = rig.wait_state(lambda s: len(s['panes']) == 3)
require(stable_layout(restored) == stable_layout(before_fullscreen) and
[(p['id'], p['generation'], p['label']) for p in restored['panes']] == original,
'fullscreen undersized quick split changed the underlying workspace')
rig.send('resize:960x600')
rig.wait_state(lambda s: (s['width'], s['height']) == (960, 600))
rig.ok('quick splits inherit local/QUIC targets, auto-name fresh sessions, and reject undersized panes')
def raw_snapshot_and_terminal_peer(rig, refs):
left = next(iter(refs))
sock, name = refs[left]
rig.focus(left)
rig.shell("head -c 4096 /dev/zero | tr '\\000' '\\177'")
rig.shell("printf '\\033[2J\\033[H\\033[38;2;255;0;0mJOURNEY-%s\\177\\033[0m\\n' RENDER")
rig.wait_marker(sock, name, 'JOURNEY-RENDER')
state = rig.wait_state(lambda s: 'JOURNEY-RENDER' in by_id(s)[left]['painted_text'])
eventually(lambda: colour_counts(rig.pixels(), by_id(state)[left]['content'], 'red')[0] > 30,
'DEL-adjacent marker was not rendered before close')
before = stable_layout(state)
rig.chord('f')
rig.wait_state(lambda s: len(s['panes']) == 1)
rig.quit()
for target_sock, session in refs.values():
require(session in sessions(target_sock), 'closing the journey ended a daemon session')
rig.launch_gui([], 'gui-journey-restored')
restored = rig.wait_state(lambda s: len(s['panes']) == 3 and
all(p['phase'] == 'attached' for p in s['panes']))
require(stable_layout(restored) == before, 'reopen changed pane or divider identity')
require('JOURNEY-RENDER' in by_id(restored)[left]['painted_text'],
'reopen lost the DEL-adjacent terminal snapshot')
eventually(lambda: colour_counts(rig.pixels(), by_id(restored)[left]['content'], 'red')[0] > 30,
'reopened framebuffer did not paint the retained marker')
terminal = rig.holder(sock, name)
rig.release_holder(terminal)
rig.wait_state(lambda s: 'HOLDER-ONLINE' in by_id(s)[left]['painted_text'])
rig.ok('close/reopen retains panes and raw terminal text; a terminal client shares the GUI session')
def independent_failures(rig, refs):
left, right, lower = list(refs)
original_sessions = {pane_id: sessions(sock) for pane_id, (sock, _) in refs.items()}
remote_sock, remote_proc = next((sock, proc) for sock, proc in rig.daemons
if sock == refs[right][0])
rig.stop_daemon(remote_sock, remote_proc)
rig.wait_state(lambda s: by_id(s)[right]['phase'] != 'attached')
# Keep the fixture port bound but silent: an unbound UDP port can fail
# immediately via ICMP, which would never exercise pending cancellation.
port = int(rig.targets[remote_sock].rsplit(':', 1)[1])
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as silent:
silent.bind(('127.0.0.1', port))
rig.focus(right)
before_fullscreen = rig.state()
original = [(p['id'], p['generation'], p['label']) for p in before_fullscreen['panes']]
rig.chord('f')
fullscreen = rig.wait_state(lambda s: len(s['panes']) == 1 and s['focus'] == right)
require([(p['id'], p['generation'], p['label']) for p in fullscreen['panes']] ==
[next(item for item in original if item[0] == right)],
'fullscreen did not retain the unavailable pane before opening')
rig.split('stacked')
pending = rig.wait_state(lambda s: s['opening'])
require(len(pending['panes']) == 1 and pending['focus'] == right and
pending['picker'] is None and 'Opening on' in pending['notice'],
'pending remote opening hid the terminal behind a picker')
require(rig.state()['opening'], 'remote opening ended before the cancellation check')
started = time.monotonic()
rig.key('escape')
cancelled = rig.wait_state(lambda s: not s['opening'])
require(time.monotonic() - started < 1 and len(cancelled['panes']) == 1 and
cancelled['focus'] == right and
[(p['id'], p['generation'], p['label']) for p in cancelled['panes']] ==
[next(item for item in original if item[0] == right)] and
cancelled['pending'] is None and
{pane_id: sessions(sock) for pane_id, (sock, _) in refs.items() if pane_id != right} ==
{pane_id: original_sessions[pane_id] for pane_id in refs if pane_id != right},
'cancelling fullscreen remote opening blocked or changed workspace/session state')
rig.chord('f')
restored = rig.wait_state(lambda s: len(s['panes']) == 3)
require(stable_layout(restored) == stable_layout(before_fullscreen) and
[(p['id'], p['generation'], p['label']) for p in restored['panes']] == original,
'cancelling fullscreen remote opening changed the underlying workspace')
rig.focus(lower)
rig.shell('exit 7')
rig.wait_state(lambda s: by_id(s)[lower]['phase'] == 'exited' and
by_id(s)[lower]['exit_code'] == 7)
rig.mark(left, *refs[left], 'JOURNEY-AFTER-EXIT')
require(rig.gui.poll() is None, 'a failed transport or exited neighbour closed the window')
rig.quit()
require('left' in sessions(refs[left][0]), 'final close ended the surviving local session')
rig.ok('remote opening stays responsive and cancels promptly; transport loss and shell exit stay confined')
missing = str(rig.root / 'missing.sock')
rig.launch_gui(['--sock', missing, '--session', 'offline'], 'gui-journey-unavailable', attached=False)
rig.wait_state(lambda s: s['panes'][0]['phase'] in ('dial_failed', 'failed', 'reconnecting'))
rig.send('resize:8x8')
state = rig.wait_state(lambda s: (s['width'], s['height']) == (8, 8))
require(len(state['panes']) == 1 and state['panes'][0]['cols'] > 0 and
state['panes'][0]['rows'] > 0, 'tiny unavailable pane acquired invalid dimensions')
rig.quit()
rig.ok('an initially unavailable target remains representable and closes cleanly')
def main():
require(len(sys.argv) == 3, 'usage: native_journey.py RELEASE_MUX RELEASE_MUXG')
rig = WheelRig(*sys.argv[1:])
rig.test_label = 'journey'
try:
release_binary(rig)
refs = basic_workspace(rig)
picker_workflow(rig, refs)
quick_panes(rig, refs)
selection_basics(rig, refs)
shell_history(rig, list(refs))
state = rig.state()
identities = [(p['id'], p['generation']) for p in state['panes']]
drag_and_keys(rig, identities, refs)
modal_leakage(rig, identities)
clamps(rig, identities, refs)
pixels_and_batch(rig, identities, refs)
fullscreen_pane(rig, refs)
flood_while_responsive(rig, refs)
family = font_family()
exercise_fonts(rig, refs, family=family)
rig.launch_gui([], 'gui-journey-after-fonts')
rig.wait_state(lambda s: len(s['panes']) == 3 and
all(p['phase'] == 'attached' for p in s['panes']))
raw_snapshot_and_terminal_peer(rig, refs)
independent_failures(rig, refs)
print(f'PASS: native journey ({rig.checkpoints} checkpoints)', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
rig.close()
if __name__ == '__main__':
main()