420 lines
24 KiB
Python
420 lines
24 KiB
Python
import contextlib
|
|
import fcntl
|
|
import io
|
|
import json
|
|
import os
|
|
import shlex
|
|
import shutil
|
|
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, RuntimeError as VMError
|
|
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], ['Hydra OS: alpha', 'Hydra OS: 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_empty_session_has_one_workspace_without_empty_names(self):
|
|
tree = ET.fromstring(session.build_labwc_config([], ['foot']))
|
|
self.assertEqual(tree.find('./desktops').get('number'), '1')
|
|
self.assertIsNone(tree.find('./desktops/names'))
|
|
|
|
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_with_missing_iso_leaves_vm_control_untouched(self):
|
|
self.iso.unlink()
|
|
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_empty_launch_refresh_rebuilds_profile_snapshot(self):
|
|
manager = Runtime(self.root / 'empty-config', self.root / 'empty-state', self.root / 'empty-run')
|
|
capture = self.root / 'refresh-capture.json'
|
|
compositor = self.root / 'refresh-labwc'
|
|
compositor.write_text('#!' + os.sys.executable + '\n' + r'''import json, os, subprocess, sys
|
|
from pathlib import Path
|
|
if '--version' in sys.argv:
|
|
print('labwc 0.20.2')
|
|
sys.exit(0)
|
|
config = Path(sys.argv[sys.argv.index('-C') + 1])
|
|
manifest = config.parent / 'session.json'
|
|
data = json.loads(manifest.read_text())
|
|
capture = Path(os.environ['CAPTURE'])
|
|
seen = json.loads(capture.read_text()) if capture.exists() else []
|
|
seen.append({'names': data['names'], 'xml': (config / 'rc.xml').read_text()})
|
|
capture.write_text(json.dumps(seen))
|
|
if len(seen) == 1:
|
|
launcher = os.environ['HYDRA_SOURCE_LAUNCHER']
|
|
for name in ('first', 'second'):
|
|
subprocess.run([sys.executable, launcher, '--config', data['config'], '--state', data['state'],
|
|
'--runtime', data['runtime'], 'add', name], check=True)
|
|
subprocess.run([sys.executable, launcher, '_session-helper', 'recover', str(manifest)],
|
|
input=':refresh\n', text=True, check=True)
|
|
sys.exit(67)
|
|
sys.exit(0)
|
|
''')
|
|
compositor.chmod(0o700)
|
|
with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
|
|
'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture),
|
|
'HYDRA_SOURCE_LAUNCHER': str(Path('bin/hydra').resolve())}), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(session.launch_session(manager), 0)
|
|
seen = json.loads(capture.read_text())
|
|
self.assertEqual([entry['names'] for entry in seen], [[], ['first', 'second']])
|
|
tree = ET.fromstring(seen[1]['xml'])
|
|
self.assertEqual(tree.find('desktops').get('number'), '2')
|
|
self.assertEqual([rule.get('title') for rule in tree.findall('./windowRules/windowRule')],
|
|
['Hydra OS: first', 'Hydra OS: second'])
|
|
self.assertFalse(manager.paths('first').disk.exists())
|
|
|
|
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=Hydra OS 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):
|
|
import importlib.util
|
|
self.assertIsNotNone(importlib.util.find_spec('hydra.menu'), 'Terminal menu is missing')
|
|
from hydra import menu
|
|
manifest, env = self._recovery_fixture()
|
|
with patch.dict(os.environ, env), patch('builtins.input', side_effect=[name, ':quit']), \
|
|
patch.object(menu, 'open_vm', return_value=0) as launch, contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(session.session_helper('recover', manifest), 0)
|
|
self.assertEqual(launch.call_args.args[1], name)
|
|
self.assertEqual(launch.call_args.kwargs['manifest'], manifest)
|
|
self.assertFalse((self.root / 'shell.opened').exists())
|
|
self.assertFalse(self.manager.paths(name).disk.exists())
|
|
|
|
def test_recovery_selects_vm_named_s(self):
|
|
self._check_recovery_vm_name('s')
|
|
|
|
def test_recovery_selects_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())
|
|
|
|
|
|
def test_refresh_payload_selects_fresh_profile_and_legacy_empty_request_is_accepted(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'}))
|
|
self.assertTrue(hasattr(session, 'request_refresh'), 'Selected refresh API is missing')
|
|
self.assertEqual(session.request_refresh(manifest, selected='beta'), 0)
|
|
request = self.root / 'refresh.request'
|
|
self.assertEqual(json.loads(request.read_text()), {'selected': 'beta'})
|
|
self.assertEqual(request.stat().st_mode & 0o777, 0o600)
|
|
request.unlink()
|
|
self.assertEqual(session.request_refresh(manifest), 0)
|
|
self.assertEqual(request.read_bytes(), b'')
|
|
|
|
def test_refresh_rejects_invalid_or_removed_selection_without_publishing(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'}))
|
|
self.assertTrue(hasattr(session, 'request_refresh'), 'Selected refresh API is missing')
|
|
for name in ('missing', '../alpha', 'bad*'):
|
|
with self.assertRaises((ValueError, OSError)):
|
|
session.request_refresh(manifest, selected=name)
|
|
self.assertFalse((self.root / 'refresh.request').exists())
|
|
|
|
def test_selected_second_workspace_survives_compositor_refresh_without_starting_guests(self):
|
|
capture = self.root / 'selected-capture.json'
|
|
compositor = self.root / 'selected-labwc'
|
|
compositor.write_text('#!' + os.sys.executable + '\n' + r'''import json, os, sys
|
|
from pathlib import Path
|
|
if '--version' in sys.argv:
|
|
print('labwc 0.20.2')
|
|
sys.exit(0)
|
|
config = Path(sys.argv[sys.argv.index('-C') + 1])
|
|
manifest = config.parent / 'session.json'
|
|
capture = Path(os.environ['CAPTURE'])
|
|
seen = json.loads(capture.read_text()) if capture.exists() else []
|
|
seen.append({'xml': (config / 'rc.xml').read_text(), 'manifest_env': os.environ.get('HYDRA_SESSION_MANIFEST'),
|
|
'manifest_path': str(manifest)})
|
|
capture.write_text(json.dumps(seen))
|
|
if len(seen) == 1:
|
|
(manifest.parent / 'refresh.request').write_text('{"selected": "beta"}')
|
|
''')
|
|
compositor.chmod(0o700)
|
|
with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
|
|
'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture)}), \
|
|
patch.object(Runtime, 'start', side_effect=AssertionError('session must never start guests')):
|
|
self.assertEqual(session.launch_session(self.manager), 0)
|
|
seen = json.loads(capture.read_text())
|
|
self.assertEqual(len(seen), 2)
|
|
self.assertEqual(ET.fromstring(seen[1]['xml']).findtext('./desktops/initial'), '2')
|
|
self.assertTrue(all(s['manifest_env'] == s['manifest_path'] for s in seen))
|
|
self.assertFalse(self.manager.paths('alpha').disk.exists())
|
|
self.assertFalse(self.manager.paths('beta').disk.exists())
|
|
|
|
def test_recovery_failure_leaves_emergency_shell_accessible(self):
|
|
manifest, env = self._recovery_fixture()
|
|
import importlib.util
|
|
self.assertIsNotNone(importlib.util.find_spec('hydra.menu'), 'Terminal menu is missing')
|
|
from hydra import menu
|
|
with patch.dict(os.environ, env), patch.object(menu, 'run_menu', side_effect=TypeError('broken menu')), \
|
|
patch('builtins.input', side_effect=[':shell', ':quit']), contextlib.redirect_stdout(io.StringIO()) as output:
|
|
try:
|
|
code = session.session_helper('recover', manifest)
|
|
except TypeError as exc:
|
|
self.fail(f'Menu failure escaped emergency recovery: {exc}')
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('emergency', output.getvalue().lower())
|
|
self.assertTrue((self.root / 'shell.opened').exists())
|
|
|
|
|
|
def test_runtime_status_error_opens_recovery_without_starting_guests(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': shutil.which('true')}))
|
|
with patch.object(Runtime, 'status', side_effect=VMError('private runtime permission problem')), \
|
|
patch.object(Runtime, 'start', side_effect=AssertionError('must not start')), \
|
|
contextlib.redirect_stderr(io.StringIO()) as error:
|
|
try:
|
|
code = session.session_helper('clients', manifest)
|
|
except VMError as exc:
|
|
self.fail(f'Status failure escaped session recovery: {exc}')
|
|
self.assertEqual(code, 0)
|
|
self.assertIn('private runtime permission problem', error.getvalue())
|
|
|
|
def test_invalid_profile_is_excluded_from_compositor_but_valid_profiles_remain(self):
|
|
(self.manager.config / 'bad.toml').write_text('memory = "broken"')
|
|
capture = self.root / 'invalid-capture.json'
|
|
compositor = self.root / 'invalid-labwc'
|
|
compositor.write_text('#!' + os.sys.executable + '\n' + r'''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])
|
|
Path(os.environ['CAPTURE']).write_text((config.parent / 'session.json').read_text())
|
|
''')
|
|
compositor.chmod(0o700)
|
|
with patch.dict(os.environ, {'HYDRA_LABWC': str(compositor), 'HYDRA_TERMINAL': str(compositor),
|
|
'HYDRA_BRIGHTNESS': str(compositor), 'CAPTURE': str(capture)}), \
|
|
contextlib.redirect_stderr(io.StringIO()) as error:
|
|
self.assertEqual(session.launch_session(self.manager), 0)
|
|
self.assertEqual(json.loads(capture.read_text())['names'], ['alpha', 'beta'])
|
|
self.assertIn('bad.toml', error.getvalue())
|
|
|
|
|
|
def test_paused_guests_reconnect_viewers_without_start_or_resume(self):
|
|
manifest, env = self._recovery_fixture()
|
|
terminal = self.root / 'exit-terminal'
|
|
terminal.write_text('#!' + os.sys.executable + '\n')
|
|
terminal.chmod(0o700)
|
|
data = json.loads(manifest.read_text())
|
|
data['terminal'] = str(terminal)
|
|
manifest.write_text(json.dumps(data))
|
|
with patch.dict(os.environ, env), patch.object(Runtime, 'status', return_value={'status': 'paused'}), \
|
|
patch.object(Runtime, 'start', side_effect=AssertionError('must not start/resume')), \
|
|
patch.object(Runtime, 'stop', side_effect=AssertionError('must not stop')), \
|
|
contextlib.redirect_stderr(io.StringIO()):
|
|
self.assertEqual(session.session_helper('clients', manifest), 0)
|
|
for name in ('s', 'q'):
|
|
self.assertTrue((self.root / (name + '.connected')).exists(), f'Paused {name} was not reconnected')
|
|
self.assertFalse(self.manager.paths(name).disk.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|