257 lines
15 KiB
Python
257 lines
15 KiB
Python
import contextlib
|
|
import fcntl
|
|
import io
|
|
import json
|
|
import os
|
|
import shlex
|
|
import socket
|
|
import subprocess
|
|
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)
|
|
|
|
def _recovery_fixture(self):
|
|
for name in ('s', 'q'):
|
|
self.manager.add(Profile(name, self.iso))
|
|
paths = self.manager.paths(name)
|
|
paths.spice.parent.mkdir(parents=True)
|
|
fixture = socket.socket(socket.AF_UNIX)
|
|
self.addCleanup(fixture.close)
|
|
fixture.bind(str(paths.spice))
|
|
viewer = self.root / 'viewer'
|
|
viewer.write_text('#!' + os.sys.executable + '\n' + '''import os, sys
|
|
from pathlib import Path
|
|
from urllib.parse import unquote, urlsplit
|
|
if '--help' in sys.argv:
|
|
print('--hide-toolbar')
|
|
else:
|
|
name = Path(unquote(urlsplit(sys.argv[-1]).path)).parent.name
|
|
Path(os.environ['RECOVERY_CAPTURE'], name + '.connected').write_text(name)
|
|
''')
|
|
viewer.chmod(0o700)
|
|
shell = self.root / 'shell'
|
|
shell.write_text('#!' + os.sys.executable + '\nimport os\nfrom pathlib import Path\nPath(os.environ["RECOVERY_CAPTURE"], "shell.opened").touch()\n')
|
|
shell.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': ['s', 'q'], 'terminal': '/bin/sh'}))
|
|
return manifest, {'WAYLAND_DISPLAY': 'unused', 'HYDRA_VIEWER': str(viewer), 'RECOVERY_CAPTURE': str(self.root), 'SHELL': str(shell)}
|
|
|
|
def _check_recovery_vm_name(self, name):
|
|
manifest, env = self._recovery_fixture()
|
|
processes = []
|
|
original_popen = subprocess.Popen
|
|
def launch(*args, **kwargs):
|
|
process = original_popen(*args, **kwargs)
|
|
processes.append(process)
|
|
return process
|
|
with patch.dict(os.environ, env), patch('builtins.input', side_effect=[name, ':quit', EOFError()]), patch.object(subprocess, 'Popen', launch), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(session.session_helper('recover', manifest), 0)
|
|
for process in processes:
|
|
self.assertEqual(process.wait(timeout=5), 0)
|
|
connected = self.root / (name + '.connected')
|
|
self.assertTrue(connected.exists(), f'VM {name} was not reconnected')
|
|
self.assertEqual(connected.read_text(), name)
|
|
self.assertFalse((self.root / 'shell.opened').exists())
|
|
self.assertFalse(self.manager.paths(name).disk.exists())
|
|
|
|
def test_recovery_reconnects_vm_named_s(self):
|
|
self._check_recovery_vm_name('s')
|
|
|
|
def test_recovery_reconnects_vm_named_q(self):
|
|
self._check_recovery_vm_name('q')
|
|
|
|
def test_recovery_colon_commands_open_shell_and_close_terminal(self):
|
|
manifest, env = self._recovery_fixture()
|
|
with patch.dict(os.environ, env), patch('builtins.input', side_effect=[':shell', ':quit', EOFError()]), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(session.session_helper('recover', manifest), 0)
|
|
self.assertTrue((self.root / 'shell.opened').exists())
|
|
self.assertFalse((self.root / 's.connected').exists())
|
|
self.assertFalse((self.root / 'q.connected').exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|