107 lines
4.7 KiB
Python
107 lines
4.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Disposable headless GUI checks. Never connects to an inherited display."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from hydra.profiles import Profile
|
|
from hydra.runtime import Runtime
|
|
from hydra.session import build_labwc_config, build_viewer_argv, _viewer_environment
|
|
|
|
|
|
def wait_for(predicate, description, timeout=8):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return
|
|
time.sleep(.05)
|
|
raise AssertionError(description)
|
|
|
|
|
|
def main():
|
|
cli = argparse.ArgumentParser(description=__doc__)
|
|
cli.add_argument('--viewer', default=os.environ.get('HYDRA_VIEWER', 'remote-viewer'))
|
|
cli.add_argument('--labwc', default=os.environ.get('HYDRA_LABWC', 'labwc'))
|
|
cli.add_argument('--keyboard', default='wtype')
|
|
args = cli.parse_args()
|
|
for name in ('viewer', 'labwc', 'keyboard'):
|
|
value = shutil.which(getattr(args, name))
|
|
if not value:
|
|
cli.error(f'missing {name}: {getattr(args, name)}')
|
|
setattr(args, name, value)
|
|
root = Path(tempfile.mkdtemp(prefix='hydra-smoke-'))
|
|
print(f'Evidence: {root}', flush=True)
|
|
xdg = root / 'xdg'
|
|
xdg.mkdir(mode=0o700)
|
|
config = root / 'labwc'
|
|
config.mkdir()
|
|
(config / 'rc.xml').write_text(build_labwc_config(['probe'], ['/bin/true']))
|
|
(config / 'autostart').write_text('#!/bin/sh\n')
|
|
(config / 'environment').touch()
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(xdg), WLR_BACKENDS='headless',
|
|
WLR_RENDERER='pixman', WLR_XWAYLAND='/bin/false', XDG_DATA_DIRS='/nonexistent',
|
|
GSETTINGS_BACKEND='memory')
|
|
for key in ('DISPLAY', 'WAYLAND_DISPLAY'):
|
|
env.pop(key, None)
|
|
manager = Runtime(root / 'config', root / 'state', root / 'run')
|
|
manager.add(Profile('probe', memory=128, cpus=1, disk_size='8M', graphics='software'))
|
|
processes = []
|
|
try:
|
|
with (root / 'labwc.log').open('w') as cl, (root / 'viewer.log').open('w') as vl:
|
|
compositor = subprocess.Popen([args.labwc, '-d', '-C', str(config)], env=env, stdout=cl, stderr=cl)
|
|
processes.append(compositor)
|
|
wait_for(lambda: list(xdg.glob('wayland-*')), 'private display did not start')
|
|
env['WAYLAND_DISPLAY'] = next(p.name for p in xdg.glob('wayland-*') if not p.name.endswith('.lock'))
|
|
keyboard = subprocess.Popen([args.keyboard, '-k', 'Escape', '-s', '30000'], env=env)
|
|
processes.append(keyboard)
|
|
time.sleep(.2)
|
|
manager.start('probe')
|
|
previous = dict(os.environ)
|
|
try:
|
|
os.environ.update(env)
|
|
viewer_env = _viewer_environment(root / 'viewer-state')
|
|
finally:
|
|
os.environ.clear()
|
|
os.environ.update(previous)
|
|
viewer_env['WAYLAND_DEBUG'] = '1'
|
|
viewer = subprocess.Popen(build_viewer_argv(manager.paths('probe'), args.viewer),
|
|
env=viewer_env, stdout=vl, stderr=vl)
|
|
processes.append(viewer)
|
|
log = root / 'viewer.log'
|
|
wait_for(lambda: 'set_title("HydraOS: probe")' in log.read_text(), 'viewer did not connect')
|
|
time.sleep(.5)
|
|
title = 'set_title("Select USB devices for redirection")'
|
|
assert title not in log.read_text(), 'USB selector opened automatically'
|
|
subprocess.run([args.keyboard, '-M', 'ctrl', '-M', 'alt', '-M', 'shift', '-k', 'u',
|
|
'-m', 'shift', '-m', 'alt', '-m', 'ctrl'], env=env, check=True)
|
|
wait_for(lambda: title in log.read_text(), 'manual USB selector did not open', timeout=3)
|
|
viewer.terminate()
|
|
viewer.wait(timeout=5)
|
|
assert manager.status('probe')['status'] == 'running'
|
|
compositor.terminate()
|
|
compositor.wait(timeout=5)
|
|
assert manager.status('probe')['status'] == 'running'
|
|
result = {'manual_usb_dialog': True, 'automatic_usb_dialog': False,
|
|
'qemu_survives_viewer_and_compositor': True,
|
|
'viewer_version': subprocess.check_output([args.viewer, '--version'], text=True).strip()}
|
|
(root / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
|
|
print(json.dumps(result, indent=2))
|
|
finally:
|
|
for proc in reversed(processes):
|
|
if proc.poll() is None:
|
|
proc.terminate()
|
|
proc.wait(timeout=5)
|
|
if manager.status('probe')['status'] != 'stopped':
|
|
manager.stop('probe', force=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|