Add terminal VM menu and focused session handoff
This commit is contained in:
Codex
@@ -0,0 +1,168 @@
|
||||
import contextlib
|
||||
import importlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from hydra.profiles import Profile
|
||||
from hydra.runtime import Runtime, RuntimeError
|
||||
|
||||
|
||||
class MenuTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix='hm-')
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.manager = Runtime(self.root / 'config', self.root / 'state', self.root / 'run')
|
||||
|
||||
def menu(self):
|
||||
self.assertIsNotNone(importlib.util.find_spec('hydra.menu'), 'Terminal menu is missing')
|
||||
return importlib.import_module('hydra.menu')
|
||||
|
||||
def run_choices(self, choices, **kwargs):
|
||||
output = io.StringIO()
|
||||
with patch('builtins.input', side_effect=choices), contextlib.redirect_stdout(output):
|
||||
code = self.menu().run_menu(self.manager, **kwargs)
|
||||
self.assertEqual(code, 0)
|
||||
return output.getvalue()
|
||||
|
||||
def test_display_is_readable_and_never_starts_a_vm(self):
|
||||
self.manager.add(Profile('alpha', memory=512))
|
||||
with patch.object(Runtime, 'start', side_effect=AssertionError('display started a VM')):
|
||||
output = self.run_choices([':quit'])
|
||||
for label in ('Hydra OS', 'alpha', 'stopped', '512 MiB', ':run', ':network', ':shell'):
|
||||
self.assertIn(label, output)
|
||||
self.assertNotIn('"status":', output)
|
||||
self.assertFalse(self.manager.paths('alpha').disk.exists())
|
||||
|
||||
def test_number_selection_persists_and_enter_reuses_it(self):
|
||||
self.manager.add(Profile('alpha'))
|
||||
self.manager.add(Profile('beta'))
|
||||
module = self.menu()
|
||||
selected = []
|
||||
def open_vm(runtime, name, **kwargs):
|
||||
selected.append(name)
|
||||
return 0
|
||||
with patch.object(module, 'open_vm', side_effect=open_vm):
|
||||
self.run_choices(['2', ':quit'])
|
||||
self.run_choices(['', ':quit'])
|
||||
self.assertEqual(selected, ['beta', 'beta'])
|
||||
self.assertEqual(json.loads((self.manager.state / 'menu.json').read_text()), {'last_selected': 'beta'})
|
||||
self.assertEqual((self.manager.state / 'menu.json').stat().st_mode & 0o777, 0o600)
|
||||
|
||||
def test_names_s_and_q_remain_vm_selections(self):
|
||||
for name in ('s', 'q'):
|
||||
self.manager.add(Profile(name))
|
||||
selected = []
|
||||
with patch.object(self.menu(), 'open_vm', side_effect=lambda runtime, name, **kw: selected.append(name) or 0):
|
||||
self.run_choices(['s', 'q', ':quit'])
|
||||
self.assertEqual(selected, ['s', 'q'])
|
||||
|
||||
def test_invalid_profile_and_unknown_identity_leave_other_vms_accessible(self):
|
||||
self.manager.add(Profile('alpha'))
|
||||
self.manager.add(Profile('beta'))
|
||||
(self.manager.config / 'bad.toml').write_text('memory = "invalid"')
|
||||
self.manager.paths('alpha').spice.parent.mkdir(parents=True)
|
||||
self.manager.paths('alpha').spice.touch()
|
||||
module = self.menu()
|
||||
with patch.object(module, 'open_vm', side_effect=RuntimeError('alpha: unknown identity; inspect status')):
|
||||
output = self.run_choices(['alpha', ':help', ':quit'])
|
||||
self.assertIn('bad.toml', output)
|
||||
self.assertIn('unknown', output)
|
||||
self.assertIn('beta', output)
|
||||
self.assertFalse((self.manager.state / 'menu.json').exists())
|
||||
|
||||
def test_empty_or_removed_selection_does_not_launch_and_shell_can_repair_profiles(self):
|
||||
module = self.menu()
|
||||
def repair(argv):
|
||||
self.manager.add(Profile('fresh'))
|
||||
return type('Result', (), {'returncode': 0})()
|
||||
with patch.object(module.subprocess, 'run', side_effect=repair), patch.object(module, 'open_vm', return_value=0) as launch:
|
||||
output = self.run_choices(['', ':shell', '', ':quit'])
|
||||
self.assertIn('No VM profiles', output)
|
||||
self.assertEqual([call.args[1] for call in launch.call_args_list], ['fresh'])
|
||||
|
||||
def test_missing_network_tool_and_cancelled_action_return_to_menu(self):
|
||||
module = self.menu()
|
||||
with patch.object(module.shutil, 'which', return_value=None):
|
||||
output = self.run_choices([':network', ':run', '', ':quit'])
|
||||
self.assertIn('nmtui', output)
|
||||
self.assertIn('cancel', output.lower())
|
||||
|
||||
def test_refresh_and_successful_session_selection_return_to_compositor(self):
|
||||
self.manager.add(Profile('beta'))
|
||||
manifest = self.root / 'manifest.json'
|
||||
module = self.menu()
|
||||
with patch.object(module, 'open_vm', return_value=0) as launch:
|
||||
self.run_choices([''], manifest=manifest)
|
||||
self.assertEqual(launch.call_args.args[1], 'beta')
|
||||
self.assertEqual(launch.call_args.kwargs['manifest'], manifest)
|
||||
with patch.object(module, 'request_refresh', return_value=0) as refresh:
|
||||
self.run_choices([':refresh'], manifest=manifest)
|
||||
self.assertEqual(refresh.call_args.args, (manifest,))
|
||||
|
||||
def test_orderly_shutdown_and_detach_are_explicit_and_do_not_force_stop(self):
|
||||
iso = self.root / 'install.iso'; iso.touch()
|
||||
self.manager.add(Profile('alpha', iso))
|
||||
module = self.menu()
|
||||
with patch.object(Runtime, 'stop', side_effect=AssertionError('must not force stop')), patch.object(Runtime, 'shutdown', return_value={'status': 'shutdown-requested'}) as shutdown:
|
||||
self.run_choices([':shutdown alpha', ':detach alpha', ':quit'])
|
||||
self.assertEqual(shutdown.call_args.args, ('alpha',))
|
||||
self.assertIsNone(self.manager.profile('alpha').iso)
|
||||
|
||||
|
||||
def test_download_renders_safe_preview_and_transfer_progress(self):
|
||||
module = self.menu()
|
||||
iso = self.root / 'saved.iso'
|
||||
def transfer(url, *, sha256=None, progress=None):
|
||||
self.assertEqual(url, 'https://example.test/live.iso?token=private-token')
|
||||
self.assertIsNotNone(progress, 'Download menu needs progress feedback')
|
||||
progress(1048576, 2097152)
|
||||
iso.write_bytes(b'complete fixture')
|
||||
progress(2097152, 2097152)
|
||||
return iso
|
||||
with patch.object(module, 'download_iso', side_effect=transfer):
|
||||
output = self.run_choices([':download https://example.test/live.iso?token=private-token', '', ':quit'])
|
||||
self.assertEqual(iso.read_bytes(), b'complete fixture')
|
||||
self.assertIn('2.0 / 2.0 MiB', output)
|
||||
self.assertIn(str(iso), output)
|
||||
self.assertNotIn('private-token', output)
|
||||
|
||||
def test_guided_run_applies_allocations_and_returns_after_selected_handoff(self):
|
||||
module = self.menu()
|
||||
def launch(runtime, source, **kwargs):
|
||||
self.assertEqual(source, '/tmp/example.iso')
|
||||
self.assertEqual(kwargs['disk_size'], '8G')
|
||||
self.assertEqual(kwargs.get('defaults'), Path('/etc/hydra/defaults.toml'))
|
||||
runtime.add(Profile(kwargs['name'], memory=kwargs['memory'], cpus=kwargs['cpus']))
|
||||
return 0
|
||||
with patch.object(module, 'run_source', side_effect=launch):
|
||||
self.run_choices([':run /tmp/example.iso', 'new-vm', '512', '1', '8G', ''], manifest=self.root / 'manifest.json')
|
||||
self.assertEqual(self.manager.profile('new-vm').memory, 512)
|
||||
self.assertEqual(self.manager.profile('new-vm').cpus, 1)
|
||||
self.assertEqual(json.loads((self.manager.state / 'menu.json').read_text()), {'last_selected': 'new-vm'})
|
||||
|
||||
def test_duplicate_guided_run_does_not_download_or_replace_profile(self):
|
||||
self.manager.add(Profile('alpha', memory=512))
|
||||
before = (self.manager.config / 'alpha.toml').read_bytes()
|
||||
with patch.object(self.menu(), 'run_source', side_effect=AssertionError('existing name must not launch')):
|
||||
output = self.run_choices([':run https://example.test/live.iso', 'alpha', ':quit'])
|
||||
self.assertIn('hydra open alpha', output)
|
||||
self.assertEqual((self.manager.config / 'alpha.toml').read_bytes(), before)
|
||||
|
||||
def test_corrupt_saved_selection_and_removed_vm_do_not_block_menu(self):
|
||||
self.manager.state.mkdir()
|
||||
(self.manager.state / 'menu.json').write_text('null')
|
||||
self.run_choices(['', ':quit'])
|
||||
(self.manager.state / 'menu.json').write_text('{"last_selected": "removed"}')
|
||||
with patch.object(self.menu(), 'open_vm', side_effect=AssertionError('removed VM must not open')):
|
||||
self.run_choices(['', ':quit'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user