Document terminal workflows and provide a backed-up guest runtime update
This commit is contained in:
Codex
@@ -0,0 +1,145 @@
|
||||
"""An installed-guest update must be bounded, inspectable and reversible."""
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
import zipfile
|
||||
|
||||
|
||||
class GuestUpdateTests(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.script = Path(__file__).resolve().parents[1] / 'dev/guest-update.py'
|
||||
|
||||
def setUp(self):
|
||||
self.assertTrue(self.script.is_file(), 'guest update helper must exist')
|
||||
spec = importlib.util.spec_from_file_location('guest_update', self.script)
|
||||
self.mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(self.mod)
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.base = Path(self.tmp.name)
|
||||
self.source = self.base / 'source'
|
||||
for name, value in {'hydra/__init__.py': '', 'hydra/cli.py': 'NEW = True\n',
|
||||
'bin/hydra': '#!/usr/bin/python3\n',
|
||||
'image/rootfs/etc/motd': 'Hydra OS\n',
|
||||
'image/rootfs/usr/share/hydra/fastfetch.txt': 'hydra logo\n'}.items():
|
||||
dest = self.source / name
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
dest.write_text(value)
|
||||
self.root = self.base / 'guest'
|
||||
marker = self.root / 'usr/share/hydra/build-manifest.json'
|
||||
marker.parent.mkdir(parents=True)
|
||||
marker.write_text('{"original": true}\n')
|
||||
self.old = self.root / 'usr/lib/hydra/hydra/cli.py'
|
||||
self.old.parent.mkdir(parents=True)
|
||||
self.old.write_text('OLD = True\n')
|
||||
self.bundle = self.base / 'update.pyz'
|
||||
|
||||
def build(self):
|
||||
return self.mod.build_bundle(self.source, self.bundle)
|
||||
|
||||
def test_dry_run_preserves_installed_files_and_user_disk(self):
|
||||
self.build()
|
||||
disk = self.root / 'home/user/.local/state/hydra/work/disk.qcow2'
|
||||
disk.parent.mkdir(parents=True)
|
||||
disk.write_bytes(b'user disk')
|
||||
result = self.mod.apply_bundle(self.bundle, self.root)
|
||||
self.assertFalse(result['applied'])
|
||||
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
||||
self.assertEqual(disk.read_bytes(), b'user disk')
|
||||
self.assertFalse((self.root / 'var/lib/hydra/updates').exists())
|
||||
|
||||
def test_apply_saves_backup_and_preserves_original_image_evidence(self):
|
||||
self.build()
|
||||
result = self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
||||
self.assertTrue(result['applied'])
|
||||
self.assertEqual(self.old.read_text(), 'NEW = True\n')
|
||||
backup = Path(result['backup'])
|
||||
self.assertEqual((backup / 'usr/lib/hydra/hydra/cli.py').read_text(), 'OLD = True\n')
|
||||
self.assertEqual(json.loads((self.root / 'usr/share/hydra/build-manifest.json').read_text()), {'original': True})
|
||||
self.assertTrue((self.root / 'usr/share/hydra/runtime-update.json').is_file())
|
||||
self.assertEqual((self.root / 'usr/share/hydra/fastfetch.txt').read_text(), 'hydra logo\n')
|
||||
|
||||
def test_branding_preserves_vendor_release_and_backs_up_etc_symlink(self):
|
||||
vendor = self.root / 'usr/lib/os-release'
|
||||
vendor.parent.mkdir(parents=True, exist_ok=True)
|
||||
vendor.write_text('NAME="Gentoo"\nID=gentoo\nVERSION_ID=2.18\n')
|
||||
release = self.root / 'etc/os-release'
|
||||
release.parent.mkdir(parents=True)
|
||||
release.symlink_to('../usr/lib/os-release')
|
||||
self.build()
|
||||
result = self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
||||
self.assertFalse(release.is_symlink())
|
||||
self.assertIn('PRETTY_NAME="Hydra OS"', release.read_text())
|
||||
self.assertIn('ID_LIKE=gentoo', release.read_text())
|
||||
self.assertIn('VERSION_ID=2.18', release.read_text())
|
||||
self.assertEqual(vendor.read_text(), 'NAME="Gentoo"\nID=gentoo\nVERSION_ID=2.18\n')
|
||||
saved = Path(result['backup']) / 'etc/os-release'
|
||||
self.assertTrue(saved.is_symlink())
|
||||
self.assertEqual(str(saved.readlink()), '../usr/lib/os-release')
|
||||
|
||||
def test_non_hydra_root_is_refused_before_changes(self):
|
||||
self.build()
|
||||
(self.root / 'usr/share/hydra/build-manifest.json').unlink()
|
||||
with self.assertRaisesRegex(self.mod.UpdateError, 'Hydra'):
|
||||
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
||||
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
||||
|
||||
def test_modified_payload_is_refused_before_changes(self):
|
||||
self.build()
|
||||
other = self.base / 'tampered.pyz'
|
||||
with zipfile.ZipFile(self.bundle) as src, zipfile.ZipFile(other, 'w') as dst:
|
||||
for name in src.namelist():
|
||||
data = b'CORRUPTED\n' if name == 'payload/usr/lib/hydra/hydra/cli.py' else src.read(name)
|
||||
dst.writestr(name, data)
|
||||
with self.assertRaisesRegex(self.mod.UpdateError, 'checksum'):
|
||||
self.mod.apply_bundle(other, self.root, execute=True)
|
||||
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
||||
|
||||
def test_symlink_destination_cannot_redirect_privileged_update(self):
|
||||
self.build()
|
||||
outside = self.base / 'outside'
|
||||
outside.write_text('KEEP\n')
|
||||
self.old.unlink()
|
||||
self.old.symlink_to(outside)
|
||||
with self.assertRaisesRegex(self.mod.UpdateError, '[Ss]ymlink'):
|
||||
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
||||
self.assertEqual(outside.read_text(), 'KEEP\n')
|
||||
|
||||
def test_bundle_never_overwrites_an_existing_output(self):
|
||||
self.bundle.write_bytes(b'keep this artifact')
|
||||
with self.assertRaises((FileExistsError, self.mod.UpdateError)):
|
||||
self.build()
|
||||
self.assertEqual(self.bundle.read_bytes(), b'keep this artifact')
|
||||
|
||||
def test_self_contained_bundle_applies_without_source_checkout(self):
|
||||
self.build()
|
||||
result = subprocess.run([sys.executable, str(self.bundle), '--root', str(self.root), '--apply'],
|
||||
capture_output=True, text=True, timeout=10, cwd=self.base)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(self.old.read_text(), 'NEW = True\n')
|
||||
self.assertIn('Runtime updated', result.stdout)
|
||||
|
||||
def test_write_failure_rolls_back_files_already_replaced(self):
|
||||
self.build()
|
||||
actual_write = self.mod.atomic_write
|
||||
calls = 0
|
||||
|
||||
def fail_once(*args):
|
||||
nonlocal calls
|
||||
calls += 1
|
||||
if calls == 3:
|
||||
raise OSError('injected full disk')
|
||||
return actual_write(*args)
|
||||
|
||||
with patch.object(self.mod, 'atomic_write', side_effect=fail_once):
|
||||
with self.assertRaisesRegex(OSError, 'full disk'):
|
||||
self.mod.apply_bundle(self.bundle, self.root, execute=True)
|
||||
self.assertEqual(self.old.read_text(), 'OLD = True\n')
|
||||
self.assertFalse((self.root / 'etc/motd').exists())
|
||||
self.assertFalse((self.root / 'usr/lib/hydra/bin/hydra').exists())
|
||||
Reference in New Issue
Block a user