fix: keep manual USB selection available during keyboard grabs

2026-10-13 04:29:37 +03:00
parent 48ee598e9f
commit 38914cf67f
2 changed files with 145 additions and 4 deletions
@@ -0,0 +1,106 @@
+#!/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()