#!/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("Hydra OS: ' + 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')) # Exercise a real menu selection of the second, already running VM. # Its request refreshes only the private compositor/viewers. before_beta_focus = len(focus_events(blog)) manifest = Path(second['config_directory']).parent / 'session.json' with (root / 'selection.log').open('w') as selection_log: selection = subprocess.run(cli + ['_session-helper', 'recover', str(manifest)], input='beta\n', text=True, env=dict(env, WAYLAND_DISPLAY=display), stdout=selection_log, stderr=selection_log, timeout=20) assert selection.returncode == 0, 'menu selection failed' wait_for(lambda: (root / 'snapshot-2.json').exists(), 'selected menu did not refresh compositor', timeout=15) third = json.loads((root / 'snapshot-2.json').read_text()) from xml.etree import ElementTree as ET assert ET.fromstring((root / 'snapshot-2.xml').read_text()).findtext('./desktops/initial') == '2' wait_for(lambda: any(p.name.startswith('wayland-') and not p.name.endswith('.lock') for p in xdg.iterdir()), 'refreshed private display is not ready') 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: len(focus_events(blog)) > before_beta_focus and focus_events(blog)[-1] == 'enter', 'second VM did not gain fresh focus after menu selection') 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(105, blog, alog) intervals = {'beta': measure_hidden_interval(lambda: counter(manager, 'beta'), blog, alog)} switch(106, alog, blog) intervals['alpha'] = measure_hidden_interval(lambda: counter(manager, 'alpha'), alog, blog) 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(third['pid']) / 'cmdline' assert third['config_directory'].encode() in command.read_bytes() os.kill(third['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'], third['pid']], selected_name='beta', selected_workspace=2, 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()