#!/usr/bin/env python3 """Exercise the guest build/deploy transaction without a VM, root, or network.""" import os import fcntl from pathlib import Path import shlex import signal import socket import subprocess import tarfile import tempfile import unittest import time GUEST_SCRIPT = Path(__file__).parent / "micromanager/dev-macos-guest.sh" SESSION_SCRIPT = Path(__file__).parent / "micromanager/dev-session.sh" STUB = r'''#!/bin/bash set -eu name=${0##*/} block_command() { exec python3 - <<'PY' import fcntl, os, subprocess from pathlib import Path try: fcntl.fcntl(9, fcntl.F_GETFD) inherited_lock = True except OSError: inherited_lock = False # The sleep stub exits at once; only the real command holds the cancellation point. child = subprocess.Popen(["/bin/sleep", "60"]) Path(os.environ["TEST_ROOT"], "blocked").write_text( f"{os.getpid()} {child.pid} {os.getpgrp()} {int(inherited_lock)}") child.wait() PY } case "$name" in chmod|rm|install) if [ "$1" = --version ]; then exit 0; fi if [ "$name" = rm ] && [ "${CLEANUP_FAIL:-0}" = 1 ] \ && [[ "${2:-}" = "$TEST_ROOT"/cache/rollback.* ]]; then exit 49; fi if [ "$name" = install ]; then exec /usr/bin/install "$@"; fi exec "/bin/$name" "$@" ;; findmnt) path=${!#} selected= flags= while read -r device mount type options rest; do if [[ "$mount" = / || "$path" = "$mount" || "$path" = "$mount/"* ]]; then if [ "${#mount}" -gt "${#selected}" ]; then selected=$mount; flags=$options; fi fi done < "$TEST_ROOT/mounts" echo "$flags" ;; uname) echo "${GUEST_ARCH:-aarch64}" ;; cc) [ "${MISSING_KERNEL_HEADERS:-0}" = 0 ] || [ -f "$TEST_ROOT/kernel-deps-ready" ] ;; apt-get) echo "$*" >>"$TEST_ROOT/dependency-installs" [ "${DEPENDENCY_INSTALL_FAIL:-0}" = 0 ] || exit 48 touch "$TEST_ROOT/kernel-deps-ready" ;; make|flex|bison|bc|perl|musl-gcc|pkg-config|cmake|journalctl|sleep) exit 0 ;; objcopy) if [ "$1" = --version ]; then exit 0; fi echo "$*" >>"$TEST_ROOT/objcopy-events" [ "$1" = --compress-debug-sections=zlib ] && [ "$#" = 3 ] exec /bin/cp "$2" "$3" ;; df) echo 'Filesystem 1024-blocks Used Available Capacity Mounted on' echo "/dev/os 2900000 0 ${OS_FREE_KIB:-1048576} 0% /" ;; flock) [ "${LOCK_FAIL:-0}" = 0 ] ;; curl) [ "${HEALTH_FAIL:-0}" = 0 ] ;; rustup) echo "$*" >>"$TEST_ROOT/rustup-events" if [ "$1" = toolchain ]; then [ "${TOOLCHAIN_INSTALL_FAIL:-0}" = 0 ]; exit; fi [ "$1" = run ] || exit 0 if [ "$3" = rustc ]; then [ "${TOOLCHAIN_MISSING:-0}" = 0 ]; exit; fi [ "${BUILD_FAIL:-0}" = 0 ] || exit 42 if [ "${BLOCK_BUILD:-0}" = 1 ]; then block_command; fi test "$(cat "$TEST_ROOT/cache/source/firecrab-api/src/main.rs")" = 'edited local source' fragment="$TEST_ROOT/cache/source/firecrab-api/src/oci/guest/install-packages.sh" if [ -f "$fragment" ]; then cp "$fragment" "$TEST_ROOT/compiled-fragment"; fi profile=debug for arg in "$@"; do [ "$arg" != --release ] || profile=release; done target= while [ "$#" -gt 0 ]; do if [ "$1" = --target ]; then target=$2; break; fi shift done [ "$target" = "${GUEST_ARCH:-aarch64}-unknown-linux-gnu" ] mkdir -p "$CARGO_TARGET_DIR/$target/$profile" for unit in firecrab-api firecrab-helper firecrab; do printf '#!/bin/bash\necho built-from-source\n' >"$CARGO_TARGET_DIR/$target/$profile/$unit" done if [ "${DISK_FAIL_AFTER_BUILD:-0}" = 1 ]; then printf '/dev/os / ext4 ro,relatime 0 0\n' >"$TEST_ROOT/mounts" fi ;; systemctl) if [ "$*" = 'show --property=LoadState --value firecrab-helper.service' ]; then if [ "${LEGACY_HELPER:-0}" = 1 ]; then echo not-found; else echo loaded; fi exit 0 fi echo "$*" >>"$TEST_ROOT/events" if [ "$1" = list-units ]; then [ "${VM_LIST_FAIL:-0}" = 0 ] || exit 46 if [ -f "$TEST_ROOT/vm-units" ]; then cat "$TEST_ROOT/vm-units"; fi exit 0 fi if [ "$1" = stop ] && [[ "$2" = firecrab-vm-* ]]; then [ "${VM_STOP_FAIL:-0}" = 0 ] || exit 47 if [ -f "$TEST_ROOT/vm-runtime-pid" ]; then kill "$(cat "$TEST_ROOT/vm-runtime-pid")" # Wait for the exclusive runtime resource to be released before starting # the installed API, just as synchronous systemctl stop waits for the unit. python3 - <<'PYWAIT' import fcntl, os with open(os.path.join(os.environ["TEST_ROOT"], "tap-lock"), "w") as resource: fcntl.flock(resource, fcntl.LOCK_EX) PYWAIT fi : >"$TEST_ROOT/vm-units" fi if [ "$1" = stop ] && [ "$2" = firecrab-api ] \ && [ "${BLOCK_ACTIVATE:-0}" = 1 ] && [ ! -f "$TEST_ROOT/blocked" ]; then block_command; fi if [ "$*" = 'stop firecrab-api' ] && [ "${STOP_FAIL:-0}" = 1 ]; then exit 44; fi if [ "$1" = start ] && [ "$2" != firecrab-api ] && [ "${HELPER_START_FAIL:-0}" = 1 ]; then exit 45; fi if [ "$*" = 'start firecrab-api' ] && [ "${START_FAIL:-0}" = 1 ] \ && [ -f "$TEST_ROOT/units/firecrab-api.service.d/90-firecrab-dev.conf" ]; then exit 43; fi ;; *) echo "unexpected stub: $name" >&2; exit 1 ;; esac ''' class GuestDevelopmentTest(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix="firecrab-dev-test-", dir="/private/tmp" if Path("/private/tmp").exists() else None) self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name).resolve() self.cache = self.root / "cache" self.binaries = self.root / "binaries" self.units = self.root / "units" self.commands = self.root / "commands" self.commands.mkdir() self.mounts = self.root / "mounts" self.mounts.write_text("/dev/os / ext4 rw,relatime 0 0\n") stub = self.commands / "stub" stub.write_text(STUB) stub.chmod(0o755) for name in ["findmnt", "chmod", "rm", "install", "uname", "cc", "make", "flex", "bison", "bc", "perl", "apt-get", "musl-gcc", "pkg-config", "cmake", "objcopy", "df", "journalctl", "sleep", "flock", "curl", "rustup", "systemctl"]: (self.commands / name).symlink_to(stub) # Linux mv -T replaces a symlink rather than following its directory. mv = self.commands / "mv" mv.write_text("#!/usr/bin/env python3\nimport os, sys\nos.replace(sys.argv[-2], sys.argv[-1])\n") mv.chmod(0o755) rustup = self.cache / "cargo/bin/rustup" rustup.parent.mkdir(parents=True) rustup.symlink_to(stub) self.sock = socket.socket(socket.AF_UNIX) self.sock.bind(str(self.root / "helper.sock")) self.addCleanup(self.sock.close) script = GUEST_SCRIPT.read_text() for original, replacement in [ ("root=/var/lib/firecrab/dev", "root=" + shlex.quote(str(self.cache))), ("bin_root=/usr/local/lib/firecrab-dev", "bin_root=" + shlex.quote(str(self.binaries))), ("unit_root=/etc/systemd/system", "unit_root=" + shlex.quote(str(self.units))), ("os_root=/", "os_root=" + shlex.quote(str(self.root))), ("/run/firecrab/net-helper.sock", str(self.root / "helper.sock")), ]: script = script.replace(original, replacement) self.script = self.root / "guest.sh" self.script.write_text(script) source = self.root / "local/firecrab-api/src/main.rs" source.parent.mkdir(parents=True) source.write_text("edited local source") tools = self.root / "local/scripts/firecracker-menual" tools.mkdir(parents=True) for tool in ["extract-vmlinux", "extract-arm64-image"]: (tools / tool).write_text("#!/bin/sh\necho runtime-tool\n") ingress_builder = tools.parent / "build-ingress.sh" ingress_builder.write_text('#!/bin/bash\nset -eu\nmkdir -p "$1/ingress"\nprintf "#!/bin/sh\\nexit 0\\n" >"$1/firecrab-ingress"\nchmod +x "$1/firecrab-ingress"\nprintf "fixture" >"$1/ingress/LICENSE.nginx"\n') guest_builder = tools.parent / "build-guest-console.sh" guest_builder.write_text('#!/bin/bash\nset -eu\nmkdir -p "$1/guest-console"\nprintf "#!/bin/sh\\nexit 0\\n" >"$1/firecrab-guest-console"\nchmod +x "$1/firecrab-guest-console"\nprintf "musl fixture" >"$1/guest-console/LICENSE.musl"\n') self.archive = self.cache / "incoming-test.tar" with tarfile.open(self.archive, "w") as archive: archive.add(source, arcname="firecrab-api/src/main.rs") archive.add(tools, arcname="scripts/firecracker-menual") archive.add(ingress_builder, arcname="scripts/build-ingress.sh") archive.add(guest_builder, arcname="scripts/build-guest-console.sh") def test_existing_guest_installs_missing_kernel_headers_before_build(self): result = self.run_guest(MISSING_KERNEL_HEADERS=1) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) installs = (self.root / 'dependency-installs').read_text() for package in ('build-essential', 'flex', 'bison', 'bc', 'perl', 'xz-utils', 'libelf-dev', 'libssl-dev'): self.assertIn(package, installs) def test_dependency_install_failure_keeps_the_running_api(self): result = self.run_guest(MISSING_KERNEL_HEADERS=1, DEPENDENCY_INSTALL_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.root / 'events').exists()) self.assertFalse((self.binaries / 'bin').exists()) def test_guest_console_helper_and_attribution_are_staged_next_to_the_api(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stdout + result.stderr) current = self.binaries / "bin" self.assertTrue((current / "firecrab-guest-console").is_file()) self.assertEqual((current / "guest-console/LICENSE.musl").read_text(), "musl fixture") def run_guest(self, profile="debug", shell="bash", **flags): environment = os.environ.copy() environment.update(TEST_ROOT=str(self.root), FIRECRAB_DEV_ISOLATED="0", PATH=str(self.commands) + ":" + environment["PATH"]) environment.update({key: str(value) for key, value in flags.items()}) args = [shell, str(self.script), profile] if profile not in ["start", "stop", "restart", "clean", "check"]: args.extend(["1.97.1", self.archive.name]) return subprocess.run(args, env=environment, text=True, capture_output=True, timeout=10) def override(self, unit="firecrab-api"): return self.units / (unit + ".service.d/90-firecrab-dev.conf") def start_session(self, **flags): # Use a real inherited-file-description lock on Linux and macOS. flock = self.commands / "flock" flock.unlink() flock.write_text("#!/usr/bin/env python3\nimport fcntl\nfcntl.flock(9, fcntl.LOCK_EX | fcntl.LOCK_NB)\n") flock.chmod(0o755) environment = os.environ.copy() environment.update(TEST_ROOT=str(self.root), FIRECRAB_DEV_ISOLATED="0", PATH=str(self.commands) + ":" + environment["PATH"]) environment.update({key: str(value) for key, value in flags.items()}) output = open(self.root / "session.stdout", "w+") errors = open(self.root / "session.stderr", "w+") self.addCleanup(output.close) self.addCleanup(errors.close) process = subprocess.Popen( ["bash", "-c", SESSION_SCRIPT.read_text(), "--", "debug", "1.97.1", self.archive.name], stdin=subprocess.PIPE, stdout=output, stderr=errors, env=environment, start_new_session=True) def cleanup(): if not process.stdin.closed: process.stdin.close() try: process.wait(timeout=10) except subprocess.TimeoutExpired: marker = self.root / "blocked" if marker.exists(): group = int(marker.read_text().split()[2]) try: os.killpg(group, signal.SIGKILL) except ProcessLookupError: pass process.kill() process.wait(timeout=5) self.addCleanup(cleanup) process.stdin.write(self.script.read_bytes() + b"\0") process.stdin.flush() return process def wait_for_blocked_command(self, process): deadline = time.monotonic() + 5 while time.monotonic() < deadline: marker = self.root / "blocked" if marker.exists() and len(marker.read_text().split()) == 4: return [int(value) for value in marker.read_text().split()] self.assertIsNone(process.poll(), (self.root / "session.stderr").read_text()) time.sleep(0.01) self.fail("development command did not reach the cancellation point") def test_caller_disconnect_stops_compilers_releases_lock_and_allows_retry(self): archive = self.archive.read_bytes() unrelated = subprocess.Popen(["sleep", "60"], start_new_session=True) self.addCleanup(unrelated.wait) self.addCleanup(unrelated.terminate) process = self.start_session(BLOCK_BUILD=1) compiler, descendant, group, inherited_lock = self.wait_for_blocked_command(process) self.assertEqual(inherited_lock, 0, "compiler inherited deployment lock") self.assertNotEqual(group, os.getpgrp()) with open(self.cache / "deploy.lock", "w") as lock: with self.assertRaises(BlockingIOError): fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) process.stdin.close() self.assertEqual(process.wait(timeout=5), 130) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) self.assertIsNone(unrelated.poll()) for pid in [compiler, descendant]: deadline = time.monotonic() + 3 while time.monotonic() < deadline: try: os.kill(pid, 0) except ProcessLookupError: break time.sleep(0.01) else: self.fail(f"cancelled process {pid} is still alive") self.archive.write_bytes(archive) result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) def test_caller_disconnect_during_activation_restores_previous_services(self): previous = self.binaries / "previous" previous.mkdir(parents=True) (self.binaries / "bin").symlink_to(previous) self.override().parent.mkdir(parents=True) self.override().write_text("previous override\n") process = self.start_session(BLOCK_ACTIVATE=1) self.wait_for_blocked_command(process) process.stdin.close() self.assertEqual(process.wait(timeout=5), 130) self.assertEqual((self.binaries / "bin").resolve(), previous.resolve()) self.assertEqual(self.override().read_text(), "previous override\n") self.assertTrue((self.root / "events").read_text().endswith("start firecrab-api\n")) def test_live_caller_can_finish_without_stdin_eof(self): process = self.start_session() self.assertEqual(process.wait(timeout=10), 0, (self.root / "session.stderr").read_text()) self.assertFalse(process.stdin.closed) self.assertNotIn("[CANCEL]", (self.root / "session.stderr").read_text()) def test_transport_signals_cancel_only_the_active_deployment(self): archive = self.archive.read_bytes() for shutdown in [signal.SIGINT, signal.SIGTERM, signal.SIGHUP]: with self.subTest(signal=shutdown): self.archive.write_bytes(archive) (self.root / "blocked").unlink(missing_ok=True) process = self.start_session(BLOCK_BUILD=1) self.wait_for_blocked_command(process) process.send_signal(shutdown) self.assertEqual(process.wait(timeout=5), 130) self.assertFalse((self.root / "events").exists()) with open(self.cache / "deploy.lock", "w") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) def test_read_only_os_rejects_start_restart_and_build_even_with_healthy_api(self): self.mounts.write_text("/dev/os / ext4 ro,relatime 0 0\n") for action in ["check", "start", "restart", "clean", "debug", "restore"]: with self.subTest(action=action): result = self.run_guest(action) self.assertEqual(result.returncode, 74, result.stderr) self.assertIn("is not writable", result.stderr) self.assertNotIn("[PASS]", result.stdout) self.assertFalse((self.root / "events").exists()) self.assertFalse((self.root / "rustup-events").exists()) def test_read_only_data_mount_is_checked_separately_from_os(self): self.mounts.write_text(f"/dev/os / ext4 rw 0 0\n/dev/data {self.cache} ext4 ro 0 0\n") self.cache.mkdir(exist_ok=True) result = self.run_guest() self.assertEqual(result.returncode, 74, result.stderr) self.assertIn(str(self.cache), result.stderr) self.assertFalse((self.root / "events").exists()) def test_read_only_overmount_takes_precedence_over_the_original_mount(self): self.mounts.write_text("/dev/os / ext4 ro 0 0\n/dev/os / ext4 rw 0 0\n") result = self.run_guest("start") self.assertEqual(result.returncode, 74, result.stderr) self.assertNotIn("[PASS]", result.stdout) def test_stop_remains_available_on_read_only_os(self): self.mounts.write_text("/dev/os / ext4 ro 0 0\n") result = self.run_guest("stop") self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("[PASS]", result.stdout) def test_disk_failure_during_build_preserves_services_and_recovery_files(self): result = self.run_guest(DISK_FAIL_AFTER_BUILD=1) self.assertEqual(result.returncode, 74, result.stderr) self.assertIn("firecrab service repair", result.stderr) self.assertIn("[PRESERVE]", result.stderr) self.assertTrue(self.archive.exists()) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) def test_private_disk_failure_preserves_state_without_primary_repair_advice(self): result = self.run_guest(DISK_FAIL_AFTER_BUILD=1, FIRECRAB_DEV_ISOLATED=1) self.assertEqual(result.returncode, 74, result.stderr) self.assertIn("selected private guest OS filesystem is unhealthy", result.stderr) self.assertIn("[PRESERVE]", result.stderr) self.assertNotIn("firecrab service", result.stdout + result.stderr) self.assertTrue(self.archive.exists()) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) def test_low_os_disk_space_fails_before_staging_or_stopping_services(self): result = self.run_guest(OS_FREE_KIB=1024) self.assertNotEqual(result.returncode, 0) self.assertIn("[FAIL] guest OS disk has 1 MiB free", result.stderr) self.assertIn("[HINT]", result.stderr) self.assertIn("firecrab service shell --", result.stderr) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) self.assertEqual(list(self.binaries.glob("build.*")), []) def test_private_low_disk_space_does_not_suggest_the_primary_shell(self): result = self.run_guest(OS_FREE_KIB=1024, FIRECRAB_DEV_ISOLATED=1) self.assertNotEqual(result.returncode, 0) self.assertIn("[FAIL] guest OS disk has 1 MiB free", result.stderr) self.assertIn("disk usage and logs in the selected private guest", result.stderr) self.assertNotIn("firecrab service", result.stdout + result.stderr) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) self.assertEqual(list(self.binaries.glob("build.*")), []) def test_cleanup_failure_advice_keeps_the_selected_guest_scope(self): archive = self.archive.read_bytes() for isolated in [0, 1]: with self.subTest(isolated=isolated): self.archive.write_bytes(archive) result = self.run_guest(CLEANUP_FAIL=1, FIRECRAB_DEV_ISOLATED=isolated) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("deployment cleanup failed", result.stderr) if isolated: self.assertIn("selected private guest", result.stderr) self.assertNotIn("firecrab service", result.stdout + result.stderr) else: self.assertIn("firecrab service debug --logs", result.stderr) def test_staging_compresses_debug_sections_and_removes_interrupted_builds(self): running = self.binaries / "build.running" running.mkdir(parents=True) (self.binaries / "bin").symlink_to(running) interrupted = self.binaries / "build.interrupted" interrupted.mkdir() result = self.run_guest(HEALTH_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertFalse(interrupted.exists()) self.assertEqual((self.binaries / "bin").resolve(), running.resolve()) calls = (self.root / "objcopy-events").read_text().splitlines() self.assertEqual(len(calls), 3) self.assertTrue(all(call.startswith("--compress-debug-sections=zlib ") for call in calls)) def test_unreadable_system_executable_is_reported_as_os_failure(self): chmod = self.commands / "chmod" chmod.unlink() chmod.write_text("#!/bin/sh\necho 'Input/output error' >&2\nexit 126\n") chmod.chmod(0o755) result = self.run_guest("start") self.assertEqual(result.returncode, 74, result.stderr) self.assertIn("OS cannot execute chmod", result.stderr) self.assertIn("firecrab service debug --logs", result.stderr) self.assertFalse((self.root / "events").exists()) def test_private_unreadable_executable_does_not_suggest_primary_debug(self): chmod = self.commands / "chmod" chmod.unlink() chmod.write_text("#!/bin/sh\necho 'Input/output error' >&2\nexit 126\n") chmod.chmod(0o755) result = self.run_guest("start", FIRECRAB_DEV_ISOLATED=1) self.assertEqual(result.returncode, 74, result.stderr) self.assertIn("selected private guest OS cannot execute chmod", result.stderr) self.assertNotIn("firecrab service", result.stdout + result.stderr) self.assertFalse((self.root / "events").exists()) def test_legacy_service_uses_the_new_helper_executable_and_restore_removes_its_override(self): result = self.run_guest(LEGACY_HELPER=1) self.assertEqual(result.returncode, 0, result.stderr + result.stdout) legacy = self.override("firecrab-net-helper") self.assertIn("/bin/firecrab-helper", legacy.read_text()) self.assertFalse(self.override("firecrab-helper").exists()) restored = self.run_guest(profile="restore", LEGACY_HELPER=1) self.assertEqual(restored.returncode, 0, restored.stderr) self.assertFalse(legacy.exists()) def test_alias_name_development_override_is_removed_when_deploying_or_restoring(self): legacy = self.override("firecrab-net-helper") legacy.parent.mkdir(parents=True) legacy.write_text("old alias override") result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(legacy.exists()) legacy.write_text("old alias override") restored = self.run_guest(profile="restore") self.assertEqual(restored.returncode, 0, restored.stderr) self.assertFalse(legacy.exists()) self.assertFalse(self.override("firecrab-helper").exists()) def test_cached_toolchain_does_not_require_network_refresh(self): result = self.run_guest(TOOLCHAIN_INSTALL_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) calls = (self.root / "rustup-events").read_text() self.assertIn("run 1.97.1 rustc --version", calls) self.assertNotIn("toolchain install", calls) def test_missing_toolchain_is_installed_before_build(self): result = self.run_guest(TOOLCHAIN_MISSING=1) self.assertEqual(result.returncode, 0, result.stderr) calls = (self.root / "rustup-events").read_text() self.assertLess(calls.index("toolchain install"), calls.index("cargo build")) def test_toolchain_download_failure_preserves_running_services(self): result = self.run_guest(TOOLCHAIN_MISSING=1, TOOLCHAIN_INSTALL_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) def test_debug_and_release_deploy_edited_source(self): for profile in ["debug", "release"]: with self.subTest(profile=profile): # Each upload is consumed by its build. if not self.archive.exists(): source = self.root / "local/firecrab-api/src/main.rs" with tarfile.open(self.archive, "w") as archive: archive.add(source, arcname="firecrab-api/src/main.rs") archive.add(self.root / "local/scripts/firecracker-menual", arcname="scripts/firecracker-menual") result = self.run_guest(profile) self.assertEqual(result.returncode, 0, result.stderr + result.stdout) self.assertIn("ExecStart=" + str(self.binaries) + "/bin/firecrab-api", self.override().read_text()) self.assertIn("built-from-source", (self.binaries / "bin/firecrab-helper").read_text()) self.assertIn("runtime-tool", (self.binaries / "bin/extract-arm64-image").read_text()) if profile == "debug": self.assertTrue((self.binaries / "bin/firecrab-ingress").is_file()) self.assertIn("Environment=FIRECRAB_INGRESS_BINARY=" + str(self.binaries) + "/bin/firecrab-ingress", self.override("firecrab-helper").read_text()) self.assertFalse(self.archive.exists()) def test_restore_keeps_the_engine_of_a_configured_ingress(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) ingress = self.units / "firecrab-ingress.service" ingress.write_text(f"[Service]\nExecStart={self.binaries}/bin/firecrab-ingress -p /run/firecrab-ingress/\n") restored = self.run_guest(profile="restore") self.assertEqual(restored.returncode, 0, restored.stderr) self.assertFalse(self.override().exists()) self.assertTrue((self.binaries / "bin/firecrab-ingress").is_file()) def test_clean_preserves_configured_ingress_binaries_while_removing_build_caches(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) pointer = self.binaries / "bin" previous = pointer.resolve() engine = (pointer / "firecrab-ingress").read_bytes() resource = (pointer / "ingress/LICENSE.nginx").read_bytes() ingress = self.units / "firecrab-ingress.service" unit = f"[Service]\nExecStart={pointer}/firecrab-ingress -p /run/firecrab-ingress/\n" ingress.write_text(unit) lock_inode = (self.cache / "deploy.lock").stat().st_ino for _ in range(2): events = self.root / "events" events.unlink(missing_ok=True) result = self.run_guest("clean", BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("independent ingress preserved", result.stdout) self.assertFalse(self.override().exists()) self.assertFalse(self.override("firecrab-helper").exists()) self.assertEqual(pointer.resolve(), previous) self.assertEqual((pointer / "firecrab-ingress").read_bytes(), engine) self.assertEqual((pointer / "ingress/LICENSE.nginx").read_bytes(), resource) self.assertEqual(ingress.read_text(), unit) self.assertNotIn("firecrab-ingress", events.read_text()) self.assertEqual(sorted(path.name for path in self.cache.iterdir()), ["deploy.lock"]) self.assertEqual((self.cache / "deploy.lock").stat().st_ino, lock_inode) def test_debug_deploy_works_with_system_bash(self): result = self.run_guest(shell="/bin/bash") self.assertEqual(result.returncode, 0, result.stderr + result.stdout) self.assertTrue(self.override().is_file()) self.assertIn("built-from-source", (self.binaries / "bin/firecrab-api").read_text()) def test_build_failure_does_not_touch_services_or_previous_binaries(self): result = self.run_guest(BUILD_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertFalse((self.root / "events").exists()) self.assertFalse(self.override().exists()) self.assertFalse((self.binaries / "bin").exists()) def test_windows_x86_64_checkout_deploys_debug_and_release_with_crlf_tools(self): for profile in ["debug", "release"]: with self.subTest(profile=profile): tools = self.root / "local/scripts/firecracker-menual" named_tool = tools / "runtime tool 한글.sh" named_tool.write_text("runtime tool") named_tool.chmod(0o751) for tool in tools.iterdir(): tool.write_bytes(b"#!/bin/sh\r\necho runtime-tool\r\n") fragment = self.root / "local/firecrab-api/src/oci/guest/install-packages.sh" fragment.parent.mkdir(parents=True, exist_ok=True) fragment.write_bytes(b"echo embedded-guest-script\r\nprintf 'keep\rinside'\r\n") with tarfile.open(self.archive, "w") as archive: archive.add(self.root / "local/firecrab-api/src/main.rs", arcname="firecrab-api/src/main.rs") archive.add(fragment, arcname="firecrab-api/src/oci/guest/install-packages.sh") archive.add(tools, arcname="scripts/firecracker-menual") result = self.run_guest(profile, GUEST_ARCH="x86_64") self.assertEqual(result.returncode, 0, result.stderr + result.stdout) self.assertIn("built-from-source", (self.binaries / "bin/firecrab-api").read_text()) self.assertNotIn(b"\r", (self.binaries / "bin/extract-vmlinux").read_bytes()) self.assertEqual((self.root / "compiled-fragment").read_bytes(), b"echo embedded-guest-script\nprintf 'keep\rinside'\n") staged_tool = self.cache / "source/scripts/firecracker-menual" / named_tool.name self.assertEqual(staged_tool.read_bytes(), b"#!/bin/sh\necho runtime-tool\n") self.assertEqual(staged_tool.stat().st_mode & 0o777, 0o751) def test_unsupported_architecture_fails_before_mutating_services(self): result = self.run_guest(GUEST_ARCH="riscv64") self.assertNotEqual(result.returncode, 0) self.assertIn("Linux ARM64 or x86_64 guest required", result.stderr) self.assertFalse((self.root / "events").exists()) def test_failed_first_deployment_restores_release_services(self): result = self.run_guest(START_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertIn("[ROLLBACK]", result.stderr) self.assertFalse(self.override().exists()) self.assertFalse((self.binaries / "bin").exists()) self.assertTrue((self.root / "events").read_text().endswith("start firecrab-api\n")) def test_failed_update_restores_previous_development_executables(self): previous = self.binaries / "previous" previous.mkdir(parents=True) (self.binaries / "bin").symlink_to(previous) self.override().parent.mkdir(parents=True) self.override().write_text("previous override\n") result = self.run_guest(HEALTH_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertEqual((self.binaries / "bin").resolve(), previous.resolve()) self.assertEqual(self.override().read_text(), "previous override\n") def test_restore_does_not_build_and_preserves_other_overrides(self): self.override().parent.mkdir(parents=True) self.override().write_text("development override\n") custom = self.override().parent / "custom.conf" custom.write_text("custom settings\n") result = self.run_guest("restore", BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(self.override().exists()) self.assertEqual(custom.read_text(), "custom settings\n") def test_restore_after_source_deployment_removes_development_pointer(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) result = self.run_guest("restore", BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) self.assertFalse(self.override().exists()) self.assertFalse((self.binaries / "bin").exists()) def test_concurrent_build_is_rejected_before_mutating_services(self): result = self.run_guest(LOCK_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertIn("in progress", result.stderr) self.assertFalse((self.root / "events").exists()) def test_clean_restores_release_and_removes_only_development_files(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) (self.root / "rustup-events").unlink() custom = self.override().parent / "custom.conf" custom.write_text("custom settings\n") data = self.root / "persistent-data" data.mkdir() (data / "disk.raw").write_text("VM data\n") (self.cache / ".hidden-cache").write_text("cached file") (self.cache / "external").symlink_to(data, target_is_directory=True) (self.binaries / "external").symlink_to(data, target_is_directory=True) lock_inode = (self.cache / "deploy.lock").stat().st_ino result = self.run_guest("clean", BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("development files cleaned", result.stdout) self.assertFalse(self.override().exists()) self.assertFalse(self.binaries.exists()) self.assertEqual(sorted(path.name for path in self.cache.iterdir()), ["deploy.lock"]) self.assertEqual((self.cache / "deploy.lock").stat().st_ino, lock_inode) self.assertFalse((self.root / "rustup-events").exists()) self.assertEqual(custom.read_text(), "custom settings\n") self.assertEqual((data / "disk.raw").read_text(), "VM data\n") self.assertEqual((self.root / "local/firecrab-api/src/main.rs").read_text(), "edited local source") def test_clean_without_a_deployment_is_repeatable_and_needs_no_checkout(self): for _ in range(2): result = self.run_guest("clean", BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) self.assertIn("[PASS]", result.stdout) self.assertFalse(self.binaries.exists()) self.assertEqual(sorted(path.name for path in self.cache.iterdir()), ["deploy.lock"]) self.assertFalse((self.root / "rustup-events").exists()) def test_failed_clean_restoration_preserves_development_binaries_and_cache(self): result = self.run_guest() self.assertEqual(result.returncode, 0, result.stderr) previous = (self.binaries / "bin").resolve() override = self.override().read_text() result = self.run_guest("clean", HEALTH_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertIn("[ROLLBACK]", result.stderr) self.assertNotIn("development files cleaned", result.stdout) self.assertEqual((self.binaries / "bin").resolve(), previous) self.assertEqual(self.override().read_text(), override) self.assertTrue((self.cache / "source/firecrab-api/src/main.rs").exists()) self.assertTrue((self.cache / "target").is_dir()) self.assertTrue((self.cache / "cargo/bin/rustup").exists()) def test_release_transitions_stop_vm_units_before_helper_and_binary_changes(self): for action in ["clean", "restore"]: for legacy in [0, 1]: with self.subTest(action=action, legacy=legacy): helper = "firecrab-net-helper" if legacy else "firecrab-helper" vm = "firecrab-vm-9ef8f079dfb345918d349954213dc41e.service" failed = "firecrab-vm-0794b2e090b4417eb4b02ca2261189d9.service" (self.root / "vm-units").write_text( f"{vm} loaded active running MicroVM\n" f"{failed} loaded failed failed MicroVM\n" "unrelated.service loaded active running unrelated\n") events = self.root / "events" events.unlink(missing_ok=True) result = self.run_guest(action, LEGACY_HELPER=legacy) self.assertEqual(result.returncode, 0, result.stderr) changes = events.read_text().splitlines() self.assertLess(changes.index("stop firecrab-api"), changes.index(f"stop {vm} {failed}")) self.assertLess(changes.index(f"stop {vm} {failed}"), changes.index(f"stop {helper}")) self.assertLess(changes.index(f"stop {helper}"), changes.index("daemon-reload")) self.assertLess(changes.index("daemon-reload"), changes.index("start firecrab-api")) self.assertFalse((self.root / "vm-units").read_text()) self.assertFalse(any(line.startswith("stop unrelated") for line in changes)) def test_source_deployments_and_controls_preserve_running_vm_units(self): vm = "firecrab-vm-9ef8f079dfb345918d349954213dc41e.service" listing = f"{vm} loaded active running MicroVM\n" archive = self.archive.read_bytes() for action in ["debug", "release", "start", "stop", "restart"]: with self.subTest(action=action): self.archive.write_bytes(archive) (self.root / "vm-units").write_text(listing) (self.root / "events").unlink(missing_ok=True) result = self.run_guest(action) self.assertEqual(result.returncode, 0, result.stderr) self.assertEqual((self.root / "vm-units").read_text(), listing) self.assertNotIn(vm, (self.root / "events").read_text()) def test_failed_vm_listing_or_stop_preserves_development_deployment(self): self.assertEqual(self.run_guest().returncode, 0) previous = (self.binaries / "bin").resolve() override = self.override().read_text() vm = "firecrab-vm-9ef8f079dfb345918d349954213dc41e.service" for action in ["clean", "restore"]: for failure in ["VM_LIST_FAIL", "VM_STOP_FAIL"]: with self.subTest(action=action, failure=failure): (self.root / "vm-units").write_text(f"{vm} loaded active running MicroVM\n") events = self.root / "events" events.unlink(missing_ok=True) result = self.run_guest(action, **{failure: 1}) self.assertNotEqual(result.returncode, 0) self.assertIn("[ROLLBACK]", result.stderr) self.assertNotIn("development files cleaned", result.stdout) self.assertEqual((self.binaries / "bin").resolve(), previous) self.assertEqual(self.override().read_text(), override) self.assertTrue((self.cache / "source/firecrab-api/src/main.rs").exists()) self.assertTrue((self.cache / "target").is_dir()) self.assertTrue((self.root / "vm-units").read_text()) self.assertNotIn("stop firecrab-helper", events.read_text().splitlines()) def test_clean_releases_the_previous_vm_resource_before_start_and_is_repeatable(self): resource = self.root / "tap-lock" holder = subprocess.Popen(["python3", "-c", """ import fcntl, pathlib, sys, time resource = pathlib.Path(sys.argv[1]) with resource.open('w') as lock: fcntl.flock(lock, fcntl.LOCK_EX) resource.with_name('vm-ready').touch() time.sleep(60) """, str(resource)]) def stop_holder(): if holder.poll() is None: holder.kill() holder.wait(timeout=5) self.addCleanup(stop_holder) deadline = time.monotonic() + 5 while not (self.root / "vm-ready").exists(): self.assertLess(time.monotonic(), deadline) time.sleep(0.01) (self.root / "vm-runtime-pid").write_text(str(holder.pid)) vm = "firecrab-vm-9ef8f079dfb345918d349954213dc41e.service" (self.root / "vm-units").write_text(f"{vm} loaded active running MicroVM\n") for _ in range(2): result = self.run_guest("clean") self.assertEqual(result.returncode, 0, result.stderr) self.assertIsNotNone(holder.poll(), "clean left the previous VM running") result = self.run_guest("start") self.assertEqual(result.returncode, 0, result.stderr) with resource.open('w') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) def test_controls_preserve_deployment_and_do_not_build(self): for legacy in [0, 1]: helper = "firecrab-net-helper" if legacy else "firecrab-helper" override = self.override(helper) override.parent.mkdir(parents=True, exist_ok=True) override.write_text("development helper override\n") self.override().parent.mkdir(parents=True, exist_ok=True) self.override().write_text("development API override\n") previous = self.binaries / "previous" previous.mkdir(parents=True, exist_ok=True) pointer = self.binaries / "bin" if not pointer.exists(): pointer.symlink_to(previous) for action, expected in [ ("start", [f"start {helper}", "start firecrab-api"]), ("stop", ["stop firecrab-api", f"stop {helper}"]), ("restart", ["stop firecrab-api", f"stop {helper}", f"start {helper}", "start firecrab-api"]), ]: with self.subTest(legacy=legacy, action=action): events = self.root / "events" events.unlink(missing_ok=True) result = self.run_guest(action, LEGACY_HELPER=legacy, BUILD_FAIL=1) self.assertEqual(result.returncode, 0, result.stderr) changes = [line for line in events.read_text().splitlines() if line.startswith(("start ", "stop "))] self.assertEqual(changes, expected) self.assertIn("[PASS]", result.stdout) self.assertEqual(override.read_text(), "development helper override\n") self.assertEqual(self.override().read_text(), "development API override\n") self.assertEqual(pointer.resolve(), previous.resolve()) self.assertFalse((self.root / "rustup-events").exists()) self.assertTrue(self.archive.exists()) def test_controls_share_the_deployment_lock(self): for action in ["start", "stop", "restart", "clean"]: with self.subTest(action=action): result = self.run_guest(action, LOCK_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertIn("in progress", result.stderr) self.assertFalse((self.root / "events").exists()) self.assertTrue(self.archive.exists()) def test_failed_stop_does_not_stop_the_helper_or_start_services(self): for action in ["stop", "restart"]: with self.subTest(action=action): (self.root / "events").unlink(missing_ok=True) result = self.run_guest(action, STOP_FAIL=1) self.assertEqual(result.returncode, 44) self.assertEqual((self.root / "events").read_text(), "stop firecrab-api\n") self.assertNotIn("[PASS]", result.stdout) def test_failed_helper_start_does_not_start_the_api(self): result = self.run_guest("start", HELPER_START_FAIL=1) self.assertEqual(result.returncode, 45) self.assertEqual((self.root / "events").read_text(), "start firecrab-helper\n") self.assertNotIn("[PASS]", result.stdout) def test_start_and_restart_report_an_unhealthy_api(self): for action in ["start", "restart"]: with self.subTest(action=action): result = self.run_guest(action, HEALTH_FAIL=1) self.assertNotEqual(result.returncode, 0) self.assertIn("guest services did not become ready", result.stderr) self.assertNotIn("[PASS]", result.stdout) self.assertFalse(self.override().exists()) if __name__ == "__main__": unittest.main()