From 8b3281950a5cb3d61a0f56963a632c87446c0809 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 04:18:34 +0300 Subject: [PATCH] feat: add fullscreen HydraOS session --- dev/tools.nix | 23 +++ hydra/cli.py | 20 +- hydra/session.py | 263 ++++++++++++++++++++++++ patches/virt-viewer-hydra-session.patch | 112 ++++++++++ session/hydra-session | 4 + tests/test_session.py | 197 ++++++++++++++++++ 6 files changed, 617 insertions(+), 2 deletions(-) create mode 100644 dev/tools.nix create mode 100644 hydra/session.py create mode 100644 patches/virt-viewer-hydra-session.patch create mode 100755 session/hydra-session create mode 100644 tests/test_session.py diff --git a/dev/tools.nix b/dev/tools.nix new file mode 100644 index 0000000..aae3944 --- /dev/null +++ b/dev/tools.nix @@ -0,0 +1,23 @@ +# Temporary tools only: nix-shell dev/tools.nix. No host/profile installation. +let + pkgs = import (builtins.fetchTarball { + url = "https://github.com/NixOS/nixpkgs/archive/7c8764b7c7b09b34f632464276218ef9090eaa11.tar.gz"; + sha256 = "sha256-MWhktfFtFE8Jl7bf9Pkp7TR5MSdYCjCswSo0Gkb4ywc="; + }) {}; + graphics = import (builtins.fetchTarball { + url = "https://github.com/NixOS/nixpkgs/archive/7fd4004d0511991fff3a0524856a9be09e28c5f0.tar.gz"; + sha256 = "sha256-i5qk1kLiit98rRbh1AXmWBNT9QXC/y4l5Wk/u5EdpDQ="; + }) {}; + viewer = pkgs.virt-viewer.overrideAttrs (old: { + src = pkgs.fetchurl { + url = "https://releases.pagure.org/virt-viewer/virt-viewer-11.0.tar.xz"; + hash = "sha256-pD+iMlxMHHelyMmAZaww7wURohrJjlkPIjQIabrZq9A="; + }; + patches = (old.patches or []) ++ [ ../patches/virt-viewer-hydra-session.patch ]; + doCheck = true; + }); +in pkgs.mkShell { + packages = [ pkgs.python3 pkgs.qemu viewer graphics.labwc pkgs.foot + pkgs.brightnessctl pkgs.wayland-utils pkgs.wlr-randr ]; + passthru = { inherit viewer; labwc = graphics.labwc; }; +} diff --git a/hydra/cli.py b/hydra/cli.py index f998065..3b60a5c 100644 --- a/hydra/cli.py +++ b/hydra/cli.py @@ -3,11 +3,13 @@ import argparse import json import os +import subprocess import sys from pathlib import Path from .profiles import Profile from .runtime import Runtime, RuntimeError +from .session import launch_session, session_helper, view def parser() -> argparse.ArgumentParser: @@ -33,6 +35,14 @@ def parser() -> argparse.ArgumentParser: cmd.add_argument('name') if command == 'stop': cmd.add_argument('--force', action='store_true', required=True) + viewer = commands.add_parser('view', help='Connect a viewer without changing VM power') + viewer.add_argument('name') + viewer.add_argument('--windowed', action='store_true') + session = commands.add_parser('session', help='Start the minimal HydraOS labwc session') + session.add_argument('--nested', action='store_true') + helper = commands.add_parser('_session-helper', help=argparse.SUPPRESS) + helper.add_argument('operation') + helper.add_argument('manifest', type=Path) return cli @@ -40,7 +50,13 @@ def main(argv: list[str] | None = None) -> int: args = parser().parse_args(argv) runtime = Runtime(args.config, args.state, args.runtime) try: - if args.command == 'add': + if args.command == 'view': + return view(runtime, args.name, windowed=args.windowed) + elif args.command == 'session': + return launch_session(runtime, nested=args.nested) + elif args.command == '_session-helper': + return session_helper(args.operation, args.manifest) + elif args.command == 'add': profile = Profile(args.name, args.iso, args.memory, args.cpus, args.disk_size, args.graphics, args.network, args.bridge) runtime.add(profile) @@ -55,6 +71,6 @@ def main(argv: list[str] | None = None) -> int: result = getattr(runtime, args.command)(args.name) print(json.dumps(result, indent=2)) return 0 - except (RuntimeError, OSError, ValueError) as exc: + except (RuntimeError, OSError, ValueError, subprocess.TimeoutExpired) as exc: print(f'hydra: {exc}', file=sys.stderr) return 1 diff --git a/hydra/session.py b/hydra/session.py new file mode 100644 index 0000000..a6cf1f6 --- /dev/null +++ b/hydra/session.py @@ -0,0 +1,263 @@ +"""Minimal labwc session; viewer and compositor lifetimes never control QEMU.""" + +import fcntl +import json +import os +import re +import shlex +import shutil +import stat +import subprocess +import sys +import tempfile +import time +from pathlib import Path +from urllib.parse import quote +from xml.etree import ElementTree as ET + +from .profiles import validate_name +from .runtime import Runtime, VMPaths, _private_dir + + +class SessionError(ValueError): + """An actionable viewer/session error safe to present at the CLI.""" + + +def viewer_title(name: str) -> str: + return 'HydraOS: ' + validate_name(name) + + +def build_viewer_argv(paths: VMPaths, viewer: str, *, windowed=False) -> list[str]: + argv = [viewer, '--title', viewer_title(paths.name), '--hotkeys', + 'release-cursor=ctrl+alt+shift+r,usb-device-select=ctrl+alt+shift+u', + '--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0', + '--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0'] + if not windowed: + argv += ['--full-screen', '--hide-toolbar'] + # QEMU denies clipboard/file transfer; no shared-dir/CD flag is added here. + return argv + ['spice+unix://' + quote(str(paths.spice), safe='/')] + + +def build_labwc_config(names: list[str], recovery_argv: list[str], + brightness='brightnessctl', *, initial: str | None = None) -> str: + names = [validate_name(name) for name in names] + if len(set(names)) != len(names): + raise SessionError('Duplicate VM workspace name') + if initial is not None and initial not in names: + raise SessionError('Initial VM workspace must be configured') + root = ET.Element('labwc_config') + core = ET.SubElement(root, 'core') + ET.SubElement(core, 'primarySelection').text = 'no' + desktops = ET.SubElement(root, 'desktops', number=str(max(1, len(names)))) + labels = ET.SubElement(desktops, 'names') + for name in names: + ET.SubElement(labels, 'name').text = name + if initial is not None: + ET.SubElement(desktops, 'initial').text = str(names.index(initial) + 1) + ET.SubElement(desktops, 'popupTime').text = '0' + theme = ET.SubElement(root, 'theme') + ET.SubElement(theme, 'keepBorder').text = 'no' + keyboard = ET.SubElement(root, 'keyboard') + for key, direction in (('C-A-S-Left', 'left'), ('C-A-S-Right', 'right')): + binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes') + ET.SubElement(binding, 'action', name='GoToDesktop', to=direction, wrap='yes') + commands = [('C-A-S-Escape', recovery_argv), + ('XF86MonBrightnessUp', [brightness, '-c', 'backlight', 'set', '+5%']), + ('XF86MonBrightnessDown', [brightness, '-c', 'backlight', 'set', '5%-'])] + for key, argv in commands: + binding = ET.SubElement(keyboard, 'keybind', key=key, overrideInhibition='yes') + ET.SubElement(binding, 'action', name='Execute', command=shlex.join(argv)) + # An empty mouse section restores labwc defaults. A binding with no action + # suppresses that fallback without consuming guest window pointer events. + context = ET.SubElement(ET.SubElement(root, 'mouse'), 'context', name='Root') + ET.SubElement(context, 'mousebind', button='Left', action='Press') + device = ET.SubElement(ET.SubElement(root, 'libinput'), 'device', category='touchpad') + ET.SubElement(device, 'tap').text = 'yes' + rules = ET.SubElement(root, 'windowRules') + for index, name in enumerate(names, 1): + rule = ET.SubElement(rules, 'windowRule', title=viewer_title(name), + serverDecoration='no') + ET.SubElement(rule, 'action', name='SendToDesktop', to=str(index), follow='no') + ET.SubElement(rule, 'action', name='SetDecorations', decorations='none') + ET.indent(root) + return ET.tostring(root, encoding='unicode', xml_declaration=True) + '\n' + + +def _tool(variable: str, default: str) -> str: + value = os.environ.get(variable, default) + executable = shutil.which(value) + if not executable: + raise SessionError(f'Missing {default} tool ({variable}); enter the temporary dev/tools.nix environment') + return executable + + +def _display() -> None: + if not os.environ.get('WAYLAND_DISPLAY') and not os.environ.get('DISPLAY'): + raise SessionError('No graphical display; start a session or use a graphical host terminal') + + +def _viewer() -> str: + viewer = _tool('HYDRA_VIEWER', 'remote-viewer') + result = subprocess.run([viewer, '--help'], capture_output=True, text=True, timeout=10) + if result.returncode or '--hide-toolbar' not in result.stdout: + raise SessionError('Viewer lacks HydraOS toolbar/USB adaptation; use dev/tools.nix') + return viewer + + +def _socket(paths: VMPaths) -> None: + try: + mode = paths.spice.lstat().st_mode + except FileNotFoundError as exc: + raise SessionError(f'{paths.name}: no SPICE socket; run hydra start {paths.name}, then retry') from exc + if not stat.S_ISSOCK(mode): + raise SessionError(f'{paths.name}: SPICE path is not a Unix socket') + + +def _viewer_environment(directory: Path) -> dict[str, str]: + config = directory / 'config' + settings = config / 'virt-viewer' + settings.mkdir(parents=True, mode=0o700) + (settings / 'settings').write_text('[virt-viewer]\nshare-clipboard=false\n', encoding='utf-8') + env = dict(os.environ, XDG_CONFIG_HOME=str(config), GSETTINGS_BACKEND='memory') + # A dedicated backend prevents accidental XWayland input behavior in HydraOS. + if env.get('WAYLAND_DISPLAY'): + env['GDK_BACKEND'] = 'wayland' + return env + + +def view(runtime: Runtime, name: str, *, windowed=False) -> int: + _display() + viewer = _viewer() + runtime.profile(name) + paths = runtime.paths(name) + _socket(paths) + _private_dir(runtime.runtime) + with tempfile.TemporaryDirectory(prefix='viewer-', dir=runtime.runtime) as tmp: + return subprocess.run(build_viewer_argv(paths, viewer, windowed=windowed), + env=_viewer_environment(Path(tmp))).returncode + + +def compositor_environment(environ: dict[str, str], *, nested=False) -> dict[str, str]: + env = dict(environ) + if nested: + if env.get('WAYLAND_DISPLAY'): + env['WLR_BACKENDS'] = 'wayland' + elif env.get('DISPLAY'): + env['WLR_BACKENDS'] = 'x11' + else: + raise SessionError('Nested session requires an existing graphical display') + else: + env.pop('WAYLAND_DISPLAY', None) + env.pop('DISPLAY', None) + env['WLR_BACKENDS'] = 'drm,libinput' + env['XDG_CURRENT_DESKTOP'] = 'HydraOS' + return env + + +def _module_argv(command: str, manifest: Path) -> list[str]: + # Absolute source launcher also works when labwc changes the working directory. + return [sys.executable, str(Path(__file__).resolve().parents[1] / 'bin/hydra'), + '_session-helper', command, str(manifest)] + + +def _recovery_argv(manifest: Path, terminal: str) -> list[str]: + return [terminal, '--title=HydraOS-recovery'] + _module_argv('recover', manifest) + + +def launch_session(runtime: Runtime, *, nested=False) -> int: + env = compositor_environment(dict(os.environ), nested=nested) + labwc = _tool('HYDRA_LABWC', 'labwc') + terminal = _tool('HYDRA_TERMINAL', 'foot') + brightness = _tool('HYDRA_BRIGHTNESS', 'brightnessctl') + version = subprocess.run([labwc, '--version'], capture_output=True, text=True, timeout=10) + match = re.search(r'(\d+)\.(\d+)\.(\d+)', version.stdout) + if version.returncode or not match or tuple(map(int, match.groups())) < (0, 20, 0): + raise SessionError('HydraOS requires labwc >= 0.20.0 for reserved host input bindings') + profiles = runtime.profiles() + initial = next((p.name for p in profiles if runtime.status(p.name)['status'] == 'running'), None) + _private_dir(runtime.runtime) + with tempfile.TemporaryDirectory(prefix='session-', dir=runtime.runtime) as tmp: + directory = Path(tmp) + config = directory / 'labwc' + config.mkdir(mode=0o700) + manifest = directory / 'session.json' + manifest.write_text(json.dumps({'config': str(runtime.config), 'state': str(runtime.state), + 'runtime': str(runtime.runtime), 'names': [p.name for p in profiles], + 'terminal': terminal, 'directory': str(directory)}), encoding='utf-8') + (config / 'rc.xml').write_text(build_labwc_config([p.name for p in profiles], + _recovery_argv(manifest, terminal), brightness, + initial=initial), encoding='utf-8') + (config / 'autostart').write_text('#!/bin/sh\n' + shlex.join(_module_argv('clients', manifest)) + ' &\n', encoding='utf-8') + # Empty environment files prevent loading unrelated /etc/xdg defaults. + (config / 'environment').touch() + env['PYTHONUNBUFFERED'] = '1' + return subprocess.run([labwc, '-C', str(config)], env=env).returncode + + +def session_helper(command: str, manifest: Path) -> int: + data = json.loads(manifest.read_text(encoding='utf-8')) + if (not isinstance(data, dict) or + any(not isinstance(data.get(key), str) or not Path(data[key]).is_absolute() + for key in ('config', 'state', 'runtime', 'terminal')) or + not isinstance(data.get('names'), list)): + raise SessionError('Invalid session manifest') + for name in data['names']: + validate_name(name) + if command not in ('clients', 'recover') and ( + not command.startswith('reconnect:') or command.split(':', 1)[1] not in data['names']): + raise SessionError('Invalid session helper operation') + runtime = Runtime(Path(data['config']), Path(data['state']), Path(data['runtime'])) + if command == 'recover': + print('HydraOS local recovery. VMs keep running independently.\n' + 'Ctrl+Alt+Shift+Left/Right switches workspaces; Ctrl+Alt+Shift+R releases input; Ctrl+Alt+Shift+U opens USB.\n' + 'Enter a VM name to reconnect, s for a host shell, q to close this terminal.') + while True: + try: + print(json.dumps([runtime.status(name) for name in data['names']], indent=2)) + choice = input('hydra recovery> ').strip() + except (EOFError, KeyboardInterrupt): + return 0 + if choice == 'q': + return 0 + if choice == 's': + subprocess.run([os.environ.get('SHELL', '/bin/sh')]) + elif choice in data['names']: + subprocess.Popen(_module_argv('reconnect:' + choice, manifest), start_new_session=True) + else: + print('Enter a configured VM name, s, or q.') + if command.startswith('reconnect:'): + name = command.split(':', 1)[1] + # One viewer per VM inside this session; independent host view commands + # retain their separate lifecycle and never share VM manager locks. + with (manifest.parent / (name + '.viewer.lock')).open('a') as lock: + try: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + print(f'{name}: viewer is already connected; close recovery or switch workspace') + return 0 + return view(runtime, name) + processes = [] + recovery = None + def open_recovery(): + nonlocal recovery + if recovery is None or recovery.poll() is not None: + recovery = subprocess.Popen(_recovery_argv(manifest, data['terminal'])) + for name in data['names']: + try: + if runtime.status(name)['status'] == 'running': + processes.append(subprocess.Popen(_module_argv('reconnect:' + name, manifest))) + except (OSError, ValueError) as exc: + print(f'hydra: {name}: {exc}', file=sys.stderr) + if not processes: + open_recovery() + # No signals to QEMU or viewers: compositor disconnect tears down clients, + # while the VM manager's detached QEMU processes retain their lifetime. + while processes or (recovery is not None and recovery.poll() is None): + for process in processes[:]: + code = process.poll() + if code is not None: + processes.remove(process) + print(f'hydra: viewer exited ({code}); reconnect from local recovery', file=sys.stderr) + open_recovery() + time.sleep(0.2) + return 0 diff --git a/patches/virt-viewer-hydra-session.patch b/patches/virt-viewer-hydra-session.patch new file mode 100644 index 0000000..f32225e --- /dev/null +++ b/patches/virt-viewer-hydra-session.patch @@ -0,0 +1,112 @@ +--- a/src/virt-viewer-app.c ++++ b/src/virt-viewer-app.c +@@ -134,6 +134,7 @@ + gboolean authretry; + gboolean started; + gboolean fullscreen; ++ gboolean hide_toolbar; + gboolean attach; + gboolean shared; + gboolean quitting; +@@ -951,6 +952,13 @@ + { + gchar *subtitle = NULL; + const gchar *title = virt_viewer_app_get_title(app); ++ ++ /* Hidden fullscreen windows use the requested title as stable identity. */ ++ if (title != NULL && virt_viewer_app_get_hide_toolbar(app) && ++ virt_viewer_app_get_fullscreen(app)) { ++ g_object_set(window, "subtitle", title, NULL); ++ return; ++ } + + if (title != NULL) { + VirtViewerDisplay *display = virt_viewer_window_get_display(window); +@@ -2134,6 +2142,7 @@ + static gboolean opt_verbose = FALSE; + static gboolean opt_debug = FALSE; + static gboolean opt_fullscreen = FALSE; ++static gboolean opt_hide_toolbar = FALSE; + static gboolean opt_kiosk = FALSE; + static gboolean opt_kiosk_quit = FALSE; + static gchar *opt_cursor = NULL; +@@ -2437,6 +2446,7 @@ + { "smartcard-remove", "app.smartcard-remove", {"F9", NULL, NULL} }, + { "secure-attention", "win.secure-attention", {"End", NULL, NULL} }, + { "usb-device-reset", "win.usb-device-reset", {"r", NULL, NULL} }, ++ { "usb-device-select", "win.usb-device-select", {NULL, NULL, NULL} }, + }; + + static gchar **hotkey_names; +@@ -2484,6 +2494,7 @@ + priv->resource = virt_viewer_get_resource(); + + virt_viewer_app_set_debug(opt_debug); ++ priv->hide_toolbar = opt_hide_toolbar; + virt_viewer_app_set_fullscreen(self, opt_fullscreen); + + virt_viewer_app_set_keymap(self, opt_keymap); +@@ -3098,6 +3109,14 @@ + } + + gboolean ++virt_viewer_app_get_hide_toolbar(VirtViewerApp *self) ++{ ++ VirtViewerAppPrivate *priv = virt_viewer_app_get_instance_private(self); ++ ++ return priv->hide_toolbar; ++} ++ ++gboolean + virt_viewer_app_get_fullscreen(VirtViewerApp *self) + { + g_return_val_if_fail(VIRT_VIEWER_IS_APP(self), FALSE); +@@ -3557,6 +3576,8 @@ + N_("Zoom level of window, in percentage"), "ZOOM" }, + { "full-screen", 'f', 0, G_OPTION_ARG_NONE, &opt_fullscreen, + N_("Open in full screen mode (adjusts guest resolution to fit the client)"), NULL }, ++ { "hide-toolbar", '\0', 0, G_OPTION_ARG_NONE, &opt_hide_toolbar, ++ N_("Hide the toolbar in full screen mode"), NULL }, + { "hotkeys", 'H', 0, G_OPTION_ARG_STRING, &opt_hotkeys, + N_("Customise hotkeys"), NULL }, + { "auto-resize", '\0', 0, G_OPTION_ARG_STRING, &opt_resize, +--- a/src/virt-viewer-app.h ++++ b/src/virt-viewer-app.h +@@ -100,6 +100,7 @@ + GList* virt_viewer_app_get_windows(VirtViewerApp *self); + VirtViewerSession* virt_viewer_app_get_session(VirtViewerApp *self); + gboolean virt_viewer_app_get_fullscreen(VirtViewerApp *app); ++gboolean virt_viewer_app_get_hide_toolbar(VirtViewerApp *app); + void virt_viewer_app_clear_hotkeys(VirtViewerApp *app); + GList* virt_viewer_app_get_initial_displays(VirtViewerApp* self); + gint virt_viewer_app_get_initial_monitor_for_display(VirtViewerApp* self, gint display); +--- a/src/virt-viewer-window.c ++++ b/src/virt-viewer-window.c +@@ -740,9 +740,10 @@ + return; + } + +- if (!self->kiosk) { +- virt_viewer_timed_revealer_force_reveal(self->revealer, TRUE); +- } ++ gboolean hide_toolbar = virt_viewer_app_get_hide_toolbar(self->app); ++ gtk_widget_set_visible(GTK_WIDGET(self->revealer), !hide_toolbar); ++ virt_viewer_timed_revealer_force_reveal(self->revealer, ++ !self->kiosk && !hide_toolbar); + + if (self->display) { + virt_viewer_display_set_monitor(self->display, monitor); +@@ -1354,6 +1355,13 @@ + title = g_strdup(g_get_application_name()); + } + ++ /* Keep exact workspace identity across pointer/keyboard grab changes. */ ++ if (self->app && virt_viewer_app_get_hide_toolbar(self->app) && ++ virt_viewer_app_get_fullscreen(self->app) && self->subtitle) { ++ g_free(title); ++ title = g_strdup(self->subtitle); ++ } ++ + gtk_window_set_title(GTK_WINDOW(self->window), title); + if (self->subtitle) { + gtk_header_bar_set_title(GTK_HEADER_BAR(header), self->subtitle); diff --git a/session/hydra-session b/session/hydra-session new file mode 100755 index 0000000..f07e452 --- /dev/null +++ b/session/hydra-session @@ -0,0 +1,4 @@ +#!/bin/sh +# Portable source launcher. Desktop/session packaging may call this directly. +HYDRA_CHECKOUT=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd) || exit 1 +exec "$HYDRA_CHECKOUT/bin/hydra" session "$@" diff --git a/tests/test_session.py b/tests/test_session.py new file mode 100644 index 0000000..d68abe2 --- /dev/null +++ b/tests/test_session.py @@ -0,0 +1,197 @@ +import contextlib +import fcntl +import io +import json +import os +import shlex +import socket +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch +from xml.etree import ElementTree as ET + +from hydra.profiles import Profile +from hydra.runtime import Runtime +from hydra import session +from hydra.cli import main, parser + + +class SessionTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix='hs-') + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.iso = self.root / 'scratch.iso' + self.iso.touch() + self.manager = Runtime(self.root / 'config', self.root / 'state', self.root / 'run') + self.manager.add(Profile('alpha', self.iso)) + self.manager.add(Profile('beta', self.iso)) + + def test_config_has_exact_titles_workspaces_and_explicit_fullscreen(self): + tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot', 'a & b', "x'y"])) + self.assertEqual([n.text for n in tree.findall('./desktops/names/name')], ['alpha', 'beta']) + self.assertEqual(tree.findtext('./desktops/popupTime'), '0') + rules = tree.findall('./windowRules/windowRule') + self.assertEqual([r.get('title') for r in rules], ['HydraOS: alpha', 'HydraOS: beta']) + for index, rule in enumerate(rules, 1): + self.assertEqual(rule.get('serverDecoration'), 'no') + self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('to'), str(index)) + self.assertEqual(rule.find("./action[@name='SendToDesktop']").get('follow'), 'no') + # Fullscreen is requested by the viewer, never toggled by a rule. + self.assertIsNone(rule.find("./action[@name='ToggleFullscreen']")) + recovery = tree.find("./keyboard/keybind[@key='C-A-S-Escape']/action") + self.assertEqual(shlex.split(recovery.get('command')), ['foot', 'a & b', "x'y"]) + + def test_only_reserved_host_keys_and_inert_mouse_binding(self): + tree = ET.fromstring(session.build_labwc_config(['alpha'], ['foot'])) + self.assertIsNone(tree.find('./keyboard/default')) + self.assertIsNone(tree.find('./mouse/default')) + bindings = tree.findall('./keyboard/keybind') + self.assertEqual({b.get('key') for b in bindings}, {'C-A-S-Left', 'C-A-S-Right', 'C-A-S-Escape', 'XF86MonBrightnessUp', 'XF86MonBrightnessDown'}) + self.assertTrue(all(b.get('overrideInhibition') == 'yes' for b in bindings)) + self.assertEqual(len(tree.findall('./mouse/context/mousebind')), 1) + self.assertEqual(tree.findall('./mouse/context/mousebind/action'), []) + self.assertEqual(tree.findtext('./libinput/device/tap'), 'yes') + + def test_viewer_argv_is_private_safe_and_manual_usb_only(self): + paths = self.manager.paths('alpha') + argv = session.build_viewer_argv(paths, '/tool path/viewer') + self.assertEqual(argv[0], '/tool path/viewer') + self.assertIn('--full-screen', argv) + self.assertIn('--hide-toolbar', argv) + self.assertFalse(any(arg.startswith('--spice-shared-dir') for arg in argv)) + self.assertFalse(any(arg.startswith('--spice-share-cd') for arg in argv)) + self.assertIn('--spice-usbredir-auto-redirect-filter=-1,-1,-1,-1,0', argv) + self.assertIn('--spice-usbredir-redirect-on-connect=-1,-1,-1,-1,0', argv) + self.assertIn('usb-device-select=ctrl+alt+shift+u', argv[argv.index('--hotkeys') + 1]) + self.assertEqual(argv[-1], 'spice+unix://' + str(paths.spice)) + windowed = session.build_viewer_argv(paths, 'remote-viewer', windowed=True) + self.assertNotIn('--full-screen', windowed) + self.assertNotIn('--hide-toolbar', windowed) + with self.assertRaises(ValueError): + session.viewer_title('bad*') + + def test_missing_display_and_viewer_are_recoverable_cli_errors(self): + base = ['--config', str(self.manager.config), '--state', str(self.manager.state), '--runtime', str(self.manager.runtime)] + with patch.dict(os.environ, {}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error: + self.assertEqual(main(base + ['view', 'alpha']), 1) + self.assertIn('display', error.getvalue().lower()) + with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': '/missing/remote-viewer'}, clear=True), contextlib.redirect_stderr(io.StringIO()) as error: + self.assertEqual(main(base + ['view', 'alpha']), 1) + self.assertIn('viewer', error.getvalue().lower()) + + def test_windowed_viewer_exit_leaves_vm_control_untouched(self): + viewer = self.root / 'fake-viewer' + viewer.write_text('#!/bin/sh\ncase "$1" in --help) echo --hide-toolbar; exit 0;; esac\nexit 7\n') + viewer.chmod(0o700) + self.manager.paths('alpha').spice.parent.mkdir(parents=True, exist_ok=True) + sock = socket.socket(socket.AF_UNIX) + self.addCleanup(sock.close) + sock.bind(str(self.manager.paths('alpha').spice)) + with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer)}), patch.object(self.manager, 'stop', side_effect=AssertionError('must not stop')): + self.assertEqual(session.view(self.manager, 'alpha', windowed=True), 7) + self.assertTrue(self.manager.paths('alpha').spice.exists()) + + def test_session_parser_and_default_physical_backend(self): + self.assertTrue(parser().parse_args(['session', '--nested']).nested) + physical = session.compositor_environment({'WAYLAND_DISPLAY': 'active', 'DISPLAY': ':0', 'WLR_BACKENDS': 'wayland'}, nested=False) + self.assertNotIn('WAYLAND_DISPLAY', physical) + self.assertNotIn('DISPLAY', physical) + self.assertEqual(physical['WLR_BACKENDS'], 'drm,libinput') + nested = session.compositor_environment({'WAYLAND_DISPLAY': 'active'}, nested=True) + self.assertEqual(nested['WLR_BACKENDS'], 'wayland') + with self.assertRaises(session.SessionError): + session.compositor_environment({}, nested=True) + + def test_uri_escapes_percent_query_and_fragment(self): + from dataclasses import replace + paths = replace(self.manager.paths('alpha'), spice=self.root / 'a %?#.sock') + self.assertTrue(session.build_viewer_argv(paths, 'viewer')[-1].endswith('/a%20%25%3F%23.sock')) + + def test_workspace_names_cannot_be_interpreted_as_relative_actions(self): + tree = ET.fromstring(session.build_labwc_config(['left', 'right', 'current'], ['foot'])) + self.assertEqual([a.get('to') for a in tree.findall("./windowRules/windowRule/action[@name='SendToDesktop']")], ['1', '2', '3']) + + def test_launch_writes_only_selected_private_runtime_and_serializes_paths(self): + capture = self.root / 'capture.json' + labwc = self.root / 'fake-labwc' + labwc.write_text('#!' + os.sys.executable + '\n' + '''import json, os, sys +from pathlib import Path +if '--version' in sys.argv: + print('labwc 0.20.2') +else: + config = Path(sys.argv[sys.argv.index('-C') + 1]) + manifest = json.loads((config.parent / 'session.json').read_text()) + Path(os.environ['CAPTURE']).write_text(json.dumps({'manifest': manifest, 'config': str(config), 'xml': (config / 'rc.xml').read_text(), 'env': dict(os.environ), 'autostart': (config / 'autostart').read_text()})) +''') + labwc.chmod(0o700) + old_home = self.root / 'untouched-home' + old_home.mkdir() + marker = old_home / 'keep' + marker.write_text('valuable') + with patch.dict(os.environ, {'HYDRA_LABWC': str(labwc), 'HYDRA_TERMINAL': str(labwc), 'HYDRA_BRIGHTNESS': str(labwc), 'CAPTURE': str(capture), 'HOME': str(old_home), 'DISPLAY': ':fake', 'WAYLAND_DISPLAY': 'active'}): + self.assertEqual(session.launch_session(self.manager), 0) + data = json.loads(capture.read_text()) + self.assertTrue(Path(data['config']).is_relative_to(self.manager.runtime)) + self.assertFalse(Path(data['config']).exists()) + self.assertEqual(data['manifest']['config'], str(self.manager.config)) + self.assertNotIn('DISPLAY', data['env']) + self.assertNotIn('WAYLAND_DISPLAY', data['env']) + self.assertEqual(list(old_home.iterdir()), [marker]) + self.assertEqual(marker.read_text(), 'valuable') + + def test_missing_socket_and_stock_viewer_do_not_create_disk(self): + viewer = self.root / 'viewer' + viewer.write_text('#!/bin/sh\necho --hide-toolbar\n') + viewer.chmod(0o700) + with patch.dict(os.environ, {'WAYLAND_DISPLAY': 'fake', 'HYDRA_VIEWER': str(viewer)}): + with self.assertRaisesRegex(session.SessionError, 'no SPICE socket'): + session.view(self.manager, 'alpha') + viewer.write_text('#!/bin/sh\necho stock-viewer\n') + with self.assertRaisesRegex(session.SessionError, 'adaptation'): + session.view(self.manager, 'alpha') + self.assertFalse(self.manager.paths('alpha').disk.exists()) + + def test_viewer_settings_are_isolated_and_disable_clipboard(self): + env = session._viewer_environment(self.root / 'viewer-state') + self.assertEqual(Path(env['XDG_CONFIG_HOME']).joinpath('virt-viewer/settings').read_text(), '[virt-viewer]\nshare-clipboard=false\n') + self.assertEqual(env['GSETTINGS_BACKEND'], 'memory') + + def test_initial_workspace_can_select_a_running_vm_after_stopped_profiles(self): + tree = ET.fromstring(session.build_labwc_config(['alpha', 'beta'], ['foot'], initial='beta')) + self.assertEqual(tree.findtext('./desktops/initial'), '2') + with self.assertRaises(session.SessionError): + session.build_labwc_config(['alpha'], ['foot'], initial='missing') + + def test_bad_helper_manifest_is_a_cli_error(self): + manifest = self.root / 'bad-session.json' + manifest.write_text('[]') + with contextlib.redirect_stderr(io.StringIO()) as error: + self.assertEqual(main(['_session-helper', 'recover', str(manifest)]), 1) + self.assertIn('manifest', error.getvalue().lower()) + + def test_no_running_vms_opens_recovery_without_start_or_stop(self): + terminal = self.root / 'terminal' + capture = self.root / 'recovery-argv.json' + terminal.write_text('#!' + os.sys.executable + '\nimport json, os, sys\nfrom pathlib import Path\nPath(os.environ["CAPTURE"]).write_text(json.dumps(sys.argv[1:]))\n') + terminal.chmod(0o700) + manifest = self.root / 'session.json' + manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha', 'beta'], 'terminal': str(terminal), 'directory': str(self.root)})) + with patch.dict(os.environ, {'CAPTURE': str(capture)}), patch.object(Runtime, 'start', side_effect=AssertionError('must not start')), patch.object(Runtime, 'stop', side_effect=AssertionError('must not stop')): + self.assertEqual(session.session_helper('clients', manifest), 0) + self.assertEqual(json.loads(capture.read_text())[0], '--title=HydraOS-recovery') + self.assertEqual(self.manager.status('alpha')['status'], 'stopped') + self.assertFalse(self.manager.paths('alpha').disk.exists()) + + def test_reconnect_does_not_duplicate_a_session_viewer(self): + manifest = self.root / 'session.json' + manifest.write_text(json.dumps({'config': str(self.manager.config), 'state': str(self.manager.state), 'runtime': str(self.manager.runtime), 'names': ['alpha'], 'terminal': '/bin/sh'})) + with (self.root / 'alpha.viewer.lock').open('a') as lock: + fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + with patch.object(session, 'view', side_effect=AssertionError('must not duplicate')), contextlib.redirect_stdout(io.StringIO()): + self.assertEqual(session.session_helper('reconnect:alpha', manifest), 0) + + +if __name__ == '__main__': + unittest.main()