#!/usr/bin/env python3 """Exercise production Qt input controls through IBus/Mozc on private X11 or Wayland.""" import argparse import hashlib import json import os from pathlib import Path import platform import signal import subprocess import sys import tempfile import time import uuid ROOT = Path(__file__).resolve().parents[2] def sha(path): with Path(path).open('rb') as f: return hashlib.file_digest(f, 'sha256').hexdigest() def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--binary', type=Path, required=True) p.add_argument('--canonical-build', type=Path, required=True) p.add_argument('--output', type=Path, required=True) p.add_argument('--platform', choices=['xcb', 'wayland'], default='xcb') p.add_argument('--private-session', action='store_true', help=argparse.SUPPRESS) a = p.parse_args() if a.output.exists(): p.error('Choose a new output directory') if os.environ.get('QTWEBENGINE_DISABLE_SANDBOX', '') not in ('', '0') or 'sandbox' in os.environ.get('QTWEBENGINE_CHROMIUM_FLAGS', ''): p.error('Sandbox overrides are not permitted') if not a.private_session: with tempfile.TemporaryDirectory(prefix='docview-ime-') as private: env = os.environ.copy() for key in ['DISPLAY', 'WAYLAND_DISPLAY', 'WAYLAND_SOCKET', 'SWAYSOCK', 'I3SOCK', 'DBUS_SESSION_BUS_ADDRESS', 'IBUS_ADDRESS', 'QT_SCREEN_SCALE_FACTORS', 'QT_AUTO_SCREEN_SCALE_FACTOR', 'QT_DEVICE_PIXEL_RATIO', 'QT_QPA_PLATFORMTHEME']: env.pop(key, None) env.update(XDG_RUNTIME_DIR=private, XDG_CONFIG_HOME=private + '/config', XDG_STATE_HOME=private + '/state', XDG_CACHE_HOME=private + '/cache', XDG_DATA_HOME=private + '/data', QT_QPA_PLATFORM=a.platform, DOCVIEW_IME_PLATFORM=a.platform, QT_QUICK_BACKEND='software', QT_SCALE_FACTOR='1', QT_FONT_DPI='96', QT_IM_MODULE='ibus', QTWEBENGINE_CHROMIUM_FLAGS='--disable-gpu', DOCVIEW_PRIVATE_IME='1', DOCVIEW_IME_RUN_ID=str(uuid.uuid4())) bus = Path(private) / 'bus.conf' bus.write_text('sessionEXTERNAL' f'unix:tmpdir={private}' '') display = ['xvfb-run', '-a', '-s', '-screen 0 1100x760x24 -dpi 96 -nolisten tcp'] if a.platform == 'xcb' else [] return subprocess.run(['dbus-run-session', '--config-file', str(bus), '--', *display, sys.executable, str(Path(__file__).resolve()), *sys.argv[1:], '--private-session'], env=env).returncode out = a.output.resolve(); out.mkdir(parents=True, exist_ok=False) binary, build = a.binary.resolve(), a.canonical_build.resolve() paths = [binary] + [build / n for n in ['docview', 'docview-pdf-worker', 'docview-archive-worker', 'libdocview_core.a', 'libdocview_common.a', 'libdocview_font_broker.a']] paths += [ROOT / 'tests/fixtures/pdf/navigation.pdf', ROOT / 'resources.qrc'] paths += [p for folder in ['src', 'qml', 'resources', 'tests/ime'] for p in (ROOT / folder).rglob('*') if p.is_file() and '__pycache__' not in p.parts] before = {os.path.relpath(path, ROOT): sha(path) for path in sorted(set(paths))} report = {'success': False, 'os': platform.freedesktop_os_release(), 'kernel': platform.release(), 'inputs': before, 'requestedPlatform': a.platform, 'scope': ('Actual IBus/Mozc through virtual XTest keys and US/JP XKB on private X11' if a.platform == 'xcb' else 'Actual Qt Wayland and IBus/Mozc over D-Bus; private Sway and wtype virtual-keyboard keysyms, not physical US/JP mappings or compositor text-input protocol') + '; no physical keyboard, candidate-panel appearance or Windows acceptance'} address = 'unix:path=' + os.environ['XDG_RUNTIME_DIR'] + '/ibus.sock' env = os.environ.copy(); env.update(DOCVIEW_IME_CHILD_ID=str(uuid.uuid4()), IBUS_ADDRESS=address, DOCVIEW_IME_OUTPUT=str(out), DOCVIEW_IME_PDF=str(ROOT / 'tests/fixtures/pdf/navigation.pdf'), DOCVIEW_IME_RAW_KEY=str(ROOT / 'tests/ime/raw_key.py')) daemon = compositor = keyboard_keeper = None tag = ('DOCVIEW_IME_CHILD_ID=' + env['DOCVIEW_IME_CHILD_ID']).encode() def owned(): found = [] for directory in Path('/proc').glob('[0-9]*'): try: pid = int(directory.name) if pid != os.getpid() and tag in (directory / 'environ').read_bytes().split(b'\0'): found.append(pid) except (OSError, ValueError): pass return found try: for name in ['docview-pdf-worker', 'docview-archive-worker']: assert sha(binary.parent / name) == sha(build / name) report['packages'] = subprocess.check_output(['dpkg-query', '-W', 'ibus', 'ibus-mozc', 'mozc-server', 'xdotool', 'libxtst6', 'xkb-data', 'sway', 'wtype'], text=True) if a.platform == 'wayland': runtime = Path(env['XDG_RUNTIME_DIR']) config = runtime / 'sway.conf' config.write_text('xwayland disable\nswaybg_command -\nswaynag_command -\n' 'output HEADLESS-1 mode 1920x1080\nfor_window [app_id=".*"] floating enable\n' 'default_border none\ndefault_floating_border none\nfocus_follows_mouse no\nmouse_warping none\n') (out / 'sway.conf').write_bytes(config.read_bytes()) env.update(WLR_BACKENDS='headless', WLR_RENDERER='pixman', WLR_HEADLESS_OUTPUTS='1', XDG_CURRENT_DESKTOP='sway', XDG_SESSION_TYPE='wayland', QT_USE_PORTAL='0') with (out / 'compositor.log').open('w') as log: compositor = subprocess.Popen(['sway', '-c', str(config)], env=env, stdout=log, stderr=subprocess.STDOUT) for _ in range(100): sockets = [p for p in runtime.glob('wayland-*') if p.is_socket()] ipc = list(runtime.glob('sway-ipc.*.sock')) if sockets and ipc or compositor.poll() is not None: break time.sleep(.1) if not sockets or not ipc or compositor.poll() is not None: raise RuntimeError('Private Sway did not become ready') env.update(WAYLAND_DISPLAY=sockets[0].name, SWAYSOCK=str(ipc[0])) report['waylandOutputs'] = json.loads(subprocess.check_output(['swaymsg', '-s', env['SWAYSOCK'], '-t', 'get_outputs'], env=env)) # Keep a virtual keyboard attached throughout the test. Otherwise # each short wtype process disconnects the seat's only keyboard, # clears Qt's active focus and cancels an in-progress composition. # The deferred key is never reached: test timeout is 180 seconds, # and cleanup terminates this uniquely tagged child first. with (out / 'keyboard-keeper.log').open('w') as log: keyboard_keeper = subprocess.Popen(['wtype', '-s', '300000', '-k', 'Shift_L'], env=env, stdout=log, stderr=subprocess.STDOUT) for _ in range(100): inputs = json.loads(subprocess.check_output(['swaymsg', '-s', env['SWAYSOCK'], '-t', 'get_inputs'], env=env)) if any(row.get('type') == 'keyboard' for row in inputs) or keyboard_keeper.poll() is not None: break time.sleep(.05) if not any(row.get('type') == 'keyboard' for row in inputs) or keyboard_keeper.poll() is not None: raise RuntimeError('Persistent virtual keyboard did not attach') report['waylandInputsBeforeApplication'] = inputs with (out / 'ibus.log').open('w') as log: daemon = subprocess.Popen(['ibus-daemon', '--address', address, '--panel', 'disable', '--config', 'disable', '--emoji-extension', 'disable', '--cache', 'none'], env=env, stdout=log, stderr=subprocess.STDOUT) ready = False for _ in range(50): query = subprocess.run(['ibus', 'list-engine'], env=env, capture_output=True, text=True, timeout=5) if query.returncode == 0 and 'mozc-jp' in query.stdout: (out / 'engines.txt').write_text(query.stdout); ready = True; break if daemon.poll() is not None: break time.sleep(.1) if not ready: raise RuntimeError('IBus/Mozc did not become ready') # Launch the installed engine explicitly: this session deliberately # has no ambient D-Bus service activation. with (out / 'mozc.log').open('w') as mozc_log: mozc = subprocess.Popen(['/usr/lib/ibus-mozc/ibus-engine-mozc', '--ibus'], env=env, stdout=mozc_log, stderr=subprocess.STDOUT) time.sleep(.5) if mozc.poll() is not None: raise RuntimeError(f'Mozc engine exited: {mozc.returncode}') probe = subprocess.run(['gdbus', 'call', '--address', address, '--dest', 'org.freedesktop.IBus', '--object-path', '/org/freedesktop/IBus', '--method', 'org.freedesktop.IBus.SetGlobalEngine', 'mozc-jp'], env=env, capture_output=True, text=True, timeout=20) (out / 'engine-probe.txt').write_text(probe.stdout + probe.stderr) with (out / 'test.log').open('w') as testlog: test = subprocess.run([str(binary)], env=env, stdout=testlog, stderr=subprocess.STDOUT, timeout=180) report['testExitCode'] = test.returncode layouts = ['us', 'jp'] if a.platform == 'xcb' else ['virtual-keysyms'] report['cases'] = {layout: json.loads((out / f'{layout}.json').read_text()) for layout in layouts} report['success'] = test.returncode == 0 and all(case['success'] and case['qtPlatform'] == a.platform for case in report['cases'].values()) except Exception as error: report['error'] = str(error) finally: if daemon: try: subprocess.run(['ibus', 'exit'], env=env, capture_output=True, timeout=5) except subprocess.TimeoutExpired: pass for sig in [signal.SIGTERM, signal.SIGKILL]: for pid in owned(): try: os.kill(pid, sig) except ProcessLookupError: pass time.sleep(.2) if daemon: daemon.wait(timeout=5) if compositor: compositor.wait(timeout=5) if keyboard_keeper: keyboard_keeper.wait(timeout=5) report['remainingOwnedProcesses'] = owned() report['inputsUnchanged'] = all(sha(ROOT / name) == value for name, value in before.items()) report['success'] &= report['inputsUnchanged'] and not report['remainingOwnedProcesses'] (out / 'report.json').write_text(json.dumps(report, indent=2) + '\n') print(json.dumps({k: report.get(k) for k in ['success', 'testExitCode', 'error', 'inputsUnchanged', 'remainingOwnedProcesses']})) return 0 if report['success'] else 1 if __name__ == '__main__': raise SystemExit(main())