test/native_lifecycle.py
Ref: Size: 31.0 KiB History
#!/usr/bin/env python3
"""Native lifecycle and persistence checks using real sessions and passive oracles."""
import fcntl
import json
import os
from pathlib import Path
import shlex
import signal
import struct
import subprocess
import sys
import termios
import time
sys.dont_write_bytecode = True
from native_tiling import colour_counts, eventually, require
from native_resize import ResizeRig, by_id, edge
from native_picker import create, end, region, sessions
class LifecycleRig(ResizeRig):
float_size = (960, 600)
float_scale = None
def __init__(self, mux, muxg):
super().__init__(mux, muxg)
self.saved = Path(self.env['XDG_STATE_HOME']) / 'mux/native-workspace.json'
self.cli_layout = self.saved.parent / 'layout'
self.saved.parent.mkdir(exist_ok=True)
self.cli_bytes = b'CLI-LAYOUT-UNTOUCHED\n'
self.cli_layout.write_bytes(self.cli_bytes)
def failure_artifacts(self):
print('Native lifecycle failure artifacts:', self.root, file=sys.stderr)
for path in self.root.glob('gui*.log'):
print(path.name + ':\n' + path.read_text()[-3000:], file=sys.stderr)
def assert_cli_untouched(self):
require(self.cli_layout.read_bytes() == self.cli_bytes,
'native workspace changed terminal mux layout state')
def menu(self, label):
state = self.wait_state(lambda s: s.get('recovery') and
any(row['label'] == label for row in s['recovery']['rows']))
menu = state['recovery']
index = next(i for i, row in enumerate(menu['rows']) if row['label'] == label)
for _ in range(abs(index - menu['selected'])):
self.key('down' if index > menu['selected'] else 'up')
self.key('enter')
def recover(self, label):
self.chord('p')
self.menu(label)
def holder(self, sock, name):
label = 'holder-' + str(time.time_ns())
log = (self.root / (label + '.log')).open('wb')
self.logs.append(log)
proc = subprocess.Popen([self.mux, '--sock', sock, '--session', name],
env=self.env, stdin=subprocess.PIPE, stdout=log, stderr=subprocess.STDOUT)
self.procs.append(proc)
proc.stdin.write(b"printf '\\033[2J\\033[HHOLDER-%s\\n' ONLINE\n")
proc.stdin.flush()
self.wait_marker(sock, name, 'HOLDER-ONLINE')
return proc
def release_holder(self, proc):
proc.stdin.write(b'\x1c\x1c')
proc.stdin.flush()
require(proc.wait(timeout=3) == 0, 'real terminal client failed to detach')
proc.stdin.close()
def saved_json(self):
return json.loads(self.saved.read_text())
def wait_saved(self, previous=None):
def probe():
if not self.saved.exists():
return None
raw = self.saved.read_bytes()
if raw == previous:
return None
return json.loads(raw)
return eventually(probe, 'committed native workspace did not save')
def mark(self, pane_id, sock, session, marker):
self.focus(pane_id)
# The complete marker occurs only in command output, not terminal echo.
left, right = marker.rsplit('-', 1)
self.shell("printf '\\033[?25l\\033[2J\\033[H\\033[38;2;0;255;0m%s-%s\\033[0m\\n' " +
shlex.quote(left) + ' ' + shlex.quote(right))
self.wait_marker(sock, session, marker)
state = self.wait_state(lambda s: marker in by_id(s)[pane_id]['painted_text'])
eventually(lambda: colour_counts(self.last_pixels(), by_id(state)[pane_id]['content'], 'green')[0] > 20,
'terminal marker did not produce retained framebuffer pixels')
def remember_tty(self, pane_id):
self.focus(pane_id)
path = self.root / f'tty-{pane_id}-{time.time_ns()}'
self.shell('tty > ' + shlex.quote(str(path)))
eventually(lambda: path.exists() and path.stat().st_size > 0, 'tty path did not arrive')
self.tty_paths[pane_id] = path.read_text().strip()
def kernel_sizes(self):
state = self.state()
for pane in state['panes']:
expected = (pane['content']['h'] // state['cell_h'],
pane['content']['w'] // state['cell_w'])
require(expected == (pane['rows'], pane['cols']), 'pane cell claim disagrees with geometry')
actual = None
def probe():
nonlocal actual
fd = os.open(self.tty_paths[pane['id']], os.O_RDONLY | os.O_NOCTTY | os.O_NONBLOCK)
try:
actual = struct.unpack('HHHH', fcntl.ioctl(fd, termios.TIOCGWINSZ, bytes(8)))[:2]
return actual == expected
finally:
os.close(fd)
try:
eventually(probe, f"pane {pane['id']} kernel dimensions differ from {expected}")
except AssertionError as error:
raise AssertionError(f'{error}; observed {actual}') from error
def identity(state):
return (state['tab'], state['focus'],
[(p['id'], p['label']) for p in state['panes']],
[(d['id'], d['direction']) for d in state['dividers']])
def durable_structure(document):
# Focus is deliberately omitted: exercising the usable neighbour commits
# a legitimate focus change while an independent host is unavailable.
return (document['version'], document['active_tab_id'], document['next_pane_id'],
[(tab['id'], tab['tree'], tab['panes']) for tab in document['tabs']])
def start_workspace(rig, *, quic=False, picker_checks=False):
"""Build the shared three-pane fixture without re-testing its prerequisites."""
a, _ = rig.daemon('lifecycle-left')
b, _ = rig.daemon('lifecycle-right', quic=quic)
first_target = '--sock ' + a
second_target = rig.targets.get(b, '--sock ' + b)
rig.catalogue([first_target, second_target])
rig.launch_gui([], 'gui-persistent-first', attached=False)
rig.picker('hosts')
if picker_checks:
require(not rig.state()['panes'], 'first launch created a shell before explicit selection')
rig.choose(first_target)
rig.picker('sessions')
rig.new_session('left')
rig.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'attached')
if picker_checks:
rig.open_picker('beside')
armed = rig.picker('hosts')
require(armed['pending']['direction'] == 'beside',
'b did not arm a split beside the focused pane')
stats = rig.command('d', 'stats', '--sock', b).stdout
require('attaches=0' in stats, 'picker attached its target before insertion: ' + stats)
require(rig.command('a', 'status', '--sock', b, '--session', 'right',
'--timeout', '200', check=False).returncode != 0,
'picker created its named session before insertion')
rig.key('escape')
cancelled = rig.wait_state(lambda s: s['picker'] is None)
require(len(cancelled['panes']) == 1 and cancelled['pending'] == armed['pending'],
'closing the host picker changed layout or forgot the armed split')
rig.key('escape')
rig.wait_state(lambda s: s['pending'] is None)
rig.host(second_target, 'beside')
rig.new_session('right')
before_split = rig.wait_state(lambda s: len(s['panes']) == 2 and
all(p['phase'] == 'attached' for p in s['panes']))
if picker_checks:
rig.mark(before_split['panes'][1]['id'], b, 'right', 'PICKER-RIGHT')
rig.open_picker('stacked')
armed = rig.picker('hosts')
require(armed['pending']['direction'] == 'stacked',
'v did not arm a split below the focused pane')
rig.key('escape')
rig.key('escape')
rig.wait_state(lambda s: s['pending'] is None)
rig.host(first_target, 'stacked')
rig.choose('0')
state = rig.wait_state(lambda s: len(s['panes']) == 3 and
all(p['phase'] == 'attached' for p in s['panes']))
if picker_checks:
require(state['panes'][0]['outer'] == before_split['panes'][0]['outer'],
'nested insertion rebalanced the unrelated pane')
rig.shell("printf '\\033[2J\\033[H\\033[38;2;255;0;0mPICKER-%s\\033[0m\\n' EXISTING")
rig.wait_marker(a, '0', 'PICKER-EXISTING')
def inserted_pixels():
pixels = rig.last_pixels()
green = colour_counts(pixels, state['panes'][1]['content'], 'green')
red = colour_counts(pixels, state['panes'][2]['content'], 'red')
return green[0] > 30 and red[0] > 30 and green[1] == red[1] == 0
eventually(inserted_pixels, 'nested insertion erased or misplaced session glyphs')
refs = {state['panes'][0]['id']: (a, 'left'), state['panes'][1]['id']: (b, 'right'),
state['panes'][2]['id']: (a, '0')}
rig.divider_ids = [(d['id'], d['direction']) for d in state['dividers']]
for pane_id in refs:
rig.remember_tty(pane_id)
rig.wait_saved()
expected_ids = set(refs)
eventually(lambda: {p['id'] for p in rig.saved_json()['tabs'][0]['panes']} == expected_ids,
'saved workspace did not converge on all three fixture panes')
return refs
def start_persistent(rig):
"""Build the persistence fixture and give it observable, non-default state."""
refs = start_workspace(rig)
state = rig.state()
for pane_id, (sock, name) in refs.items():
rig.mark(pane_id, sock, name, 'PERSIST-' + name)
rig.drag('beside', dx=state['cell_w'] * 7)
rig.drag('stacked', dy=-state['cell_h'] * 4)
rig.focus(state['panes'][1]['id'])
rig.kernel_sizes()
rig.wait_saved()
rig.assert_cli_untouched()
return refs
def single_writer(rig):
env = rig.env.copy()
env.pop('MUXG_TEST_FIFO', None)
second = rig.spawn([rig.muxg], 'gui-second-writer', env)
require(second.wait(timeout=3) != 0, 'second default workspace writer was accepted')
diagnostic = (rig.root / 'gui-second-writer.log').read_text().lower()
require('already' in diagnostic and 'open' in diagnostic, 'writer refusal lacked an actionable reason')
require(rig.gui.poll() is None, 'second writer disturbed the first window')
rig.ok('a second persistent writer is refused while the first remains usable')
def save_failure(rig):
before = rig.saved.read_bytes()
mode = rig.saved.parent.stat().st_mode & 0o777
rig.saved.parent.chmod(0o555)
try:
require(not os.access(rig.saved.parent, os.W_OK), 'test UID can bypass the save-failure fixture')
state = rig.state()
rig.drag('beside', dx=state['cell_w'] * -3)
rig.wait_state(lambda s: abs(edge(s, 'beside') - edge(state, 'beside')) >= state['cell_w'] * 2)
rig.wait_state(lambda s: 'sav' in s.get('notice', '').lower() and
any(word in s['notice'].lower() for word in ('fail', 'error', 'cannot', 'denied')))
require(rig.saved.read_bytes() == before, 'failed save damaged the previous workspace bytes')
rig.kernel_sizes()
finally:
rig.saved.parent.chmod(mode)
rig.drag('beside', dx=rig.state()['cell_w'])
rig.wait_saved(before)
rig.assert_cli_untouched()
rig.ok('real filesystem save failure preserves prior bytes and reports the error without blocking panes')
def temporary_and_roundtrip(rig, refs):
before = rig.state()
tree = rig.saved_json()['tabs'][0]['tree']
rig.send('resize:8x8')
rig.wait_state(lambda s: (s['logical_width'], s['logical_height']) == (8, 8))
rig.kernel_sizes()
rig.quit()
saved = rig.saved.read_bytes()
document = json.loads(saved)
require(document['tabs'][0]['tree'] == tree, 'saving a clamped window rewrote relative split weights')
for sock, name in refs.values():
require(name in sessions(sock), 'window close ended a saved daemon session')
# A saved target must remain usable after being removed from discovery.
rig.catalogue([])
sock, name = next(iter(refs.values()))
rig.launch_gui(['--sock', sock, '--session', name], 'gui-explicit-temporary')
rig.send('resize:740x440')
rig.wait_state(lambda s: s['logical_width'] == 740)
rig.quit()
require(rig.saved.read_bytes() == saved, 'explicit target overwrote the default workspace')
if os.environ.get('MUXG_LIFECYCLE_RESTORE_WAYLAND') == '1':
require(os.environ.get('MUXG_TEST_SWAY_FLOAT') == '1', 'density restore needs its own floating fixture')
rig.env['SDL_VIDEO_DRIVER'] = 'wayland'
rig.launch_gui([], 'gui-restored')
restored = rig.state()
if os.environ.get('MUXG_LIFECYCLE_RESTORE_WAYLAND') == '1':
require(before['width'] == before['logical_width'] and
restored['width'] == restored['logical_width'] * 2,
'restore did not transition from 100% offscreen to actual 200% Wayland')
require(identity(restored) == identity(before), 'restore lost stable IDs, targets, focus or split ancestry')
require(rig.saved_json() == document, 'restore rewrote persisted split weights or identities')
rig.kernel_sizes()
for pane_id, (sock, name) in refs.items():
# Tiny PTYs may legitimately reflow old markers into history. Compare
# the actual current daemon grid before sending any new terminal input.
def normalized(text):
return '\n'.join(line.rstrip() for line in text.splitlines()).rstrip('\n')
expected = normalized(rig.dump(sock, name))
rig.wait_state(lambda s: by_id(s)[pane_id]['painted_seq'] > 0 and
normalized(by_id(s)[pane_id]['painted_text']) == expected)
rig.mark(pane_id, sock, name, 'RESTORED-' + name)
rig.assert_cli_untouched()
rig.ok('close/reopen restores nested targets, identity, focus, weights and PTYs independently of the catalogue')
rig.ok('explicit target remains temporary and neither mode writes terminal layout state')
def preservation_case(mux, muxg, label, original, unreadable=False):
rig = LifecycleRig(mux, muxg)
try:
sock, _ = rig.daemon(label + '-recovery')
rig.catalogue(['--sock ' + sock])
rig.saved.write_bytes(original)
if unreadable:
rig.saved.chmod(0)
require(not os.access(rig.saved, os.R_OK), 'test UID can bypass unreadable-file fixture')
rig.launch_gui([], 'gui-' + label, attached=False)
rig.wait_state(lambda s: 'sav' in s.get('notice', '').lower() or
'preserv' in s.get('notice', '').lower())
# Even after permissions recover, this run must remain sealed against
# replacing the original file with the fallback workspace.
rig.saved.chmod(0o600)
rig.picker('hosts')
rig.choose('--sock ' + sock)
rig.picker('sessions')
rig.new_session('unsaved')
state = rig.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'attached')
rig.mark(state['panes'][0]['id'], sock, 'unsaved', label.upper() + '-USABLE')
rig.quit()
require(rig.saved.read_bytes() == original, label + ' state was overwritten')
require('unsaved' in sessions(sock), 'preservation fallback ended its live session on close')
rig.assert_cli_untouched()
print(f'lifecycle OK: {label} original bytes preserved', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
if rig.saved.exists():
rig.saved.chmod(0o600)
rig.close()
def relative_target_reopen(mux, muxg):
rig = LifecycleRig(mux, muxg)
try:
sock, _ = rig.daemon('relative')
rig.gui_cwd = rig.root
rig.catalogue(['--sock relative.sock'])
rig.launch_gui([], 'gui-relative-original', attached=False)
rig.picker('hosts')
rig.choose('--sock relative.sock')
rig.picker('sessions')
rig.new_session('relative-kept')
state = rig.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'attached')
pane_id = state['panes'][0]['id']
rig.mark(pane_id, sock, 'relative-kept', 'RELATIVE-ORIGINAL')
rig.quit()
saved = rig.saved_json()
require(saved['tabs'][0]['panes'][0]['target'] == {'sock': sock},
'saved relative transport was not anchored to its original working directory')
rig.gui_cwd = rig.root / 'different-working-directory'
rig.gui_cwd.mkdir()
rig.catalogue([])
rig.launch_gui([], 'gui-relative-restored')
require(rig.state()['panes'][0]['id'] == pane_id, 'relative target restore replaced its pane ID')
rig.mark(pane_id, sock, 'relative-kept', 'DIFFERENT-CWD')
rig.quit()
rig.assert_cli_untouched()
require('relative-kept' in sessions(sock), 'relative target session did not survive reopen')
print('lifecycle OK: relative socket survives actual changed-cwd reopen', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
rig.close()
def lifecycle_actions(rig, refs):
left, right, lower = list(refs)
right_sock, right_name = refs[right]
rig.focus(right)
holder = rig.holder(right_sock, right_name)
rig.chord('x')
state = rig.wait_state(lambda s: s.get('recovery') and
any(row['label'] == 'End for all clients' for row in s['recovery']['rows']))
require(state['recovery']['rows'][state['recovery']['selected']]['label'] == 'Cancel',
'shared-session force confirmation did not default to Cancel')
require(right in by_id(state) and right_name in sessions(right_sock),
'daemon End refusal removed the pane or shared session')
rig.key('escape')
require(holder.poll() is None, 'refused End disconnected another terminal client')
rig.release_holder(holder)
rig.chord('x')
rig.wait_state(lambda s: right not in by_id(s))
eventually(lambda: right_name not in sessions(right_sock), 'accepted End left its daemon session live')
rig.ok('End respects an actual second terminal client and succeeds when the GUI is the sole attachment')
rig.focus(left)
rig.chord('d')
state = rig.wait_state(lambda s: len(s['panes']) == 1 and left not in by_id(s))
sock, name = refs[left]
require(name in sessions(sock), 'Detach terminated its daemon session')
require(state['panes'][0]['outer'] == {'x': 0, 'y': 0, 'w': state['width'], 'h': state['height']},
'remaining sibling did not fill the detached pane area')
rig.ok('Detach removes only its leaf and retains the independently observed daemon session')
rig.shell('exit 7')
state = rig.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'exited')
require(state['panes'][0]['id'] == lower and state['panes'][0]['exit_code'] == 7,
'ordinary exit lost its stable pane or exit status')
require(rig.gui.poll() is None, 'a sole exited pane closed the GUI')
generation = state['panes'][0]['generation']
rig.recover('Choose session')
rig.picker('hosts')
rig.choose('--sock ' + sock)
rig.picker('sessions')
rig.choose(name)
state = rig.wait_state(lambda s: len(s['panes']) == 1 and s['panes'][0]['phase'] == 'attached')
require(state['panes'][0]['id'] == lower and state['panes'][0]['generation'] > generation,
'Choose session failed to replace the exited attachment in place')
rig.mark(lower, sock, name, 'RECOVERY-EXISTING')
rig.ok('ordinary shell exit stays visible and Choose session replaces its attachment in place')
holder = rig.holder(sock, name)
rig.chord('x')
rig.menu('End for all clients')
rig.wait_state(lambda s: not s['panes'])
eventually(lambda: name not in sessions(sock), 'explicit shared End left the session live')
holder.wait(timeout=3)
holder.stdin.close()
require(rig.gui.poll() is None, 'ending the last pane closed the empty workspace')
rig.chord('enter')
rig.picker('hosts')
rig.assert_cli_untouched()
rig.ok('explicit force ends the shared session; an empty tab remains available for adding a pane')
def recovery_workflow(mux, muxg):
rig = LifecycleRig(mux, muxg)
stopped = None
try:
refs = start_persistent(rig)
left, right, lower = list(refs)
a, missing = refs[lower]
b, right_name = refs[right]
rig.quit()
before = rig.saved.read_bytes()
end(a, missing)
create(a, 'replacement')
stopped = next(proc for sock, proc in rig.daemons if sock == b)
os.kill(stopped.pid, signal.SIGSTOP)
rig.catalogue([])
rig.launch_gui([], 'gui-restore-unavailable', attached=False)
rig.wait_state(lambda s: len(s['panes']) == 3 and by_id(s)[left]['phase'] == 'attached')
rig.mark(left, *refs[left], 'RESTORE-INDEPENDENT')
require(missing not in sessions(a), 'restore silently recreated a missing saved session')
started = time.monotonic()
rig.quit()
require(time.monotonic() - started < 3, 'shutdown waited for an unavailable restore attachment')
require(durable_structure(rig.saved_json()) == durable_structure(json.loads(before)),
'unavailable restore rewrote saved target identities')
rig.ok('an unavailable restore does not block another pane or bounded window shutdown')
rig.launch_gui([], 'gui-recovery-in-place', attached=False)
rig.wait_state(lambda s: len(s['panes']) == 3 and by_id(s)[left]['phase'] == 'attached')
rig.focus(lower)
rig.recover('Retry')
rig.wait_state(lambda s: by_id(s)[lower]['phase'] in ('failed', 'refused', 'exited', 'dial_failed'))
require(missing not in sessions(a), 'Retry recreated the missing saved session')
rig.focus(right)
state = rig.state()
generation = by_id(state)[right]['generation']
rig.recover('Choose session')
# Discovery was deliberately cleared; add only an explicit recovery
# target. Restoration itself already demonstrated catalogue independence.
rig.choose('Add host...')
rig.send('text:--sock ' + a, 'key:enter')
rig.picker('sessions')
rig.choose('replacement')
state = rig.wait_state(lambda s: by_id(s)[right]['phase'] == 'attached')
require(by_id(state)[right]['generation'] > generation, 'retarget reused the cancelled attachment generation')
new_generation = by_id(state)[right]['generation']
rig.mark(right, a, 'replacement', 'RETARGET-CURRENT')
os.kill(stopped.pid, signal.SIGCONT)
stopped = None
require(right_name in sessions(b), 'cancelling the old attachment terminated its daemon session')
# Delayed old transport activity must not replace the committed target.
deadline = time.monotonic() + .3
while time.monotonic() < deadline:
state = rig.state()
pane = by_id(state)[right]
require(pane['generation'] == new_generation and pane['label'].endswith('#replacement'),
'a cancelled attachment changed its replacement pane')
rig.focus(right)
rig.recover('Choose session')
rig.choose('Add host...')
rig.send('text:--sock ' + b, 'key:enter')
rig.picker('sessions')
rig.choose(right_name)
rig.wait_state(lambda s: by_id(s)[right]['phase'] == 'attached')
rig.mark(right, b, right_name, 'RETURNED-HOST')
rig.assert_cli_untouched()
rig.ok('missing sessions stay missing; in-place recovery cancels old work and a returned host is usable')
rig.quit()
except BaseException:
rig.failure_artifacts()
raise
finally:
if stopped is not None:
os.kill(stopped.pid, signal.SIGCONT)
rig.close()
def delayed_end_transitions(mux, muxg):
# Stop only our real daemon to hold its reply across ordinary GUI input.
# A second real client supplies the daemon's refusal; no wire reply is forged.
for transition in ('command', 'focus', 'picker', 'resize', 'drag', 'replacement', 'detach', 'accepted'):
rig = LifecycleRig(mux, muxg)
paused = None
try:
a, daemon = rig.daemon('delayed-end')
b, _ = rig.daemon('healthy-neighbour')
rig.start_gui(a, b, 'gui-delayed-' + transition)
rig.host('--sock ' + b, 'beside')
rig.choose('0')
state = rig.wait_state(lambda s: len(s['panes']) == 2 and
all(p['phase'] == 'attached' for p in s['panes']))
origin, neighbour = [p['id'] for p in state['panes']]
generation = by_id(state)[origin]['generation']
holder = None if transition == 'accepted' else rig.holder(a, 'left')
rig.focus(origin)
# A delayed reply makes any transient progress popup observable.
# Compare the quiet centre of the actual framebuffer, away from
# pane headers and shell cursors, as well as modal state.
state = rig.state()
centre = {'x': state['width'] // 2 - 40, 'y': state['height'] // 2 - 30,
'w': 80, 'h': 60}
before_pixels = region(rig.pixels(), centre)
daemon.send_signal(signal.SIGSTOP)
paused = daemon
rig.chord('x')
state = rig.wait_state(lambda s: s['pending_end'] is not None)
require(state['recovery'] is None and state['picker'] is None,
'pending End opened a progress popup')
require(state['pending_end']['key']['pane'] == origin,
'pending End lost its original target')
require(region(rig.pixels(), centre) == before_pixels,
'pending End painted a popup over the workspace')
if transition in ('focus', 'accepted'):
rig.focus(neighbour)
elif transition == 'command':
rig.key('prefix')
rig.wait_state(lambda s: s['command_mode'])
elif transition == 'picker':
rig.chord('enter')
rig.picker('hosts')
elif transition == 'resize':
rig.chord('r')
rig.wait_state(lambda s: s['resize_mode'])
elif transition == 'drag':
state = rig.state()
rect = state['dividers'][0]['rect']
point = rig.point(state, rect['x'] + rect['w'] / 2,
rect['y'] + rect['h'] / 4)
rig.send('mousedown:' + point)
rig.wait_state(lambda s: s['drag'] is not None)
elif transition == 'replacement':
rig.recover('Choose session')
state = rig.wait_state(lambda s: 'pending End' in s['notice'])
require(state['picker'] is None and by_id(state)[origin]['generation'] == generation,
'replacement started while its End outcome was pending')
# Leave the recovery menu open while the refusal arrives.
rig.chord('p')
rig.wait_state(lambda s: s['recovery'] is not None)
elif transition == 'detach':
rig.chord('d')
rig.wait_state(lambda s: origin not in by_id(s))
daemon.send_signal(signal.SIGCONT)
paused = None
state = rig.wait_state(lambda s: s['pending_end'] is None)
if transition == 'accepted':
state = rig.wait_state(lambda s: origin not in by_id(s))
eventually(lambda: 'left' not in sessions(a), 'accepted End left its original session alive')
require(state['focus'] == neighbour and neighbour in by_id(state),
'late accepted End removed or focused the wrong pane')
else:
require('left' in sessions(a) and holder.poll() is None,
'delayed shared-session End bypassed the daemon refusal')
if transition != 'detach':
require(state['notice'] == 'others attached',
'expected the real daemon refusal, not a timeout or unknown outcome')
if transition == 'replacement':
require(state['recovery'] is not None and
all(row['label'] != 'End for all clients' for row in state['recovery']['rows']),
'late refusal replaced the active recovery menu')
else:
require(state['recovery'] is None, 'late refusal installed a force menu in another context')
if transition == 'command':
require(state['command_mode'], 'late refusal interrupted command mode')
elif transition == 'focus':
require(state['focus'] == neighbour, 'late refusal stole focus from the neighbour')
elif transition == 'picker':
require(state['picker'] is not None, 'late refusal dismissed the picker')
elif transition == 'resize':
require(state['resize_mode'], 'late refusal interrupted resize mode')
elif transition == 'drag':
require(state['drag'] is not None, 'late refusal interrupted the active drag')
rig.send('mouseup:' + point)
if transition in ('command', 'picker', 'resize', 'replacement'):
rig.key('escape')
if transition == 'replacement':
rig.recover('Choose session')
rig.picker('hosts')
rig.choose('--sock ' + b)
rig.picker('sessions')
rig.choose('0')
state = rig.wait_state(lambda s: by_id(s)[origin]['generation'] > generation and
by_id(s)[origin]['phase'] == 'attached')
require(state['recovery'] is None, 'old End refusal leaked into replacement')
rig.release_holder(holder)
rig.mark(neighbour, b, '0', 'DELAYED-END-' + transition.upper())
rig.assert_cli_untouched()
rig.quit()
rig.ok('real delayed End preserves target and GUI context through ' + transition)
except BaseException:
rig.failure_artifacts()
raise
finally:
if paused is not None and paused.poll() is None:
paused.send_signal(signal.SIGCONT)
rig.close()
def main():
require(len(sys.argv) == 3, 'usage: native_lifecycle.py MUX MUXG')
mux, muxg = sys.argv[1:]
rig = LifecycleRig(mux, muxg)
try:
refs = start_persistent(rig)
single_writer(rig)
save_failure(rig)
temporary_and_roundtrip(rig, refs)
lifecycle_actions(rig, refs)
rig.quit()
valid_bytes = rig.saved.read_bytes()
print(f'native lifecycle main OK ({rig.checkpoints} checkpoints)', flush=True)
except BaseException:
rig.failure_artifacts()
raise
finally:
rig.close()
recovery_workflow(mux, muxg)
preservation_case(mux, muxg, 'malformed', b'{"version": 1, "tabs": [ BROKEN ORIGINAL\n')
preservation_case(mux, muxg, 'unreadable', valid_bytes, unreadable=True)
relative_target_reopen(mux, muxg)
delayed_end_transitions(mux, muxg)
if __name__ == '__main__':
main()