Emil_Shanaty/Hydra-OS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix raw USB live boot and verify both installation media paths

Browse Source
This commit is contained in:
Codex
2026-10-14 07:37:24 +03:00
parent c26442f6f5
commit b54693be0c
11 changed files with 339 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_image.py
+21
View File
@@ -501,6 +501,27 @@ class ImageBuildTests(unittest.TestCase):
path.write_text('[{"package":"app-emulation/qemu-8.0", "use": [], "build_time": null}]')
self.assertFalse(self.builder.complete('packages'))
def test_iso_refuses_hfs_hybrid_that_shadows_live_volume_on_usb(self):
from hydra.build import BuildError
self.inventory()
self.builder.complete = lambda step: step == 'rootfs'
self.builder.assembly = lambda: 'isolated-assembler'
artifacts = self.builder.work / 'artifacts'
def pod(*args, **kwargs):
self.assertEqual(args[:2], ('exec', 'isolated-assembler'))
(artifacts / 'hydra-os.iso').write_bytes(b'ISO fixture')
(artifacts / 'iso-inspection.txt').write_text(
'Volume id : HYDRA_OS\n'
'El Torito boot img : 1 UEFI y none\n'
'System area summary: MBR protective-msdos-label GPT APM\n'
'GPT partname local : 3 HFSPLUS\n')
self.builder.pod = pod
with self.assertRaisesRegex(BuildError, 'HFS|APM'):
self.builder.iso()
self.assertFalse(self.builder.manifest()['iso']['complete'])
def test_malformed_inventory_with_matching_digest_is_refused(self):
for data in ['{', '{}', '[]', '[{"package":"wrong"}]',
'[{"package":"app-emulation/qemu-9.2.4", "use":"spice", "build_time":"123"}]',
tests/test_iso_smoke.py
+111 -3
View File
@@ -63,6 +63,103 @@ class SmokeTests(unittest.TestCase):
self.assertNotIn('media=cdrom', ' '.join(argv))
self.assertNotIn('-cdrom', argv)
def test_usb_live_boot_uses_readonly_raw_storage_and_explicit_bootindex(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
iso = self.root / 'trusted.iso'
iso.write_bytes(b'ISO')
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, iso, 2048, live_media='usb')
drives = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-drive']
devices = [argv[i + 1] for i, value in enumerate(argv[:-1]) if value == '-device']
backend = next(dict(field.split('=', 1) for field in drive.split(','))
for drive in drives if str(iso) in drive)
self.assertEqual(backend['file'], str(iso))
self.assertEqual(backend['if'], 'none')
self.assertEqual(backend['format'], 'raw')
self.assertEqual(backend['readonly'], 'on')
self.assertNotIn('media', backend, 'USB must exercise a disk rather than a CD-ROM')
self.assertTrue(any(device.startswith('qemu-xhci,') for device in devices))
storage = next(device for device in devices if device.startswith('usb-storage,'))
self.assertIn('drive=' + backend['id'], storage)
self.assertIn('bootindex=1', storage)
self.assertNotIn('order=d', ' '.join(argv))
self.assertNotIn('media=cdrom', ' '.join(argv))
def test_usb_installed_reboot_removes_live_medium_and_keeps_disk_boot(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
argv = self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, None, 2048, live_media='usb')
self.assertNotIn('usb-storage', ' '.join(argv))
self.assertNotIn('media=cdrom', ' '.join(argv))
self.assertIn('order=c,strict=on', argv)
def test_usb_cannot_be_combined_with_recovery_optical_topology(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
for medium in (self.code, None):
with self.subTest(medium=medium):
with self.assertRaisesRegex(self.mod.SmokeError, 'keep_cdrom_drive.*usb'):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, medium, 2048, live_media='usb', keep_cdrom_drive=True)
def test_usb_rejects_devices_and_ambiguous_paths_as_live_media(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
ambiguous = self.root / 'image,readonly=off.iso'
ambiguous.write_bytes(b'ISO')
for medium in (Path('/dev/null'), ambiguous):
with self.subTest(medium=medium):
with self.assertRaises(self.mod.SmokeError):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, medium, 2048, live_media='usb')
def test_unknown_live_media_is_refused_instead_of_silently_booting_cd(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
with self.assertRaisesRegex(self.mod.SmokeError, 'live media'):
self.mod.qemu_argv('qemu', self.root, disk, self.code,
self.variables, self.code, 2048, live_media='typo')
def test_live_only_checks_boot_and_graphics_without_installing_or_rebooting(self):
import json
from unittest.mock import Mock, patch
iso = self.root / 'trusted.iso'
iso.write_bytes(b'ISO')
vm = Mock()
vm.kvm = {'enabled': True}
vm.console.command.return_value = b'live probe passed'
with patch.object(self.mod, 'VM', return_value=vm) as create_vm, \
patch.object(self.mod.subprocess, 'run') as create_disk:
try:
status = self.mod.main(['--iso', str(iso), '--ovmf-code', str(self.code),
'--ovmf-vars', str(self.variables), '--scratch-parent', str(self.root),
'--live-media', 'usb', '--live-only'])
except SystemExit as exc:
self.fail(f'Live-only USB acceptance must be supported: {exc}')
self.assertEqual(status, 0)
create_vm.assert_called_once()
args, scratch, disk, medium, _password, phase = create_vm.call_args.args
self.assertEqual(args.live_media, 'usb')
self.assertEqual(medium, iso)
self.assertEqual(phase, 'live')
self.assertEqual(disk.parent, scratch)
self.assertEqual(create_disk.call_args.args[0][-2], str(disk))
self.assertTrue(disk.is_file(), 'Only a newly reserved disposable disk is used')
vm.console.send.assert_called_once_with('sudo -n poweroff\n')
vm.console.expect_installer.assert_not_called()
self.assertEqual([call.args[0] for call in vm.console.command.call_args_list],
[self.mod.LIVE_CHECK, self.mod.LIVE_GRAPHICAL_CHECK])
vm.wait_poweroff.assert_called_once_with(90)
vm.close.assert_called_once()
report = json.loads((scratch / 'report.json').read_text())
self.assertEqual(report['status'], 'passed')
self.assertEqual(report['live_media'], 'usb')
self.assertEqual(report['scope'], 'live-only')
self.assertEqual(set(report['checks']),
{'live_qmp_kvm', 'live_boot', 'live_graphical_session', 'live_poweroff'})
def test_persistent_optical_drive_keeps_topology_with_no_installed_medium(self):
disk = self.root / 'disk.qcow2'
self.mod.reserve_disk(disk)
@@ -289,7 +386,7 @@ class SmokeTests(unittest.TestCase):
self.mod.qemu_argv('qemu', self.root, comma, self.code,
self.variables, None, 2048)
def probe_fixture(self, case, live=False):
def probe_fixture(self, case, live=False, graphical=False):
# Run the production shell against ordinary scratch files. Commands that
# query guest state are boundary fixtures; grep/find/test are real tools.
import os
@@ -349,16 +446,19 @@ elif name == 'getent':
else: print('127.0.0.1 STREAM example')
elif name == 'ip': print('10.203.0.1/24' if 'addr' in args else 'default via 10.0.2.2')
elif name == 'nmcli': print('connected')
elif name == 'pgrep':
sys.exit(1 if args[-1] == case else 0)
else: print('fixture tool')
''')
dispatcher.chmod(0o755)
for command in ('id', 'findmnt', 'systemctl', 'getent', 'ip', 'nft',
'nmcli', 'sudo', 'hydra', 'hydra-install',
'nmcli', 'pgrep', 'sudo', 'hydra', 'hydra-install',
'qemu-system-x86_64', 'qemu-img', 'remote-viewer', 'foot', 'labwc'):
path = commands / command
if not path.exists():
path.symlink_to(dispatcher)
probe = self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK
probe = (self.mod.LIVE_GRAPHICAL_CHECK if graphical else
self.mod.LIVE_CHECK if live else self.mod.INSTALLED_CHECK)
for path in ('/etc/', '/usr/share', '/proc/cmdline', '/sys/firmware/efi', '/run/initramfs/live/LiveOS/squashfs.img'):
probe = probe.replace(path, str(root) + path)
import shutil
@@ -405,6 +505,14 @@ else: print('fixture tool')
with self.subTest(service=service):
self.assertNotEqual(self.probe_fixture(service, live=True), 0)
def test_live_graphical_probe_requires_greetd_compositor_and_terminal(self):
self.assertTrue(hasattr(self.mod, 'LIVE_GRAPHICAL_CHECK'),
'Live-only acceptance needs graphical session evidence')
self.assertEqual(self.probe_fixture('clean', live=True, graphical=True), 0)
for case in ('greetd', 'labwc', 'foot'):
with self.subTest(case=case):
self.assertNotEqual(self.probe_fixture(case, live=True, graphical=True), 0)
def tcp_probe(self, fragments):
# Execute the exact Python client embedded in the guest network probe,
# redirecting only its address to a fresh loopback socket.