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

fix: preserve installed helpers, adaptations and inventory provenance

Browse Source
This commit is contained in:
Codex
2026-10-13 04:40:18 +03:00
parent 0f71430b8e
commit de1a1698a9
5 changed files with 229 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_image.py
+144
View File
@@ -1,5 +1,8 @@
import importlib.util
import json
import os
import shlex
import sys
from pathlib import Path
import subprocess
import tempfile
@@ -90,6 +93,65 @@ class ImageTests(unittest.TestCase):
self.assertIn('binpkg-request-signature', features)
self.assertEqual((self.root / 'usr/share/hydra/build-manifest.json').is_file(), True)
def test_installed_helper_uses_relocated_payload_and_opens_empty_recovery(self):
self.kernel()
image.configure_rootfs(self.root, Path(__file__).resolve().parents[1])
terminal = Path(self.tmp.name) / 'terminal'
terminal.write_text('#!/bin/sh\nshift\nexec "$@"\n')
terminal.chmod(0o755)
manifest = Path(self.tmp.name) / 'session.json'
manifest.write_text(json.dumps({'config': str(Path(self.tmp.name) / 'config'),
'state': str(Path(self.tmp.name) / 'state'),
'runtime': str(Path(self.tmp.name) / 'runtime'),
'terminal': str(terminal), 'names': []}))
script = r"""import json, pathlib, subprocess, sys
sys.path.insert(0, sys.argv[1])
from hydra.session import _module_argv
helper = _module_argv('clients', pathlib.Path(sys.argv[2]))
help_result = subprocess.run(helper[:2] + ['--help'], text=True, capture_output=True)
print(help_result.stdout)
print(help_result.stderr, file=sys.stderr)
if help_result.returncode:
sys.exit(help_result.returncode)
result = subprocess.run(helper, input=':quit\n', text=True, capture_output=True, timeout=10)
print(result.stdout)
print(result.stderr, file=sys.stderr)
sys.exit(result.returncode)
"""
result = subprocess.run([sys.executable, '-I', '-c', script, str(self.root / 'usr/lib/hydra'), str(manifest)],
cwd=self.tmp.name, text=True, capture_output=True, timeout=20)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn('HydraOS local recovery', result.stdout)
self.assertIn('add', result.stdout)
def test_installed_updates_exclude_unadapted_qemu_and_viewer_binaries(self):
self.kernel()
source = Path(__file__).resolve().parents[1]
self.file('etc/portage/make.conf', (source / 'image/portage/make.conf').read_text())
image.configure_rootfs(self.root, source)
settings = subprocess.run(['bash', '-euc', 'source "$1"; printf "%s" "$EMERGE_DEFAULT_OPTS"',
'--', str(self.root / 'etc/portage/make.conf')], text=True, capture_output=True, check=True)
defaults = shlex.split(settings.stdout)
exclusions = [item.split('=', 1)[1].split() for item in defaults if item.startswith('--usepkg-exclude=')]
self.assertEqual(exclusions, [['app-emulation/qemu', 'app-emulation/virt-viewer']])
self.assertIn('--getbinpkg', defaults)
def test_export_enables_pulse_socket_without_starting_user_services(self):
source = Path(__file__).resolve().parents[1]
self.assertTrue(callable(getattr(image, 'enable_user_audio', None)), 'user audio enable is missing')
units = self.root / 'usr/lib/systemd/user'
units.mkdir(parents=True)
for name, section in [('pipewire.socket', 'Socket'), ('pipewire-pulse.socket', 'Socket'),
('wireplumber.service', 'Service')]:
target = 'sockets.target' if section == 'Socket' else 'default.target'
(units / name).write_text(f'[{section}]\n[Install]\nWantedBy={target}\n')
image.enable_user_audio(self.root)
for target, name in [('sockets.target', 'pipewire.socket'), ('sockets.target', 'pipewire-pulse.socket'),
('default.target', 'wireplumber.service')]:
link = self.root / 'etc/systemd/user' / (target + '.wants') / name
self.assertTrue(link.is_symlink(), name)
self.assertEqual(os.readlink(link), '/usr/lib/systemd/user/' + name)
def test_rootfs_configuration_separates_live_overrides_and_installed_auth(self):
source = Path(__file__).resolve().parents[1]
image.copy_overlay(self.root, source / 'image/rootfs')
@@ -154,6 +216,88 @@ class ImageBuildTests(unittest.TestCase):
self.project = Path(self.tmp.name)
self.builder = Builder(self.project, self.project / 'build', self.project / 'cache', podman='podman')
def inventory(self, records=None):
records = records if records is not None else [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]
self.builder.work.mkdir(exist_ok=True)
artifacts = self.builder.work / 'artifacts'
artifacts.mkdir(exist_ok=True)
path = artifacts / 'packages.json'
path.write_text(json.dumps(records))
original = self.builder.complete
self.builder.complete = lambda step: True if step == 'prepare' else original(step)
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
return path
def test_changed_inventory_invalidates_package_checkpoint(self):
path = self.inventory()
self.assertTrue(self.builder.complete('packages'))
path.write_text('[{"package":"app-emulation/qemu-8.0", "use": [], "build_time": null}]')
self.assertFalse(self.builder.complete('packages'))
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"}]',
'[{"package":"app-emulation/qemu-9.2.4", "use":[], "build_time":42}]']:
with self.subTest(data=data):
path = self.inventory()
path.write_text(data)
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
self.assertFalse(self.builder.complete('packages'))
def test_changed_inventory_invalidates_completed_rootfs_and_iso(self):
path = self.inventory()
artifacts = path.parent
self.builder.active_package_job = lambda: False
self.builder.assembly = lambda snapshot=None: 'isolated-assembler'
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
def pod(*args, **kwargs):
if args[0] == 'export':
(artifacts / 'builder-rootfs.tar').write_bytes(b'exported fixture')
elif args[0] == 'exec' and 'configure-rootfs.sh' in args[-1]:
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
(artifacts / name).write_bytes(b'artifact')
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
'packages': [{'package': 'app-emulation/qemu-9.2.4', 'use': ['spice'], 'build_time': '123'}]}))
elif args[0] == 'exec' and 'grub-mkrescue' in args[-1]:
(artifacts / 'hydra-os.iso').write_bytes(b'ISO fixture')
(artifacts / 'iso-inspection.txt').write_text('HYDRA_OS\nEl Torito boot img : 1 UEFI\n')
else:
self.fail('Unexpected Podman operation: ' + repr(args))
self.builder.pod = pod
self.builder.rootfs()
self.builder.iso()
self.assertTrue(self.builder.complete('rootfs'))
self.assertTrue(self.builder.complete('iso'))
path.write_text('[{"package":"app-emulation/qemu-9.2.5", "use":[], "build_time":null}]')
self.builder.record('packages', inventory='artifacts/packages.json', inventory_sha256=image.digest(path))
self.assertTrue(self.builder.complete('packages'))
self.assertFalse(self.builder.complete('rootfs'))
self.assertFalse(self.builder.complete('iso'))
def test_export_inventory_must_match_validated_package_checkpoint(self):
from hydra.build import BuildError
path = self.inventory()
self.builder.complete = lambda step: step == 'packages'
self.builder.active_package_job = lambda: False
self.builder.assembly = lambda snapshot: 'isolated-assembler'
self.builder.run = lambda *args, **kwargs: subprocess.CompletedProcess(args, 0, 'commit\n', '')
def pod(*args, **kwargs):
artifacts = self.builder.work / 'artifacts'
if args[0] == 'export':
(artifacts / 'builder-rootfs.tar').write_bytes(b'exported fixture')
elif args[0] == 'exec':
for name in ('vmlinuz', 'initramfs-live.img', 'squashfs.img'):
(artifacts / name).write_bytes(b'artifact')
# Actual exported package differs from the checkpoint inventory.
(artifacts / 'rootfs-manifest.json').write_text(json.dumps({'kernel': '6.12.58', 'source_files': {},
'packages': [{'package': 'app-emulation/qemu-9.2.5', 'use': ['spice'], 'build_time': '456'}]}))
else:
self.fail('Unexpected Podman operation: ' + repr(args))
self.builder.pod = pod
with self.assertRaisesRegex(BuildError, 'inventory'):
self.builder.rootfs()
self.assertFalse(self.builder.manifest()['rootfs']['complete'])
def test_changed_canonical_patch_invalidates_package_input_identity(self):
patches = self.project / 'patches'
patches.mkdir()