feat: add fullscreen HydraOS session
This commit is contained in:
Codex
@@ -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()
|
||||
Reference in New Issue
Block a user