feat: support installed guests and distribution display defaults
This commit is contained in:
Codex
@@ -0,0 +1,66 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from hydra.profiles import load_profile
|
||||
|
||||
|
||||
class CLITests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory(prefix='hc-')
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.root = Path(self.tmp.name)
|
||||
self.defaults = self.root / 'defaults.toml'
|
||||
self.base = [sys.executable, 'bin/hydra', '--config', str(self.root / 'config'),
|
||||
'--state', str(self.root / 'state'), '--runtime', str(self.root / 'run'),
|
||||
'--defaults', str(self.defaults)]
|
||||
|
||||
def run_cli(self, *args):
|
||||
return subprocess.run(self.base + list(args), capture_output=True, text=True, timeout=5)
|
||||
|
||||
def test_defaults_affect_new_profiles_only_and_explicit_options_win(self):
|
||||
self.defaults.write_text('network="bridge"\nbridge="hydrabr0"\nwidth=1920\nheight=1080\nrefresh_millihz=144000\n')
|
||||
added = self.run_cli('add', 'first', '--width', '1600', '--network', 'user', '--refresh-millihz', '0')
|
||||
self.assertEqual(added.returncode, 0, added.stderr)
|
||||
first = load_profile(self.root / 'config/first.toml')
|
||||
self.assertEqual((first.network, first.width, first.height, first.refresh_millihz), ('user', 1600, 1080, 0))
|
||||
added = self.run_cli('add', 'second')
|
||||
self.assertEqual(added.returncode, 0, added.stderr)
|
||||
second = load_profile(self.root / 'config/second.toml')
|
||||
self.assertEqual((second.network, second.width, second.height, second.refresh_millihz), ('bridge', 1920, 1080, 144000))
|
||||
self.defaults.write_text('width=1024\nnetwork="user"\n')
|
||||
self.assertEqual(load_profile(self.root / 'config/second.toml'), second)
|
||||
self.defaults.write_text('width="bad"\n')
|
||||
listed = self.run_cli('list')
|
||||
self.assertEqual(listed.returncode, 0, listed.stderr)
|
||||
self.assertEqual(len(json.loads(listed.stdout)), 2)
|
||||
self.assertEqual(self.run_cli('add', 'third').returncode, 1)
|
||||
self.assertFalse((self.root / 'config/third.toml').exists())
|
||||
|
||||
def test_missing_defaults_use_developer_preferences_and_media_cli(self):
|
||||
added = self.run_cli('add', 'guest')
|
||||
self.assertEqual(added.returncode, 0, added.stderr)
|
||||
profile = load_profile(self.root / 'config/guest.toml')
|
||||
self.assertEqual((profile.network, profile.width, profile.height, profile.refresh_millihz), ('user', 1280, 800, 0))
|
||||
iso = self.root / 'installer.iso'
|
||||
iso.touch()
|
||||
attached = self.run_cli('media', 'guest', '--iso', str(iso))
|
||||
self.assertEqual(attached.returncode, 0, attached.stderr)
|
||||
iso.unlink()
|
||||
listed = self.run_cli('list')
|
||||
self.assertEqual(listed.returncode, 0, listed.stderr)
|
||||
self.assertEqual(self.run_cli('status', 'guest').returncode, 0)
|
||||
detached = self.run_cli('media', 'guest', '--detach')
|
||||
self.assertEqual(detached.returncode, 0, detached.stderr)
|
||||
self.assertIsNone(load_profile(self.root / 'config/guest.toml').iso)
|
||||
|
||||
def test_unknown_or_invalid_distribution_defaults_are_rejected(self):
|
||||
for content in ('iso="/private.iso"\n', 'network="public"\n', 'refresh_millihz=true\n', '[display]\nwidth=1920\n'):
|
||||
self.defaults.write_text(content)
|
||||
result = self.run_cli('add', 'guest')
|
||||
self.assertEqual(result.returncode, 1, result.stderr)
|
||||
self.assertNotIn('Traceback', result.stderr)
|
||||
self.assertFalse((self.root / 'config/guest.toml').exists())
|
||||
+20
-3
@@ -33,9 +33,7 @@ class ProfilesTests(unittest.TestCase):
|
||||
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
|
||||
Profile('guest', self.iso, **{field: value})
|
||||
|
||||
def test_missing_iso_and_duplicate_profile_preserve_existing(self):
|
||||
with self.assertRaises(ValueError):
|
||||
Profile('guest', self.root / 'missing.iso')
|
||||
def test_duplicate_profile_preserves_existing(self):
|
||||
dest = self.root / 'guest.toml'
|
||||
save_profile(dest, Profile('guest', self.iso))
|
||||
original = dest.read_bytes()
|
||||
@@ -43,6 +41,25 @@ class ProfilesTests(unittest.TestCase):
|
||||
save_profile(dest, Profile('guest', self.iso, memory=8192))
|
||||
self.assertEqual(dest.read_bytes(), original)
|
||||
|
||||
def test_disk_only_profile_roundtrips_without_media(self):
|
||||
dest = self.root / 'guest.toml'
|
||||
save_profile(dest, Profile('guest'))
|
||||
self.assertIsNone(load_profile(dest).iso)
|
||||
self.assertNotIn('iso =', dest.read_text())
|
||||
|
||||
def test_missing_media_does_not_block_loading_profile(self):
|
||||
dest = self.root / 'guest.toml'
|
||||
save_profile(dest, Profile('guest', self.iso))
|
||||
self.iso.unlink()
|
||||
self.assertEqual(load_profile(dest).iso, self.iso)
|
||||
|
||||
def test_display_bounds_and_types(self):
|
||||
for field, values in {'width': (0, True, 16385), 'height': (0, '1080', 16385),
|
||||
'refresh_millihz': (-1, True, 1000001)}.items():
|
||||
for value in values:
|
||||
with self.subTest(field=field, value=value), self.assertRaises(ValueError):
|
||||
Profile('guest', self.iso, **{field: value})
|
||||
|
||||
def test_profile_name_must_match_filename(self):
|
||||
path = self.root / 'guest.toml'
|
||||
path.write_text(f'name = "other"\niso = "{self.iso}"\n')
|
||||
|
||||
@@ -88,6 +88,50 @@ class RuntimeTests(unittest.TestCase):
|
||||
except VMError:
|
||||
pass
|
||||
|
||||
def test_disk_boot_omits_media_and_requests_display(self):
|
||||
profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
|
||||
graphics='software')
|
||||
argv = build_qemu_argv(profile, self.runtime.paths('guest'))
|
||||
self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
|
||||
self.assertNotIn('ide-cd,drive=iso', argv)
|
||||
self.assertIn('order=c', argv)
|
||||
devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
|
||||
self.assertEqual(len(devices), 2)
|
||||
stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
|
||||
self.assertFalse(any('refresh_rate=' in arg for arg in stock))
|
||||
|
||||
def test_missing_iso_is_error_at_start_without_creating_disk(self):
|
||||
self.iso.unlink()
|
||||
with self.assertRaisesRegex(VMError, 'ISO'):
|
||||
self.runtime.start('guest')
|
||||
self.assertFalse(self.runtime.paths('guest').disk.exists())
|
||||
self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
|
||||
|
||||
def test_media_changes_only_while_stopped_and_preserves_disk(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
self.runtime.start('guest')
|
||||
disk = paths.disk.read_bytes()
|
||||
config = (self.runtime.config / 'guest.toml').read_bytes()
|
||||
for iso in (None, self.iso):
|
||||
with self.assertRaises(VMError):
|
||||
self.runtime.media('guest', iso)
|
||||
self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
|
||||
self.iso.unlink()
|
||||
self.assertEqual(self.runtime.profile('guest').iso, self.iso)
|
||||
self.runtime.stop('guest', force=True)
|
||||
self.runtime.media('guest', None)
|
||||
self.assertIsNone(self.runtime.profile('guest').iso)
|
||||
self.runtime.start('guest')
|
||||
self.runtime.stop('guest', force=True)
|
||||
self.assertEqual(paths.disk.read_bytes(), disk)
|
||||
replacement = self.root / 'replacement.iso'
|
||||
replacement.touch()
|
||||
self.runtime.media('guest', replacement)
|
||||
self.assertEqual(self.runtime.profile('guest').iso, replacement)
|
||||
with self.assertRaisesRegex(VMError, 'ISO'):
|
||||
self.runtime.media('guest', self.iso)
|
||||
self.assertEqual(self.runtime.profile('guest').iso, replacement)
|
||||
|
||||
def test_argv_is_private_and_comma_safe(self):
|
||||
paths = self.runtime.paths('guest')
|
||||
argv = build_qemu_argv(self.profile, paths)
|
||||
|
||||
Reference in New Issue
Block a user