kstyagi/nest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: enhance Ubuntu rootfs build process with improved logging and error handling

Browse Source
This commit is contained in:
kstyagi
2026-09-22 09:30:55 +05:30
parent 1d3644ab7c
commit bd92573158
5 changed files with 273 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_install.py
+120 -20
View File
@@ -2,8 +2,10 @@ from __future__ import annotations
import hashlib
import io
import json
import os
import shutil
import sys
import tarfile
import tempfile
import unittest
@@ -13,9 +15,9 @@ from types import SimpleNamespace
from unittest.mock import Mock, patch
from uvm.config import Settings
from uvm.errors import CommandError, UvmError
from uvm.errors import UvmError
from uvm.images import ImageStore
from uvm.install import Installer, _ROOTFS_RECIPE
from uvm.install import Installer, _GUEST_APT_SCRIPT, _ROOTFS_RECIPE
from uvm.integrity import load_manifest, write_manifest
from uvm.state import StateStore
from uvm.system import CommandResult, CommandRunner
@@ -154,7 +156,7 @@ class InstallerArchitectureTests(unittest.TestCase):
def test_only_verified_current_recipe_images_are_reused(self) -> None:
for verified, marker, force in (
(True, _ROOTFS_RECIPE, False), (False, _ROOTFS_RECIPE, False),
(True, "", False), (True, _ROOTFS_RECIPE, True),
(True, "", False), (True, "uvm-ubuntu-jammy-v1\n", False), (True, _ROOTFS_RECIPE, True),
):
with self.subTest(verified=verified, marker=marker, force=force), tempfile.TemporaryDirectory() as temporary:
settings = Settings(base=Path(temporary))
@@ -184,36 +186,109 @@ class UbuntuBuildRunner:
if "debootstrap" in args:
root = Path(args[-2])
if self.failure == "bootstrap":
raise CommandError("bootstrap interrupted")
self.write_files(root, {"debootstrap/debootstrap.log": "dpkg: --install needs at least one package archive file argument\n"})
return CommandResult(args, 1, stderr="W: Failure while unpacking the base system\n")
self.write_files(root, {
"etc/passwd": "root:x:0:0::/root:/bin/bash\n",
"etc/shadow": "root:*:1:0:99999:7:::\n", "etc/group": "root:x:0:\n",
"etc/apt/sources.list": "deb https://ports.ubuntu.com/ubuntu-ports jammy main\n",
"usr/bin/apt-get": "apt binary",
"var/lib/dpkg/status": "Package: apt\nStatus: install ok installed\nVersion: 1.0\n",
"var/lib/dpkg/info/apt.list": "/usr/bin/apt-get\n",
"var/log/bootstrap.log": "Base system installed successfully.\n",
})
for relative in ("dev", "proc", "sys", "usr/sbin"):
(root / relative).mkdir(parents=True, exist_ok=True)
(root / "etc/resolv.conf").symlink_to("../run/systemd/resolve/stub-resolv.conf")
# Ubuntu's real script removes this, even if it existed before bootstrap.
(root / "usr/sbin/policy-rc.d").unlink(missing_ok=True)
return CommandResult(args, 0, stdout="I: Base system installed successfully.\n")
if "uvm-guest-packages" in args:
root = Path(args[args.index("uvm-guest-packages") + 1])
if (root / "usr/sbin/policy-rc.d").read_text() != "#!/bin/sh\nexit 101\n":
raise AssertionError("guest services must be disabled during apt provisioning")
sources = (root / "etc/apt/sources.list").read_text()
if "jammy-updates main universe" not in sources or "jammy-security main universe" not in sources:
raise AssertionError("apt must see updates and security before installing guest packages")
if self.failure == "packages":
return CommandResult(args, 100, stdout="Reading package lists...\n", stderr="E: unmet guest dependencies\n")
status = "\n\n".join(
f"Package: {package}\nStatus: install ok installed\nVersion: 1.0"
for package in ("apt", "sudo", "openssh-server", "systemd-sysv")
) + "\n"
files = {
"etc/passwd": "root:x:0:0::/root:/bin/bash\n",
"etc/shadow": "root:*:1:0:99999:7:::\n", "etc/group": "root:x:0:\n",
"etc/sudoers": "root ALL=(ALL:ALL) ALL\n",
"etc/ssh/sshd_config": "UsePAM yes\n",
"etc/ssh/ssh_host_rsa_key": "template host key",
"usr/bin/sudo": "sudo binary", "usr/bin/apt-get": "apt binary",
"usr/bin/sudo": "sudo binary",
"usr/sbin/sshd": "sshd binary", "usr/lib/systemd/systemd": "init binary",
"var/lib/dpkg/status": status,
"var/lib/dpkg/info/apt.list": "/usr/bin/apt-get\n",
"var/lib/dpkg/info/sudo.list": "/usr/bin/sudo\n",
}
if self.failure == "metadata":
files["var/lib/dpkg/status"] = ""
for relative, contents in files.items():
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(contents, encoding="utf-8")
self.write_files(root, files)
(root / "usr/bin/sudo").chmod(0o4755)
(root / "etc/resolv.conf").symlink_to("../run/systemd/resolve/stub-resolv.conf")
return CommandResult(args, 0)
return CommandResult(args, 0, stdout="Configured guest packages.\n")
if args[0] == "systemctl":
return CommandResult(args, 0)
return self.runner.run(command, **options)
@staticmethod
def write_files(root: Path, files: dict[str, str]) -> None:
for relative, contents in files.items():
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(contents, encoding="utf-8")
class UbuntuAptStageTests(unittest.TestCase):
def test_apt_script_preserves_arguments_and_stops_after_an_update_failure(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
root = base / "guest with spaces"
policy = root / "usr/sbin/policy-rc.d"
policy.parent.mkdir(parents=True)
policy.write_text("#!/bin/sh\nexit 101\n", encoding="utf-8")
tools = base / "tools"
tools.mkdir()
for name in ("mount", "chroot"):
tool = tools / name
tool.write_text(
f"#!{sys.executable}\n"
"import json, os, sys\nfrom pathlib import Path\n"
"name = Path(sys.argv[0]).name\n"
"print(json.dumps([name, *sys.argv[1:]]))\n"
"if name == 'chroot':\n"
" assert (Path(sys.argv[1]) / 'usr/sbin/policy-rc.d').is_file()\n"
" if os.environ.get('FAIL_UPDATE') == '1' and sys.argv[-1] == 'update':\n"
" sys.exit(100)\n",
encoding="utf-8",
)
tool.chmod(0o755)
runner = CommandRunner(emit=None)
for fail in (False, True):
with self.subTest(fail=fail):
result = runner.run((
"env", f"PATH={tools}", f"FAIL_UPDATE={int(fail)}",
"/bin/sh", "-eu", "-c", _GUEST_APT_SCRIPT,
"uvm-guest-packages", root, "sudo", "openssh-server",
), check=False, capture=True)
events = [json.loads(line) for line in result.stdout.splitlines()]
mounts = [event for event in events if event[0] == "mount"]
self.assertEqual([event[-1] for event in mounts], [
str(root / "dev"), str(root / "dev"), str(root / "proc"), str(root / "sys"),
])
calls = [event for event in events if event[0] == "chroot"]
self.assertEqual(result.returncode, 100 if fail else 0)
self.assertEqual(len(calls), 1 if fail else 3)
self.assertEqual(calls[0][-3:], ["-o", "APT::Update::Error-Mode=any", "update"])
for call in calls:
self.assertEqual(call[1:6], [str(root), "/usr/bin/env", "DEBIAN_FRONTEND=noninteractive", "LC_ALL=C", "apt-get"])
if not fail:
self.assertEqual(calls[1][-1], "dist-upgrade")
self.assertEqual(calls[2][-3:], ["install", "sudo", "openssh-server"])
@unittest.skipUnless(all(shutil.which(tool) for tool in ("mkfs.ext4", "e2fsck", "debugfs")),
"e2fsprogs is required for offline Ubuntu image checks")
@@ -233,11 +308,23 @@ class UbuntuRootfsBuildTests(unittest.TestCase):
self.assertIn(f"--arch={deb_arch}", bootstrap)
self.assertEqual(bootstrap[-1], mirror)
self.assertIn("--force-check-gpg", bootstrap)
self.assertIn("--extra-suites=jammy-updates,jammy-security", bootstrap)
self.assertIn("--components=main", bootstrap)
self.assertIn("--include=ca-certificates,ubuntu-keyring", bootstrap)
self.assertFalse(any(part.startswith("--extra-suites") for part in bootstrap))
self.assertIn("--mount-proc", bootstrap)
self.assertIn("--kill-child", bootstrap)
packages = next(call for call in runner.calls if "uvm-guest-packages" in call)
self.assertLess(runner.calls.index(bootstrap), runner.calls.index(packages))
self.assertIn("sudo", packages)
self.assertIn("openssh-server", packages)
self.assertIn("--mount-proc", packages)
self.assertIn("--kill-child", packages)
self.assertEqual(settings.rootfs_image.stat().st_mode & 0o777, 0o600)
self.assertEqual(list(settings.images_dir.iterdir()), [settings.rootfs_image])
log = settings.images_dir / "ubuntu-build.log"
self.assertEqual(set(settings.images_dir.iterdir()), {settings.rootfs_image, log})
self.assertEqual(log.stat().st_mode & 0o777, 0o600)
self.assertIn("Base system installed successfully.", log.read_text())
self.assertIn("Configured guest packages.", log.read_text())
def guest(command):
return runner.runner.run(("debugfs", "-R", command, settings.rootfs_image), capture=True)
@@ -252,16 +339,29 @@ class UbuntuRootfsBuildTests(unittest.TestCase):
for path in ("/usr/sbin/policy-rc.d", "/etc/ssh/ssh_host_rsa_key"):
self.assertIn("File not found", guest(f"stat {path}").stderr)
def test_failed_bootstrap_or_missing_package_metadata_preserves_the_old_template(self) -> None:
for failure in ("bootstrap", "metadata"):
def test_failed_build_preserves_the_old_template_and_retains_diagnostics(self) -> None:
for failure in ("bootstrap", "packages", "metadata"):
with self.subTest(failure=failure), tempfile.TemporaryDirectory() as temporary:
settings = Settings(base=Path(temporary))
settings.images_dir.mkdir()
settings.rootfs_image.write_bytes(b"old template")
runner = UbuntuBuildRunner(failure=failure)
installer = Installer(settings, runner, StateStore(settings), emit=None) # type: ignore[arg-type]
with self.assertRaises(UvmError):
log = settings.images_dir / "ubuntu-build.log"
with self.assertRaisesRegex(UvmError, "details saved to .*ubuntu-build.log"):
installer._build_ubuntu_rootfs(settings.rootfs_image)
self.assertEqual(settings.rootfs_image.read_bytes(), b"old template")
self.assertEqual(list(settings.images_dir.iterdir()), [settings.rootfs_image])
self.assertEqual(set(settings.images_dir.iterdir()), {settings.rootfs_image, log})
self.assertEqual(log.stat().st_mode & 0o777, 0o600)
details = log.read_text()
if failure == "bootstrap":
self.assertIn("--install needs at least one package archive", details)
self.assertIn("Failure while unpacking", details)
self.assertFalse(any("uvm-guest-packages" in call for call in runner.calls))
elif failure == "packages":
self.assertIn("Reading package lists...", details)
self.assertIn("E: unmet guest dependencies", details)
self.assertLess(details.index("Base system installed successfully."), details.index("E: unmet guest dependencies"))
else:
self.assertIn("incomplete apt/dpkg package metadata", details)
self.assertFalse(any(call[0] == "mkfs.ext4" for call in runner.calls))