197 lines
11 KiB
Python
197 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise the product's empty-session/recovery/refresh loop privately."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from hydra.runtime import Runtime
|
|
from smoke import (wait_for, counter, focus_events, fresh_focus_transition,
|
|
measure_hidden_interval, classify_compositor_errors)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
|
|
parser.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
|
|
parser.add_argument('--keyboard', required=True, help='compiled dev/input-probe.c')
|
|
args = parser.parse_args()
|
|
for name in ('labwc', 'viewer', 'keyboard'):
|
|
executable = shutil.which(getattr(args, name))
|
|
if not executable:
|
|
parser.error('missing executable: ' + getattr(args, name))
|
|
setattr(args, name, executable)
|
|
if not os.access('/dev/kvm', os.R_OK | os.W_OK):
|
|
parser.error('KVM access required')
|
|
source = Path(__file__).resolve().parents[1]
|
|
root = Path(tempfile.mkdtemp(prefix='hydra-session-refresh-', dir='/tmp'))
|
|
print('Evidence: ' + str(root), flush=True)
|
|
xdg = root / 'xdg'; xdg.mkdir(mode=0o700)
|
|
manager = Runtime(root / 'config', root / 'state', root / 'run')
|
|
cli = [sys.executable, str(source / 'bin/hydra'), '--config', str(manager.config),
|
|
'--state', str(manager.state), '--runtime', str(manager.runtime)]
|
|
subprocess.run(['as', '--32', str(source / 'dev/guest-probe.S'), '-o', str(root / 'boot.o')], check=True)
|
|
subprocess.run(['objcopy', '-j', '.text', '-O', 'binary', str(root / 'boot.o'), str(root / 'boot.bin')], check=True)
|
|
(root / 'boot.raw').write_bytes((root / 'boot.bin').read_bytes() + bytes(8 * 1024 * 1024 - 512))
|
|
|
|
def wrapper(name, code):
|
|
path = root / name
|
|
path.write_text('#!' + sys.executable + '\n' + code)
|
|
path.chmod(0o700)
|
|
return str(path)
|
|
|
|
common = 'import json, os, pathlib, subprocess, sys, time\nroot = pathlib.Path(' + repr(str(root)) + ')\n'
|
|
labwc = wrapper('labwc-private', common + '''
|
|
if '--version' in sys.argv:
|
|
os.execv(@LABWC@, [@LABWC@] + sys.argv[1:])
|
|
config = pathlib.Path(sys.argv[sys.argv.index('-C') + 1])
|
|
number = len(list(root.glob('snapshot-*.json')))
|
|
data = json.loads((config.parent / 'session.json').read_text())
|
|
data.update(pid=os.getpid(), config_directory=str(config))
|
|
(root / ('snapshot-%d.xml' % number)).write_text((config / 'rc.xml').read_text())
|
|
(root / ('snapshot-%d.json' % number)).write_text(json.dumps(data))
|
|
# Test-only wrapper replaces physical backends before the real compositor runs.
|
|
os.environ.update(WLR_BACKENDS='headless', WLR_RENDERER='pixman',
|
|
WLR_XWAYLAND='/bin/false', LABWC_UPDATE_ACTIVATION_ENV='0',
|
|
XDG_DATA_DIRS='/nonexistent')
|
|
os.execv(@LABWC@, [@LABWC@, '-d'] + sys.argv[1:])
|
|
'''.replace('@LABWC@', repr(args.labwc)))
|
|
viewer = wrapper('viewer-private', common + '''
|
|
if '--help' in sys.argv:
|
|
os.execv(VIEWER, [VIEWER] + sys.argv[1:])
|
|
name = sys.argv[sys.argv.index('--title') + 1].split(': ', 1)[1]
|
|
(root / (name + '-viewer.json')).write_text(json.dumps({
|
|
'argv': sys.argv[1:], 'config': os.environ['XDG_CONFIG_HOME'],
|
|
'gdk_backend': os.environ.get('GDK_BACKEND'), 'wayland_display': os.environ['WAYLAND_DISPLAY']}))
|
|
log = os.open(root / (name + '.log'), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
|
|
os.dup2(log, 1); os.dup2(log, 2)
|
|
os.environ['WAYLAND_DEBUG'] = '1'
|
|
os.execv(VIEWER, [VIEWER] + sys.argv[1:])
|
|
'''.replace('VIEWER', repr(args.viewer)))
|
|
terminal = wrapper('terminal-private', common + '''
|
|
manifest = pathlib.Path(sys.argv[-1])
|
|
socket = pathlib.Path(os.environ['XDG_RUNTIME_DIR']) / os.environ['WAYLAND_DISPLAY']
|
|
(root / 'recovery-ready').write_text(str(manifest))
|
|
while not (root / 'recovery-input').exists():
|
|
if not socket.exists():
|
|
sys.exit(0)
|
|
time.sleep(.05)
|
|
if not socket.exists():
|
|
sys.exit(0)
|
|
commands = (root / 'recovery-input').read_text()
|
|
# The actual product recovery helper executes :shell and :refresh.
|
|
with (root / 'recovery.log').open('a') as log:
|
|
sys.exit(subprocess.run(sys.argv[2:], input=commands, text=True, stdout=log, stderr=log).returncode)
|
|
''')
|
|
shell = wrapper('scratch-shell', common + 'sys.path.insert(0, ' + repr(str(source)) + ')\n' + '''
|
|
from hydra.runtime import Runtime
|
|
from smoke import counter, wait_for
|
|
cli = CLI
|
|
manager = Runtime(root / 'config', root / 'state', root / 'run')
|
|
for name in ('alpha', 'beta'):
|
|
subprocess.run(cli + ['add', name, '--memory', '128', '--cpus', '1', '--disk-size', '8M',
|
|
'--graphics', 'software', '--network', 'user'], check=True)
|
|
paths = manager.paths(name); manager.prepare(paths)
|
|
subprocess.run([os.environ['HYDRA_QEMU_IMG'], 'convert', '-f', 'raw', '-O', 'qcow2',
|
|
str(root / 'boot.raw'), str(paths.disk)], check=True)
|
|
subprocess.run(cli + ['start', name], check=True)
|
|
time.sleep(.3)
|
|
identities = {name: {'pid': manager.paths(name).pid.read_text().strip(),
|
|
'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
|
|
for name in ('alpha', 'beta')}
|
|
(root / 'before.json').write_text(json.dumps(identities))
|
|
'''.replace('CLI', repr(cli)).replace('from smoke import', 'from dev.smoke import'))
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), HYDRA_LABWC=labwc,
|
|
HYDRA_VIEWER=viewer, HYDRA_TERMINAL=terminal, HYDRA_BRIGHTNESS=shutil.which('true'),
|
|
SHELL=shell, GSETTINGS_BACKEND='memory')
|
|
for key in ('WAYLAND_DISPLAY', 'DISPLAY', 'LABWC_PID'):
|
|
env.pop(key, None)
|
|
log = (root / 'session.log').open('w')
|
|
launcher, injector = None, None
|
|
result = {'evidence': str(root), 'physical_input_pending': True,
|
|
'terminal': 'test-only command feeder; real recovery helper and scratch :shell',
|
|
'backend': 'test-only wrapper around actual product launch_session and actual labwc'}
|
|
try:
|
|
launcher = subprocess.Popen(cli + ['session'], env=env, stdout=log, stderr=log)
|
|
wait_for(lambda: (root / 'recovery-ready').exists(), 'empty session did not open recovery')
|
|
first = json.loads((root / 'snapshot-0.json').read_text())
|
|
assert first['names'] == [], 'session was not initially empty'
|
|
(root / 'recovery-input').write_text(':shell\n:refresh\n')
|
|
wait_for(lambda: (root / 'snapshot-1.json').exists(), 'product loop did not restart after :refresh', timeout=15)
|
|
second = json.loads((root / 'snapshot-1.json').read_text())
|
|
assert second['names'] == ['alpha', 'beta'], second
|
|
assert first['pid'] != second['pid'], 'compositor was not refreshed'
|
|
logs = {name: root / (name + '.log') for name in ('alpha', 'beta')}
|
|
for name, path in logs.items():
|
|
wait_for(lambda: path.exists() and 'set_title("HydraOS: ' + name + '")' in path.read_text(),
|
|
name + ' product viewer did not connect')
|
|
viewers = {name: json.loads((root / (name + '-viewer.json')).read_text()) for name in logs}
|
|
assert len({data['config'] for data in viewers.values()}) == 2, 'viewer settings were shared'
|
|
for data in viewers.values():
|
|
assert data['gdk_backend'] == 'wayland'
|
|
assert data['config'].startswith(str(manager.runtime) + '/viewer-')
|
|
assert '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0' in data['argv']
|
|
assert '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0' in data['argv']
|
|
alog, blog = logs['alpha'], logs['beta']
|
|
display = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
|
|
keyboard_env = dict(env, WAYLAND_DISPLAY=display)
|
|
injector = subprocess.Popen([args.keyboard], env=keyboard_env, stdin=subprocess.PIPE, text=True)
|
|
time.sleep(.2)
|
|
wait_for(lambda: focus_events(alog) and focus_events(alog)[-1] == 'enter', 'alpha did not focus')
|
|
|
|
def switch(code, source_log, target_log):
|
|
before_source, before_target = focus_events(source_log), focus_events(target_log)
|
|
injector.stdin.write(f'down 29\ndown 56\ndown 42\ndown {code}\nup {code}\nup 42\nup 56\nup 29\nwait 100\n')
|
|
injector.stdin.flush()
|
|
wait_for(lambda: fresh_focus_transition(before_source, before_target,
|
|
focus_events(source_log), focus_events(target_log)), 'fresh reserved focus transition failed')
|
|
|
|
switch(106, alog, blog)
|
|
intervals = {'alpha': measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog)}
|
|
switch(105, blog, alog)
|
|
intervals['beta'] = measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)
|
|
before = json.loads((root / 'before.json').read_text())
|
|
after = {name: {'pid': manager.paths(name).pid.read_text().strip(),
|
|
'uuid': manager.paths(name).uuid, 'counter': counter(manager, name)}
|
|
for name in ('alpha', 'beta')}
|
|
for name in before:
|
|
assert after[name]['pid'] == before[name]['pid'] and after[name]['uuid'] == before[name]['uuid']
|
|
assert after[name]['counter'] != before[name]['counter']
|
|
# Only the PID recorded by our own wrapper, verified against its private
|
|
# config path, may be signalled. Never inherit a host compositor PID.
|
|
command = Path('/proc') / str(second['pid']) / 'cmdline'
|
|
assert second['config_directory'].encode() in command.read_bytes()
|
|
os.kill(second['pid'], signal.SIGTERM)
|
|
assert launcher.wait(timeout=10) == 0, 'product session exit failed'
|
|
for name in before:
|
|
assert manager.status(name)['status'] == 'running'
|
|
assert manager.paths(name).pid.read_text().strip() == before[name]['pid']
|
|
result.update(initial_names=first['names'], refreshed_names=second['names'],
|
|
compositor_pids=[first['pid'], second['pid']], before=before, after=after,
|
|
viewers=viewers,
|
|
hidden_intervals=intervals, survived_session_exit=True,
|
|
expected_xwayland_diagnostics=classify_compositor_errors((root / 'session.log').read_text()),
|
|
passed=True)
|
|
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
|
|
print(json.dumps(result, indent=2))
|
|
finally:
|
|
if injector and injector.poll() is None:
|
|
injector.terminate(); injector.wait(timeout=5)
|
|
if launcher and launcher.poll() is None:
|
|
launcher.send_signal(signal.SIGINT); launcher.wait(timeout=12)
|
|
for profile in manager.profiles():
|
|
if manager.status(profile.name)['status'] == 'running':
|
|
manager.stop(profile.name, force=True)
|
|
log.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|