test: add disposable laptop and base recovery acceptance harness

2026-10-13 21:57:42 +03:00
parent 261ee28f77
commit 798a5779ef
3 changed files with 666 additions and 0 deletions
@@ -0,0 +1,119 @@
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import threading
+import unittest
+import urllib.error
+import urllib.request
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / 'dev/laptop-smoke.py'
+
+
+class LaptopSmokeTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        spec = importlib.util.spec_from_file_location('laptop_smoke', SCRIPT)
+        cls.mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(cls.mod)
+
+    def test_cli_cannot_accept_existing_disks_or_private_signing_keys(self):
+        parser = self.mod.argument_parser()
+        flags = {flag for action in parser._actions for flag in action.option_strings}
+        self.assertFalse({'--disk', '--target', '--private-key', '--existing-disk'} & flags)
+        self.assertTrue({'--iso', '--ovmf-code', '--ovmf-vars'} <= flags)
+
+    def test_invariant_check_rejects_partition_foreign_data_efi_and_bootorder_changes(self):
+        original = {'gpt': {'partitions': [{'start': 2048, 'size': 1024}]},
+                    'foreign_sha256': 'data', 'foreign_efi': {'EFI/BOOT/BOOTX64.EFI': 'fallback'},
+                    'boot_order': '0001,0002'}
+        self.mod.verify_preserved(original, json.loads(json.dumps(original)))
+        for key in original:
+            changed = dict(original, **{key: 'changed'})
+            with self.subTest(key=key), self.assertRaises(self.mod.SmokeError):
+                self.mod.verify_preserved(original, changed)
+
+    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)
+        thread.start()
+        self.addCleanup(server.server_close)
+        self.addCleanup(server.shutdown)
+        url = 'http://127.0.0.1:' + str(server.server_port)
+        self.assertEqual(urllib.request.urlopen(url + '/public.pem').read(), b'TEST public')
+        for path in ('/', '/one/', '/public.pem?secret=1', '/../private.pem', '/%2e%2e/private.pem'):
+            with self.subTest(path=path), self.assertRaises(urllib.error.HTTPError):
+                urllib.request.urlopen(url + path)
+
+    def test_test_artifacts_have_real_signatures_and_no_private_key_in_share(self):
+        from hydra.update_engine import validate_manifest, validate_archive
+        with tempfile.TemporaryDirectory() as directory:
+            files = self.mod.test_artifacts(Path(directory), b'# runtime fixture\n', 'a' * 40)
+            public = Path(directory) / 'public.pem'
+            for name in ('one', 'two'):
+                manifest = validate_manifest(files['/' + name + '/manifest.json'],
+                                             files['/' + name + '/manifest.sig'], public)
+                validate_archive(files['/' + name + '/runtime.zip'], manifest)
+            self.assertEqual(set(files), {'/public.pem'} | {
+                '/' + name + '/' + file for name in ('one', 'two')
+                for file in ('manifest.json', 'manifest.sig', 'runtime.zip')})
+            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'):
+            with self.subTest(mode=mode):
+                compile(self.mod.guest_script(mode, {'sentinel': 'fixture', 'port': 1234,
+                        'checkpoint': '00000000-0000-0000-0000-000000000001',
+                        'root_uuid': '00000000-0000-0000-0000-000000000002'}), mode, 'exec')
+
+    def test_installer_only_formats_selected_root_and_registration_is_opt_in(self):
+        command = self.mod.partition_installer_command('safe123')
+        self.assertIn('--root-partition /dev/vda3', command)
+        self.assertIn('--esp-partition /dev/vda1', command)
+        self.assertIn('--confirm-format /dev/vda3', command)
+        self.assertIn('--root-fs btrfs', command)
+        self.assertIn('--register-boot-entry', command)
+        self.assertNotIn('--confirm-erase', command)
+        self.assertNotIn('--disk', command)
+        with self.assertRaises(self.mod.SmokeError):
+            self.mod.partition_installer_command('bad;token')
+
+    def test_base_state_compares_kernel_bytes_without_including_mounted_esp(self):
+        # Offline /boot/efi is an empty mountpoint; on disk-only boot it contains
+        # the mounted ESP. That mount must not produce a false base mismatch.
+        namespace = {}
+        exec(self.mod.GUEST_COMMON, namespace)
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory) / 'root'
+            esp = Path(directory) / 'esp'
+            for path, data in [('boot/vmlinuz-fixture', 'kernel'),
+                               ('boot/efi/EFI/BOOT/BOOTX64.EFI', 'mounted foreign fallback'),
+                               ('var/db/pkg/app-misc/hydra-acceptance-1.0/CONTENTS', 'package'),
+                               ('var/lib/portage/world', 'app-misc/hydra-acceptance'),
+                               ('usr/share/hydra/base-fixture.txt', '1.0')]:
+                target = root / path
+                target.parent.mkdir(parents=True, exist_ok=True)
+                target.write_text(data)
+            target = esp / 'EFI/HydraOS/grubx64.efi'
+            target.parent.mkdir(parents=True)
+            target.write_text('owned loader')
+            result = namespace['base_state'](root, esp)
+            self.assertEqual(set(result['boot']), {'vmlinuz-fixture'})
+            (root / 'boot/vmlinuz-fixture').write_text('damaged')
+            self.assertNotEqual(namespace['base_state'](root, esp)['boot'], result['boot'])
+
+    def test_real_portage_install_is_executed_in_guest_network_namespace(self):
+        import ast
+        function = next(node for node in ast.parse(self.mod.GUEST_COMMON).body
+                        if isinstance(node, ast.FunctionDef) and node.name == 'fixture')
+        commands = [node for node in ast.walk(function) if isinstance(node, ast.Call)
+                    and isinstance(node.func, ast.Name) and node.func.id == 'run']
+        emerge = next(node for node in commands if any(isinstance(arg, ast.Constant)
+                      and arg.value == 'emerge' for arg in node.args))
+        constants = [arg.value for arg in emerge.args if isinstance(arg, ast.Constant)]
+        self.assertEqual(constants[:3], ['unshare', '--net', '--'])
+        self.assertNotIn('--offline', constants, 'emerge has no --offline option')
+        self.assertIn('--ignore-default-opts', constants)
+        self.assertIn('--nodeps', constants)