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

fix(test): restore pruned live boot entries and verify mounted ESP

Browse Source
This commit is contained in:
Codex
2026-10-13 23:31:14 +03:00
parent ce04318b78
commit ebeebc7fb1
4 changed files with 289 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dev/iso-smoke.py
+3 -2
View File
@@ -651,8 +651,9 @@ def qemu_argv(qemu, scratch, disk, code, variables, iso, memory, *, keep_cdrom_d
if not Path(iso).is_file():
raise SmokeError('Trusted ISO must exist as a regular file')
if keep_cdrom_drive:
# Keep one SATA0 device even without media so OVMF retains its DVD boot
# entry across ISO-free boots. No file= means there is no optical medium.
# Keep SATA0 stable so a saved DVD boot option has the same device path
# when media returns. OVMF may still prune the empty drive's boot entry.
# No file= means there is no optical medium.
optical = 'if=ide,index=0,id=acceptance-cdrom,media=cdrom,readonly=on'
if iso is not None:
optical += ',format=raw,file=' + qemu_path(iso)
dev/laptop-smoke.py
+120 -7
View File
@@ -44,8 +44,11 @@ def argument_parser():
return parser
def verify_preserved(before, after):
for name in ('gpt', 'foreign_sha256', 'foreign_efi', 'boot_order'):
def verify_preserved(before, after, *, firmware_reboot=False):
fields = ('gpt', 'foreign_sha256', 'foreign_efi')
if not firmware_reboot:
fields += ('boot_order',)
for name in fields:
if before[name] != after[name]:
raise SmokeError('Preserved installation changed ' + name)
@@ -137,6 +140,72 @@ def sha(path):
def tree(path):
return {str(p.relative_to(path)): sha(p) for p in sorted(path.rglob('*')) if p.is_file()}
EFI_GUID = '8be4df61-93ca-11d2-aa0d-00e098032b8c'
DVD_LABEL = 'UEFI QEMU DVD-ROM QM00001'
def efi_description(data):
# The file starts with efivarfs attributes, EFI_LOAD_OPTION attributes and
# FilePathListLength, followed by a bounded UTF-16 description.
for end in range(10, min(len(data) - 1, 522), 2):
if data[end:end + 2] == b'\0\0':
try: return data[10:end].decode('utf-16-le').strip(), end + 2
except UnicodeDecodeError: break
return None, None
def validate_dvd_variable(data):
label, offset = efi_description(data)
path = bytes.fromhex('02010c00d041030a0000000001010600021f03120a000000ffff00007fff0400')
automatic = bytes.fromhex('4eac0881119f594d850ee21a522c59b2')
if (len(data) > 512 or data[:10] != bytes.fromhex('07000000010000002000')
or label != DVD_LABEL or data[offset:offset + 32] != path
or data[offset + 32:] not in (b'', automatic)):
raise RuntimeError('Invalid saved SATA0 DVD EFI load option')
def dvd_variables(variables):
matches = []
for path in variables.glob('Boot????-' + EFI_GUID):
if not re.fullmatch('Boot[0-9A-Fa-f]{4}-' + EFI_GUID, path.name): continue
data = path.read_bytes()
if efi_description(data)[0] == DVD_LABEL:
validate_dvd_variable(data)
matches.append((path.name[4:8], data))
if len(matches) > 1: raise RuntimeError('Ambiguous DVD EFI load options')
return matches
def capture_dvd_variable(variables):
matches = dvd_variables(variables)
if len(matches) != 1: raise RuntimeError('Expected one active SATA0 DVD EFI load option')
entry, data = matches[0]
current = (variables / ('BootCurrent-' + EFI_GUID)).read_bytes()
if len(current) != 6 or int.from_bytes(current[4:], 'little') != int(entry, 16):
raise RuntimeError('Initial live system did not boot from the captured DVD entry')
return {'entry': entry, 'raw_hex': data.hex(), 'sha256': hashlib.sha256(data).hexdigest()}
def restore_dvd_variable(variables, data):
validate_dvd_variable(data)
order = variables / ('BootOrder-' + EFI_GUID)
before = order.read_bytes()
matches = dvd_variables(variables)
created = not matches
if matches:
entry, existing = matches[0]
if existing != data: raise RuntimeError('Existing DVD option differs from captured bytes')
else:
occupied = {int(p.name[4:8], 16) for p in variables.glob('Boot????-' + EFI_GUID)
if re.fullmatch('Boot[0-9A-Fa-f]{4}-' + EFI_GUID, p.name)}
free = next((number for number in range(65536) if number not in occupied), None)
if free is None: raise RuntimeError('No unused UEFI boot variable')
entry = f'{free:04X}'
path = variables / ('Boot' + entry + '-' + EFI_GUID)
# Never overwrite an existing entry, including a concurrently created one.
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, 'wb', buffering=0) as target:
if target.write(data) != len(data): raise RuntimeError('Short EFI variable write')
if path.read_bytes() != data: raise RuntimeError('Saved DVD variable readback mismatch')
if order.read_bytes() != before: raise RuntimeError('DVD variable creation changed BootOrder')
return {'entry': entry, 'created': created, 'sha256': hashlib.sha256(data).hexdigest(),
'boot_order_sha256': hashlib.sha256(before).hexdigest()}
@contextlib.contextmanager
def mounted(device, filesystem, options=''):
import tempfile
@@ -146,10 +215,35 @@ def mounted(device, filesystem, options=''):
try: yield Path(directory)
finally: run('umount', directory)
def installed_esp(marker):
mounts = json.loads(run('findmnt', '--json', '--mountpoint', '/boot/efi', '--output',
'SOURCE,TARGET,FSTYPE,UUID,MAJ:MIN'))['filesystems']
devices = json.loads(run('lsblk', '--json', '--nodeps', '--output',
'NAME,MAJ:MIN,FSTYPE,UUID', '/dev/vda1'))['blockdevices']
if len(mounts) != 1 or len(devices) != 1:
raise RuntimeError('Expected one mounted installed ESP and one test ESP device')
mount, device = mounts[0], devices[0]
if (mount.get('source') != '/dev/vda1' or mount.get('target') != '/boot/efi'
or mount.get('fstype') != 'vfat' or device.get('name') != 'vda1'
or device.get('fstype') != 'vfat' or not device.get('maj:min')
or mount.get('maj:min') != device['maj:min'] or not device.get('uuid')
or mount.get('uuid') != device['uuid'] or marker.get('esp_uuid') != device['uuid']):
raise RuntimeError('Installed ESP mount/device identity does not match the storage marker')
return Path('/boot/efi')
@contextlib.contextmanager
def readonly_esp():
if Path('/etc/hydra/live-mode/active').exists():
with mounted('/dev/vda1', 'vfat', 'ro') as esp: yield esp
else:
# The installed ESP is already mounted rw by fstab. Verify it, then only
# read hashes; a second ro mount would conflict with its superblock state.
yield installed_esp(json.loads(Path('/etc/hydra/storage.json').read_text()))
def preserved():
table = json.loads(run('sfdisk', '--json', '/dev/vda'))['partitiontable']
# No filesystem UUIDs: selected root UUID must change on formatting.
with mounted('/dev/vda1', 'vfat', 'ro') as esp:
with readonly_esp() as esp:
foreign = {k: v for k, v in tree(esp / 'EFI').items() if not k.startswith('HydraOS/')}
order = re.search(r'^BootOrder:\s*(.*)$', run('efibootmgr'), re.M)
assert order, 'UEFI BootOrder was not observable'
@@ -202,6 +296,8 @@ def stop_child(source):
'''
GUEST_BODIES = {
'capture-live-boot': "result = capture_dvd_variable(Path('/sys/firmware/efi/efivars'))\n",
'restore-live-boot': "result = restore_dvd_variable(Path('/sys/firmware/efi/efivars'), bytes.fromhex(context['dvd_variable']))\n",
'seed': r'''
# Guest-only target is the new empty disk attached by this harness.
assert os.geteuid() == 0 and Path('/etc/hydra/live-mode/active').exists()
@@ -263,7 +359,7 @@ assert marker['root_subvolume'] == '@root'
assert Path('/home/hydratest/acceptance-home').read_text() == 'current-' + context['sentinel']
assert not Path('/var/db/pkg/app-misc/hydra-acceptance-2.0').exists()
assert Path('/usr/share/hydra/base-fixture.txt').read_text().strip() == '1.0'
with mounted('/dev/vda1', 'vfat', 'ro') as esp: result = base_state(Path('/'), esp)
with readonly_esp() as esp: result = base_state(Path('/'), esp)
result['home'] = Path('/home/hydratest/acceptance-home').read_text()
result['vm_data'] = tree(Path('/home/hydratest/.local/share/hydra/acceptance-guest'))
result['root_mount'] = run('findmnt', '-n', '-o', 'SOURCE,OPTIONS', '/')
@@ -417,7 +513,8 @@ def main(argv=None):
inputs = {str(p): iso_smoke.digest(p) for p in (args.iso, args.ovmf_code, args.ovmf_vars)}
report = {'schema': 1, 'status': 'running', 'inputs_sha256': inputs, 'checks': {},
'physical_hardware': 'pending', 'kernel_scope': 'exact kernel restoration; no two-kernel upgrade claim',
'trust_scope': 'disposable TEST key temporarily pinned only in this disposable installed guest'}
'trust_scope': 'disposable TEST key temporarily pinned only in this disposable installed guest',
'firmware_scope': 'BootOrder checked around each installer/checkpoint/restore/selector operation; firmware may prune empty-media entries between boots'}
context = {'sentinel': secrets.token_hex(32)}
password = secrets.token_urlsafe(24); vm = None; server = None
def record(name, evidence):
@@ -456,6 +553,8 @@ def main(argv=None):
threading.Thread(target=server.serve_forever, daemon=True).start()
console = boot(True, 'live-install')
before = probe(console, 'seed', context); record('seeded_foreign_invariants', before)
live_boot = probe(console, 'capture-live-boot', context)
context['dvd_variable'] = live_boot['raw_hex']; record('captured_live_boot_option', live_boot)
token = secrets.token_hex(12); console.buffer = b''
console.send(partition_installer_command(token) + '\n')
console.expect_installer(rb'Installed account password: ', 120, token); console.send(password + '\n')
@@ -472,8 +571,11 @@ def main(argv=None):
iso_smoke.installed_admin(console, password)
baseline = probe(console, 'baseline', context, 720); record('real_portage_baseline_doctor_vmdata', baseline)
record('signed_runtime_update_recovery', probe(console, 'runtime', context, 300))
record('harness_checkpoint_dvd_option', probe(console, 'restore-live-boot', context))
record('harness_checkpoint_bootnext', select_live(console)); shutdown(console)
console = boot(True, 'live-checkpoint')
checkpoint_preserved = probe(console, 'preserved', context)
verify_preserved(before, checkpoint_preserved, firmware_reboot=True)
base_before = probe(console, 'base-state', context); record('base_before_checkpoint', base_before)
context['root_uuid'] = console.command('sudo -n blkid -s UUID -o value /dev/vda3').decode().strip()
esp_partuuid = console.command('sudo -n blkid -s PARTUUID -o value /dev/vda1').decode().strip()
@@ -483,10 +585,16 @@ def main(argv=None):
cp = json.loads(console.command(command + ' --execute --confirm-root-uuid ' + shlex.quote(context['root_uuid']), 300))
context['checkpoint'] = cp['id']; record('real_offline_checkpoint', cp)
record('checkpoint_list', json.loads(console.command('sudo -n hydra-system list --root-uuid ' + shlex.quote(context['root_uuid']), 180)))
checkpoint_after = probe(console, 'preserved', context)
verify_preserved(checkpoint_preserved, checkpoint_after)
record('checkpoint_preserved_firmware_and_data', checkpoint_after)
select_installed(console, recreate=True); shutdown(console)
console = boot(False, 'installed-upgrade'); record('real_portage_upgrade_current_home', probe(console, 'upgrade', context, 720))
record('harness_restore_dvd_option', probe(console, 'restore-live-boot', context))
record('harness_restore_bootnext', select_live(console)); shutdown(console)
console = boot(True, 'live-restore')
restore_preserved = probe(console, 'preserved', context)
verify_preserved(before, restore_preserved, firmware_reboot=True)
record('deliberate_kernel_efi_damage', probe(console, 'damage', context))
record('restore_sigkill_real_btrfs', probe(console, 'restore-interrupt', context, 300))
restore = ('sudo -n hydra-system restore --root-uuid ' + shlex.quote(context['root_uuid']) +
@@ -495,13 +603,18 @@ def main(argv=None):
record('fresh_process_restore_resume', json.loads(console.command(restore, 300)))
base_after = probe(console, 'base-state', context)
if base_after != base_before: raise SmokeError('Offline restore did not restore exact base/Portage/kernel/ownedEFI hashes')
preserved_after = probe(console, 'preserved', context); verify_preserved(before, preserved_after)
preserved_after = probe(console, 'preserved', context)
verify_preserved(restore_preserved, preserved_after)
verify_preserved(before, preserved_after, firmware_reboot=True)
record('restored_base_foreign_invariants', {'base': base_after, 'preserved': preserved_after})
select_installed(console, recreate=True); shutdown(console)
console = boot(False, 'installed-restored'); restored = probe(console, 'restored', context)
if any(restored[k] != v for k, v in base_before.items()): raise SmokeError('Disk-only boot did not retain restored base')
if restored['vm_data'] != baseline['vm_data']: raise SmokeError('VM disk/profile bytes changed')
record('disk_only_restored_boot_current_home_vmdata', restored); shutdown(console)
record('disk_only_restored_boot_current_home_vmdata', restored)
final_preserved = probe(console, 'preserved', context)
verify_preserved(before, final_preserved, firmware_reboot=True)
record('final_disk_boot_foreign_invariants', final_preserved); shutdown(console)
if any(iso_smoke.digest(Path(path)) != digest for path, digest in inputs.items()): raise SmokeError('Input ISO/firmware changed')
report['status'] = 'passed'; return 0
except (SmokeError, OSError, ValueError, subprocess.SubprocessError, iso_smoke.QMPError) as exc:
docs/laptop-acceptance.md
+18 -6
View File
@@ -26,12 +26,16 @@ adds the following real guest operations:
invariants before the harness deliberately sets BootNext to choose that entry.
Subsequent live boots also use verified one-shot BootNext selection. The same
SATA DVD drive stays attached throughout, empty during installed boots; no ISO
is available to those boots. This keeps OVMF's device entries stable without
resetting firmware variables or changing BootOrder. Every deliberate BootNext
write is read back and checked against the unchanged BootOrder. If OVMF prunes
HydraOS's create-only entry after consuming BootNext, the harness recreates only
that known ESP/loader entry before the next selection, and verifies that both
registration and selection preserve the original order. Live test consoles
is available to those boots. OVMF can prune the empty DVD's automatic boot entry
despite that stable topology. The first live boot captures its verified EFI
load-option bytes; before returning to live, the harness restores those exact
bytes into an unused Boot#### variable only, verifies readback and unchanged
BootOrder, then selects BootNext. Existing variables are never overwritten. If
OVMF prunes HydraOS's create-only entry after consuming BootNext, the harness
similarly recreates only that known ESP/loader entry before later selections.
Initial installed selection still requires the installer's entry to exist.
Every deliberate registration and BootNext write is checked against unchanged
BootOrder. Live test consoles
reduce kernel console verbosity so asynchronous mount messages cannot corrupt
framed JSON; the messages remain in the guest kernel log.
2. Boot without ISO, verify the packaged Fastfetch art and doctor, install version
@@ -56,6 +60,14 @@ adds the following real guest operations:
base, package database, world and EFI hashes and all preserved foreign invariants.
Boot without ISO and verify version 1.0, current HOME and unchanged VM disk/profile
data. The previous failed root and checkpoint remain on the disposable disk.
Installed checks read the existing `/boot/efi` mount after verifying its device,
filesystem and UUID against the storage marker; they never remount it read-only.
BootOrder preservation is checked immediately around each installer, checkpoint,
restore and harness selection operation. It is not compared across firmware boots:
OVMF itself removes empty-media entries from BootOrder. GPT, foreign partition
bytes and foreign/fallback EFI bytes remain strict invariants across the entire
sequence, including all reboots. Both scopes are explicit in `report.json`.
The loopback HTTP fixture serves only explicitly named in-memory TEST artifacts
and their public key. It exposes no host directory, build tree, private key or
tests/test_laptop_smoke.py
+148 -1
View File
@@ -42,6 +42,13 @@ class BootConsole:
SCRIPT = Path(__file__).resolve().parents[1] / 'dev/laptop-smoke.py'
EFI_GUID = '8be4df61-93ca-11d2-aa0d-00e098032b8c'
# Captured efivarfs bytes from the actual OVMF SATA0 DVD boot option.
DVD_VARIABLE = bytes.fromhex(
'0700000001000000200055004500460049002000510045004d005500200044005600'
'44002d0052004f004d00200051004d00300030003000300031002000000002010c00'
'd041030a0000000001010600021f03120a000000ffff00007fff04004eac0881119f'
'594d850ee21a522c59b2')
class LaptopSmokeTests(unittest.TestCase):
@@ -67,6 +74,105 @@ class LaptopSmokeTests(unittest.TestCase):
with self.subTest(key=key), self.assertRaises(self.mod.SmokeError):
self.mod.verify_preserved(original, changed)
def test_cross_firmware_comparison_excludes_only_bootorder(self):
original = {'gpt': 'partition table', 'foreign_sha256': 'data',
'foreign_efi': {'fallback': 'bytes'}, 'boot_order': '0000,0001,0002'}
after = dict(original, boot_order='0000,0002')
self.mod.verify_preserved(original, after, firmware_reboot=True)
for name in ('gpt', 'foreign_sha256', 'foreign_efi'):
with self.subTest(name=name), self.assertRaises(self.mod.SmokeError):
self.mod.verify_preserved(original, dict(after, **{name: 'changed'}), firmware_reboot=True)
def dvd_namespace(self):
namespace = {}
exec(self.mod.GUEST_COMMON, namespace)
return namespace
def test_capture_requires_actual_current_unique_validated_dvd_option(self):
namespace = self.dvd_namespace()
with tempfile.TemporaryDirectory() as directory:
variables = Path(directory)
(variables / ('BootCurrent-' + EFI_GUID)).write_bytes(bytes.fromhex('060000000100'))
(variables / ('Boot0001-' + EFI_GUID)).write_bytes(DVD_VARIABLE)
saved = namespace['capture_dvd_variable'](variables)
self.assertEqual(saved['raw_hex'], DVD_VARIABLE.hex())
self.assertEqual(saved['entry'], '0001')
(variables / ('BootCurrent-' + EFI_GUID)).write_bytes(bytes.fromhex('060000000200'))
with self.assertRaises(RuntimeError):
namespace['capture_dvd_variable'](variables)
(variables / ('BootCurrent-' + EFI_GUID)).write_bytes(bytes.fromhex('060000000100'))
(variables / ('Boot0002-' + EFI_GUID)).write_bytes(DVD_VARIABLE)
with self.assertRaises(RuntimeError):
namespace['capture_dvd_variable'](variables)
def test_restore_saved_dvd_uses_free_variable_without_overwriting_or_reordering(self):
namespace = self.dvd_namespace()
with tempfile.TemporaryDirectory() as directory:
variables = Path(directory)
original = {'Boot0000-' + EFI_GUID: b'unrelated boot entry',
'Boot0001-' + EFI_GUID: b'HydraOS reused original DVD number',
'BootOrder-' + EFI_GUID: bytes.fromhex('0700000000000300'),
'unrelated-vendor-variable': b'foreign variable'}
for name, data in original.items():
(variables / name).write_bytes(data)
restored = namespace['restore_dvd_variable'](variables, DVD_VARIABLE)
self.assertEqual(restored['entry'], '0002')
self.assertTrue(restored['created'])
self.assertEqual((variables / ('Boot0002-' + EFI_GUID)).read_bytes(), DVD_VARIABLE)
for name, data in original.items():
self.assertEqual((variables / name).read_bytes(), data)
restored_again = namespace['restore_dvd_variable'](variables, DVD_VARIABLE)
self.assertFalse(restored_again['created'])
self.assertEqual(restored_again['entry'], '0002')
def test_restore_rejects_wrong_attributes_label_device_path_or_extra_payload(self):
namespace = self.dvd_namespace()
changes = [bytes.fromhex('06000000') + DVD_VARIABLE[4:],
DVD_VARIABLE[:4] + bytes.fromhex('00000000') + DVD_VARIABLE[8:],
DVD_VARIABLE.replace('QM00001'.encode('utf-16-le'), 'QM00003'.encode('utf-16-le')),
DVD_VARIABLE.replace(bytes.fromhex('03120a000000ffff0000'), bytes.fromhex('03120a000200ffff0000')),
DVD_VARIABLE + b'foreign optional loader data']
with tempfile.TemporaryDirectory() as directory:
variables = Path(directory)
(variables / ('BootOrder-' + EFI_GUID)).write_bytes(bytes.fromhex('070000000000'))
for data in changes:
with self.subTest(data=data.hex()), self.assertRaises(RuntimeError):
namespace['restore_dvd_variable'](variables, data)
self.assertEqual(len(list(variables.iterdir())), 1)
def test_restore_refuses_ambiguous_or_altered_existing_dvd_option(self):
namespace = self.dvd_namespace()
for altered in (False, True):
with self.subTest(altered=altered), tempfile.TemporaryDirectory() as directory:
variables = Path(directory)
(variables / ('BootOrder-' + EFI_GUID)).write_bytes(bytes.fromhex('070000000000'))
(variables / ('Boot0001-' + EFI_GUID)).write_bytes(DVD_VARIABLE)
(variables / ('Boot0002-' + EFI_GUID)).write_bytes(DVD_VARIABLE if not altered else
DVD_VARIABLE.replace(bytes.fromhex('03120a000000ffff0000'), bytes.fromhex('03120a000200ffff0000')))
with self.assertRaises(RuntimeError):
namespace['restore_dvd_variable'](variables, DVD_VARIABLE)
self.assertEqual(len(list(variables.iterdir())), 3)
def test_restore_never_overwrites_a_concurrently_created_boot_variable(self):
import os
from unittest.mock import patch
namespace = self.dvd_namespace()
with tempfile.TemporaryDirectory() as directory:
variables = Path(directory)
order = variables / ('BootOrder-' + EFI_GUID)
order.write_bytes(bytes.fromhex('070000000100'))
original_order = order.read_bytes()
destination = variables / ('Boot0000-' + EFI_GUID)
real_open = os.open
def race(path, flags, mode):
self.assertEqual(path, destination)
destination.write_bytes(b'concurrent foreign boot entry')
return real_open(path, flags, mode)
with patch.object(namespace['os'], 'open', side_effect=race), self.assertRaises(FileExistsError):
namespace['restore_dvd_variable'](variables, DVD_VARIABLE)
self.assertEqual(destination.read_bytes(), b'concurrent foreign boot entry')
self.assertEqual(order.read_bytes(), original_order)
def test_http_serves_only_explicit_bytes_with_no_filesystem_or_traversal(self):
server = self.mod.artifact_server({'/public.pem': b'TEST public', '/one/manifest.json': b'{}'})
thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -94,10 +200,11 @@ class LaptopSmokeTests(unittest.TestCase):
self.assertFalse((Path(directory) / 'TEST-private.pem').exists())
def test_every_guest_probe_compiles(self):
for mode in ('seed', 'preserved', 'baseline', 'upgrade', 'base-state', 'damage', 'restored', 'runtime', 'restore-interrupt'):
for mode in self.mod.GUEST_BODIES:
with self.subTest(mode=mode):
compile(self.mod.guest_script(mode, {'sentinel': 'fixture', 'port': 1234,
'checkpoint': '00000000-0000-0000-0000-000000000001',
'dvd_variable': DVD_VARIABLE.hex(),
'root_uuid': '00000000-0000-0000-0000-000000000002'}), mode, 'exec')
def test_installer_only_formats_selected_root_and_registration_is_opt_in(self):
@@ -215,6 +322,46 @@ class LaptopSmokeTests(unittest.TestCase):
self.assertEqual(calls[-1], ('mount', '-t', filesystem, '-o', options, device, str(directory)))
self.assertEqual(calls[-1], ('umount', str(directory)))
def test_installed_esp_is_read_from_verified_existing_mount_without_remount(self):
namespace = self.dvd_namespace()
calls = []
def query(*args):
calls.append(args)
if args[0] == 'findmnt':
return json.dumps({'filesystems': [{'source': '/dev/vda1', 'target': '/boot/efi',
'fstype': 'vfat', 'uuid': '1234-ABCD', 'maj:min': '253:1'}]})
if args[0] == 'lsblk':
return json.dumps({'blockdevices': [{'name': 'vda1', 'fstype': 'vfat',
'uuid': '1234-ABCD', 'maj:min': '253:1'}]})
raise AssertionError('ESP verification attempted a mutation: ' + repr(args))
namespace['run'] = query
with tempfile.TemporaryDirectory() as directory:
marker = Path(directory) / 'marker.json'
marker.write_text(json.dumps({'esp_uuid': '1234-ABCD'}))
def guest_path(path):
if path == '/etc/hydra/storage.json': return marker
if path == '/etc/hydra/live-mode/active': return Path(directory) / 'absent-live-mode'
return Path(path)
namespace['Path'] = guest_path
with namespace['readonly_esp']() as esp:
self.assertEqual(esp, Path('/boot/efi'))
self.assertEqual(calls, [
('findmnt', '--json', '--mountpoint', '/boot/efi', '--output', 'SOURCE,TARGET,FSTYPE,UUID,MAJ:MIN'),
('lsblk', '--json', '--nodeps', '--output', 'NAME,MAJ:MIN,FSTYPE,UUID', '/dev/vda1')])
def test_installed_esp_rejects_wrong_device_uuid_mountpoint_or_filesystem(self):
namespace = self.dvd_namespace()
mount = {'source': '/dev/vda1', 'target': '/boot/efi', 'fstype': 'vfat',
'uuid': '1234-ABCD', 'maj:min': '253:1'}
device = {'name': 'vda1', 'fstype': 'vfat', 'uuid': '1234-ABCD', 'maj:min': '253:1'}
for key, wrong in [('source', '/dev/vdb1'), ('target', '/boot'), ('fstype', 'ext4'),
('uuid', 'FFFF-ABCD'), ('maj:min', '253:2')]:
altered = dict(mount, **{key: wrong})
namespace['run'] = lambda *args: json.dumps(
{'filesystems': [altered]} if args[0] == 'findmnt' else {'blockdevices': [device]})
with self.subTest(key=key), self.assertRaises(RuntimeError):
namespace['installed_esp']({'esp_uuid': '1234-ABCD'})
def test_every_guest_mount_call_names_the_partition_filesystem(self):
import ast
expected = {'/dev/vda1': 'vfat', '/dev/vda2': 'ext4', '/dev/vda3': 'btrfs'}