Add an integration test suite that boots ISO scenarios in QEMU (#109)
* Move the unit tests under test/unit test/ is growing an integration suite; the fast, VM-free tests get their own directory so the two kinds cannot be confused. test/all keeps its role as the aggregate runner for everything that needs no built ISO. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Add an integration suite that boots ISO scenarios in QEMU test/integration installs an ISO once — unattended, from a generated cidata drive — and saves a reusable base image; each scenario in test/integration.d/ then boots a throwaway overlay with its own copy of the firmware vars. Shared machinery (VM lifecycle, QMP screendump + OCR console driving, virtual keystrokes, guest SSH, the cidata build) lives in base-test.sh, so a new scenario is one file. The first scenario proves omarchy-system-factory-reset hands a machine on without destroying a shared ESP: a Windows entry and a foreign Linux with its own machine-id, boot directory, and UKIs must survive both the staged reset and the first-boot provisioning that follows. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Let omarchy-iso-make keep the host package cache Purging /var/cache/pacman/pkg needs interactive sudo, which blocks unattended builds. --keep-pkg-cache skips the purge so test rigs and agents can build an ISO without a prompt. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
David Heinemeier HanssonandClaude Fable 5 authored and GitHub committed
1 parent
10bd6632d3
commit
0c1333837f
14 files changed
+993
-13
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,224 @@
|
||||
#!/bin/bash
|
||||
#
|
||||
# Unit tests for omarchy-cidata-load. The script takes a path prefix, so every
|
||||
# case runs against a throwaway sandbox with mount/umount/udevadm stubbed out:
|
||||
# "mounting" copies the fake drive's contents into the mountpoint, and every
|
||||
# stub logs its invocation so the cases can assert what was (not) called.
|
||||
|
||||
set -euo pipefail
|
||||
|
||||
ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)
|
||||
CIDATA_LOAD="$ROOT/configs/airootfs/usr/local/bin/omarchy-cidata-load"
|
||||
|
||||
pass() {
|
||||
printf 'ok - %s\n' "$1"
|
||||
}
|
||||
|
||||
fail() {
|
||||
local description="$1"
|
||||
local detail="${2:-}"
|
||||
|
||||
[[ -n $detail ]] && printf '%s\n' "$detail" >&2
|
||||
printf 'not ok - %s\n' "$description" >&2
|
||||
exit 1
|
||||
}
|
||||
|
||||
work=$(mktemp -d)
|
||||
trap 'chmod -R u+w "$work"; rm -rf "$work"' EXIT
|
||||
|
||||
stub_dir="$work/stubs"
|
||||
mkdir -p "$stub_dir"
|
||||
|
||||
cat >"$stub_dir/udevadm" <<'STUB'
|
||||
#!/bin/bash
|
||||
printf 'udevadm %s\n' "$*" >>"$TEST_LOG"
|
||||
STUB
|
||||
|
||||
cat >"$stub_dir/mount" <<'STUB'
|
||||
#!/bin/bash
|
||||
printf 'mount %s\n' "$*" >>"$TEST_LOG"
|
||||
[[ ${MOUNT_FAIL:-} == 1 ]] && exit 32
|
||||
device=$3 mountpoint=$4
|
||||
cp -a "$(readlink -f "$device")"/. "$mountpoint"/
|
||||
STUB
|
||||
|
||||
cat >"$stub_dir/umount" <<'STUB'
|
||||
#!/bin/bash
|
||||
printf 'umount %s\n' "$*" >>"$TEST_LOG"
|
||||
STUB
|
||||
|
||||
chmod +x "$stub_dir"/*
|
||||
|
||||
new_sandbox() {
|
||||
sandbox=$(mktemp -d "$work/sandbox.XXXXXX")
|
||||
mkdir -p "$sandbox/dev/disk/by-label" "$sandbox/root" "$sandbox/media"
|
||||
export TEST_LOG="$sandbox/calls.log"
|
||||
: >"$TEST_LOG"
|
||||
}
|
||||
|
||||
attach_drive() {
|
||||
ln -s "$sandbox/media" "$sandbox/dev/disk/by-label/$1"
|
||||
}
|
||||
|
||||
run_load() {
|
||||
PATH="$stub_dir:$PATH" "$CIDATA_LOAD" "$sandbox"
|
||||
}
|
||||
|
||||
write_required_pair() {
|
||||
echo '{"disk_config": {}}' >"$sandbox/media/user_configuration.json"
|
||||
echo '{"users": []}' >"$sandbox/media/user_credentials.json"
|
||||
}
|
||||
|
||||
# No cidata drive attached: fall back to the wizard without mounting anything.
|
||||
new_sandbox
|
||||
! run_load || fail "no drive exits non-zero"
|
||||
! grep -q '^mount ' "$TEST_LOG" || fail "no drive mounts nothing"
|
||||
pass "no drive falls back to the wizard"
|
||||
|
||||
# The probe must wait for udev to finish enumerating before concluding there
|
||||
# is no drive.
|
||||
grep -q '^udevadm settle$' "$TEST_LOG" || fail "probe settles udev first"
|
||||
pass "probe settles udev first"
|
||||
|
||||
# A drive with the full file set: everything lands in /root and the drive is
|
||||
# unmounted afterwards.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
echo "Jeff" >"$sandbox/media/user_full_name.txt"
|
||||
echo "jeff@example.com" >"$sandbox/media/user_email_address.txt"
|
||||
echo "false" >"$sandbox/media/user_encrypt_installation.txt"
|
||||
echo 'ssh-ed25519 AAAA jeff@host' >"$sandbox/media/authorized_keys"
|
||||
echo 'tskey-auth-kFAKEKEY' >"$sandbox/media/tailscale_authkey"
|
||||
run_load || fail "full file set loads"
|
||||
for file in user_configuration.json user_credentials.json user_full_name.txt user_email_address.txt user_encrypt_installation.txt authorized_keys tailscale_authkey; do
|
||||
[[ -f $sandbox/root/$file ]] || fail "full file set copies $file"
|
||||
done
|
||||
grep -q '^umount ' "$TEST_LOG" || fail "full file set unmounts the drive"
|
||||
pass "full file set loads, copies everything, and unmounts"
|
||||
|
||||
# The uppercase label variant some tools produce works too.
|
||||
new_sandbox
|
||||
attach_drive CIDATA
|
||||
write_required_pair
|
||||
run_load || fail "uppercase CIDATA label loads"
|
||||
pass "uppercase CIDATA label loads"
|
||||
|
||||
# The required pair alone is a valid autoinstall drive; the optional files
|
||||
# stay optional.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
run_load || fail "required pair alone loads"
|
||||
[[ ! -e $sandbox/root/authorized_keys ]] || fail "required pair alone copies no optional files"
|
||||
pass "required pair alone loads without optional files"
|
||||
|
||||
# Optional files are copied individually when present.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
echo 'ssh-ed25519 AAAA jeff@host' >"$sandbox/media/authorized_keys"
|
||||
run_load || fail "required pair plus authorized_keys loads"
|
||||
[[ -f $sandbox/root/authorized_keys ]] || fail "authorized_keys is copied when present"
|
||||
[[ ! -e $sandbox/root/user_full_name.txt ]] || fail "absent optional files are not copied"
|
||||
[[ ! -e $sandbox/root/tailscale_authkey ]] || fail "absent tailscale_authkey is not copied"
|
||||
pass "present optional files are copied, absent ones skipped"
|
||||
|
||||
# A defer-provisioning marker replaces user_credentials.json: deferred-provisioning installs
|
||||
# creation to first boot, so imaging rigs ship no credentials at all.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
echo '{"disk_config": {}}' >"$sandbox/media/user_configuration.json"
|
||||
: >"$sandbox/media/defer-provisioning"
|
||||
run_load || fail "config plus defer-provisioning marker loads"
|
||||
[[ -f $sandbox/root/defer-provisioning ]] || fail "defer-provisioning marker is copied"
|
||||
[[ ! -e $sandbox/root/user_credentials.json ]] || fail "defer-provisioning drive copies no credentials"
|
||||
pass "defer-provisioning marker stands in for credentials"
|
||||
|
||||
# The defer-provisioning marker and credentials can coexist (rig supplies its own LUKS
|
||||
# passphrase in the credentials file); both are copied.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
: >"$sandbox/media/defer-provisioning"
|
||||
run_load || fail "defer-provisioning marker plus credentials loads"
|
||||
[[ -f $sandbox/root/defer-provisioning && -f $sandbox/root/user_credentials.json ]] || fail "both defer-provisioning marker and credentials are copied"
|
||||
pass "defer-provisioning marker plus credentials copies both"
|
||||
|
||||
# An defer-provisioning marker without the configuration is still not an autoinstall drive.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
: >"$sandbox/media/defer-provisioning"
|
||||
! run_load || fail "defer-provisioning marker alone is not an autoinstall drive"
|
||||
[[ ! -e $sandbox/root/defer-provisioning ]] || fail "defer-provisioning marker alone copies nothing"
|
||||
pass "defer-provisioning marker alone falls back to the wizard"
|
||||
|
||||
# Stale deferred-provisioning inputs from a previous load in the same session are cleared before
|
||||
# the current drive is copied: a normal drive must not inherit an old defer-provisioning
|
||||
# marker or credentials.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
: >"$sandbox/root/defer-provisioning" # leftover from a prior deferred-provisioning load
|
||||
echo 'old-keys' >"$sandbox/root/authorized_keys"
|
||||
run_load || fail "normal drive after a stale defer-provisioning load loads"
|
||||
[[ ! -e $sandbox/root/defer-provisioning ]] || fail "stale defer-provisioning marker is cleared"
|
||||
[[ ! -e $sandbox/root/authorized_keys ]] || fail "stale optional inputs are cleared"
|
||||
pass "stale deferred-provisioning inputs are cleared before loading a normal drive"
|
||||
|
||||
# A drive that isn't an autoinstall drive at all still clears stale inputs so
|
||||
# the wizard that follows doesn't inherit them.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
echo '{"disk_config": {}}' >"$sandbox/media/user_configuration.json" # half a pair
|
||||
: >"$sandbox/root/defer-provisioning"
|
||||
! run_load || fail "half-pair drive still falls back"
|
||||
[[ ! -e $sandbox/root/defer-provisioning ]] || fail "stale defer-provisioning marker cleared even on fallback"
|
||||
pass "stale inputs are cleared even when falling back to the wizard"
|
||||
|
||||
# Half the required pair is not an autoinstall drive: unmount and fall back.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
echo '{"disk_config": {}}' >"$sandbox/media/user_configuration.json"
|
||||
! run_load || fail "half the required pair exits non-zero"
|
||||
[[ ! -e $sandbox/root/user_configuration.json ]] || fail "half the required pair copies nothing"
|
||||
grep -q '^umount ' "$TEST_LOG" || fail "half the required pair still unmounts"
|
||||
pass "half the required pair falls back and unmounts"
|
||||
|
||||
# An empty drive labeled cidata is not an autoinstall drive either.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
! run_load || fail "empty drive exits non-zero"
|
||||
grep -q '^umount ' "$TEST_LOG" || fail "empty drive still unmounts"
|
||||
pass "empty drive falls back and unmounts"
|
||||
|
||||
# A drive that will not mount falls back rather than failing the boot.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
! MOUNT_FAIL=1 run_load || fail "mount failure exits non-zero"
|
||||
! grep -q '^umount ' "$TEST_LOG" || fail "mount failure has nothing to unmount"
|
||||
pass "mount failure falls back to the wizard"
|
||||
|
||||
# A copy failure must not report a loaded drive: the install would start with
|
||||
# missing inputs. Fall back and let the wizard produce them instead.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
chmod 555 "$sandbox/root"
|
||||
! run_load 2>/dev/null || fail "copy failure exits non-zero"
|
||||
grep -q '^umount ' "$TEST_LOG" || fail "copy failure still unmounts"
|
||||
pass "copy failure falls back and unmounts"
|
||||
|
||||
# A partial copy must clean up after itself: the wizard removes only what it
|
||||
# writes, so anything the loader left behind would leak into the interactive
|
||||
# install that follows.
|
||||
new_sandbox
|
||||
attach_drive cidata
|
||||
write_required_pair
|
||||
echo 'ssh-ed25519 AAAA jeff@host' >"$sandbox/media/authorized_keys"
|
||||
mkdir "$sandbox/root/user_credentials.json"
|
||||
! run_load 2>/dev/null || fail "partial copy exits non-zero"
|
||||
[[ ! -e $sandbox/root/user_configuration.json ]] || fail "partial copy removes what it copied"
|
||||
[[ ! -e $sandbox/root/authorized_keys ]] || fail "partial copy leaves no authorized_keys behind"
|
||||
pass "partial copy cleans up what it copied"
|
||||
@@ -0,0 +1,131 @@
|
||||
#!/bin/bash
|
||||
#
|
||||
# create_partition() must report the number parted actually assigned, never a
|
||||
# predicted one. The regression this pins: the configurator used to compute
|
||||
# "highest existing partition + 1", while parted fills the lowest free GPT
|
||||
# slot. Any disk whose numbering has a hole — what deleting a partition to
|
||||
# free space leaves behind — then had every later command (cryptsetup, mkfs,
|
||||
# mount) pointed at device nodes that did not exist, surfacing much later as
|
||||
# "protected mode: /mnt is not a mountpoint".
|
||||
#
|
||||
# parted operates on image files directly, so this needs no root, no loop
|
||||
# devices, and no real disk.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)
|
||||
LIB="$ROOT/configs/airootfs/usr/share/omarchy-iso/disk-partitioning.sh"
|
||||
|
||||
if ! command -v parted >/dev/null 2>&1; then
|
||||
echo "SKIP: parted is not installed"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# shellcheck source=../configs/airootfs/usr/share/omarchy-iso/disk-partitioning.sh
|
||||
source "$LIB"
|
||||
|
||||
WORK=$(mktemp -d)
|
||||
trap 'rm -rf "$WORK"' EXIT
|
||||
|
||||
IMG="$WORK/disk.img"
|
||||
MIB=$((1024 * 1024))
|
||||
failures=0
|
||||
|
||||
check() {
|
||||
local label="$1" expected="$2" actual="$3"
|
||||
if [[ $expected == "$actual" ]]; then
|
||||
printf ' ok %s\n' "$label"
|
||||
else
|
||||
printf ' FAIL %s: expected %s, got %s\n' "$label" "$expected" "$actual"
|
||||
failures=$((failures + 1))
|
||||
fi
|
||||
}
|
||||
|
||||
# parted reports a partition's size inclusive of both endpoints, so it lands
|
||||
# one sector above (end - start). Same slack create_partition allows.
|
||||
check_size() {
|
||||
local label="$1" want="$2" actual="${3:-0}" delta
|
||||
delta=$((actual - want))
|
||||
(( delta < 0 )) && delta=$((-delta))
|
||||
if (( delta <= MIB )); then
|
||||
printf ' ok %s\n' "$label"
|
||||
else
|
||||
printf ' FAIL %s: expected ~%s, got %s\n' "$label" "$want" "$actual"
|
||||
failures=$((failures + 1))
|
||||
fi
|
||||
}
|
||||
|
||||
# parted's end position is inclusive, so partitions laid back to back overlap
|
||||
# by a sector and are refused. Leave a MiB between them.
|
||||
mkpart_mib() {
|
||||
parted --script "$IMG" mkpart "$1" "$2" "$(($3 * MIB))B" "$(($4 * MIB))B"
|
||||
}
|
||||
|
||||
# A disk with holes: four partitions, then delete the middle two. Slots 2 and
|
||||
# 3 are free while the highest number in use is 4 — the shape of the reported
|
||||
# failure, where the user deleted a partition to make room.
|
||||
build_holey_disk() {
|
||||
rm -f "$IMG"
|
||||
truncate -s 4G "$IMG"
|
||||
parted --script "$IMG" mklabel gpt
|
||||
mkpart_mib one ext4 1 200
|
||||
mkpart_mib two ext4 201 400
|
||||
mkpart_mib three ext4 401 600
|
||||
mkpart_mib four ext4 601 800
|
||||
parted --script "$IMG" rm 2
|
||||
parted --script "$IMG" rm 3
|
||||
created_parts=()
|
||||
}
|
||||
|
||||
echo "==> numbering on a disk with holes"
|
||||
build_holey_disk
|
||||
check "existing numbers" "1 4" "$(partition_numbers "$IMG" | sort | tr '\n' ' ' | sed 's/ $//')"
|
||||
|
||||
create_partition "$IMG" "$((1000 * MIB))" "$((1200 * MIB))" fat32 OMARCHY_EFI
|
||||
check "create_partition succeeded (esp)" "0" "$?"
|
||||
esp_num="$created_partition_number"
|
||||
check "esp took the lowest free slot" "2" "$esp_num"
|
||||
|
||||
# Starts a MiB past the ESP, the way run_partition_decide aligns ROOT_START_B
|
||||
# up from EFI_END_B + 1.
|
||||
create_partition "$IMG" "$((1201 * MIB))" "$((2000 * MIB))" btrfs OMARCHY_ROOT
|
||||
check "create_partition succeeded (root)" "0" "$?"
|
||||
root_num="$created_partition_number"
|
||||
check "root took the next free slot" "3" "$root_num"
|
||||
|
||||
# The old prediction. If either of these ever matches again, the regression is
|
||||
# back: the highest number in use was 4, so the guess would have been 5 and 6.
|
||||
check "esp is not the predicted number" "not-5" "$([[ $esp_num == 5 ]] && echo 5 || echo not-5)"
|
||||
check "root is not the predicted number" "not-6" "$([[ $root_num == 6 ]] && echo 6 || echo not-6)"
|
||||
|
||||
check "both creations tracked for rollback" "2 3" "${created_parts[*]}"
|
||||
check_size "esp size" "$((200 * MIB))" "$(partition_size_bytes "$IMG" "$esp_num")"
|
||||
check_size "root size" "$((799 * MIB))" "$(partition_size_bytes "$IMG" "$root_num")"
|
||||
|
||||
echo "==> rollback reclaims only what this run created"
|
||||
rollback_created_parts "$IMG"
|
||||
check "created partitions removed" "1 4" "$(partition_numbers "$IMG" | sort | tr '\n' ' ' | sed 's/ $//')"
|
||||
check "rollback list cleared" "0" "${#created_parts[@]}"
|
||||
|
||||
echo "==> numbering on a disk without holes"
|
||||
rm -f "$IMG"
|
||||
truncate -s 4G "$IMG"
|
||||
parted --script "$IMG" mklabel gpt
|
||||
mkpart_mib one ext4 1 200
|
||||
mkpart_mib two ext4 201 400
|
||||
created_parts=()
|
||||
|
||||
create_partition "$IMG" "$((1000 * MIB))" "$((1200 * MIB))" fat32 OMARCHY_EFI
|
||||
check "appends when there is no hole" "3" "$created_partition_number"
|
||||
|
||||
echo "==> a refused creation stays out of the rollback list"
|
||||
created_parts=()
|
||||
create_partition "$IMG" "$((100 * MIB))" "$((300 * MIB))" ext4 OVERLAP
|
||||
check "overlapping creation failed" "1" "$?"
|
||||
check "nothing tracked" "0" "${#created_parts[@]}"
|
||||
|
||||
if (( failures > 0 )); then
|
||||
printf '\n%d check(s) failed\n' "$failures"
|
||||
exit 1
|
||||
fi
|
||||
printf '\nall checks passed\n'
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Unit tests for the orchestrator's configure_ssh_access phase.
|
||||
|
||||
The phase touches the target only through the filesystem and arch-chroot, so
|
||||
the tests run it against a temp directory with subprocess.run recorded and the
|
||||
ufw side effect (writing user.rules) simulated by the fake.
|
||||
"""
|
||||
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from subprocess import CompletedProcess
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "configs/airootfs/usr/share/omarchy-iso"))
|
||||
|
||||
# phases_impl imports the archinstall adapter at module scope, which pulls in
|
||||
# the archinstall library that only exists on the live ISO. configure_ssh_access
|
||||
# never touches it, so stub the adapter out before the import.
|
||||
sys.modules["orchestrator.archinstall_adapter"] = types.ModuleType("orchestrator.archinstall_adapter")
|
||||
|
||||
from orchestrator import phases_impl # noqa: E402
|
||||
|
||||
UFW_RULE = "-A ufw-user-input -p tcp --dport 22 -j ACCEPT\n"
|
||||
|
||||
|
||||
class ConfigureSshAccessTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name)
|
||||
self.calls = []
|
||||
|
||||
info_patch = mock.patch.object(phases_impl, "info")
|
||||
info_patch.start()
|
||||
self.addCleanup(info_patch.stop)
|
||||
|
||||
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=self.fake_run)
|
||||
run_patch.start()
|
||||
self.addCleanup(run_patch.stop)
|
||||
|
||||
self.ufw_writes_rule = True
|
||||
|
||||
def fake_run(self, cmd, **kwargs):
|
||||
self.calls.append(cmd)
|
||||
if cmd[2] == "ufw":
|
||||
if self.ufw_writes_rule:
|
||||
rules = self.target / "etc" / "ufw" / "user.rules"
|
||||
rules.parent.mkdir(parents=True, exist_ok=True)
|
||||
rules.write_text(UFW_RULE)
|
||||
# ufw cannot reach netfilter inside a chroot and exits non-zero
|
||||
# even when it recorded the rule; the phase must not trust this.
|
||||
return CompletedProcess(cmd, 1)
|
||||
return CompletedProcess(cmd, 0)
|
||||
|
||||
def ctx(self, authorized_keys=None):
|
||||
authorized_keys_path = None
|
||||
if authorized_keys is not None:
|
||||
authorized_keys_path = self.target / "authorized_keys"
|
||||
authorized_keys_path.write_text(authorized_keys)
|
||||
return types.SimpleNamespace(target=self.target, username="jeff", authorized_keys_path=authorized_keys_path, defer_provisioning=False)
|
||||
|
||||
def configure(self, **kwargs):
|
||||
phases_impl.configure_ssh_access(self.ctx(**kwargs))
|
||||
|
||||
def authorized_keys(self):
|
||||
return self.target / "home" / "jeff" / ".ssh" / "authorized_keys"
|
||||
|
||||
def chrooted(self, program):
|
||||
return [cmd for cmd in self.calls if cmd[:2] == ["arch-chroot", str(self.target)] and cmd[2] == program]
|
||||
|
||||
def test_no_authorized_keys_is_a_no_op(self):
|
||||
self.configure()
|
||||
self.assertEqual(self.calls, [])
|
||||
self.assertFalse((self.target / "home" / "jeff" / ".ssh").exists())
|
||||
|
||||
def test_installs_keys_one_per_line(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\nssh-rsa BBBB jeff@work\n")
|
||||
self.assertEqual(self.authorized_keys().read_text(), "ssh-ed25519 AAAA jeff@host\nssh-rsa BBBB jeff@work\n")
|
||||
|
||||
def test_drops_blank_lines_and_comments(self):
|
||||
self.configure(authorized_keys="# work laptop\n\n ssh-ed25519 AAAA jeff@host \n")
|
||||
self.assertEqual(self.authorized_keys().read_text(), "ssh-ed25519 AAAA jeff@host\n")
|
||||
|
||||
def test_keys_with_options_pass_through(self):
|
||||
key = 'command="/usr/bin/true",no-pty ssh-ed25519 AAAA jeff@host'
|
||||
self.configure(authorized_keys=f"{key}\n")
|
||||
self.assertEqual(self.authorized_keys().read_text(), f"{key}\n")
|
||||
|
||||
def test_ssh_dir_is_private(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
mode = stat.S_IMODE((self.target / "home" / "jeff" / ".ssh").stat().st_mode)
|
||||
self.assertEqual(mode, 0o700)
|
||||
|
||||
def test_authorized_keys_is_private(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
self.assertEqual(stat.S_IMODE(self.authorized_keys().stat().st_mode), 0o600)
|
||||
|
||||
def test_chowns_ssh_dir_to_the_user(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
self.assertEqual(self.chrooted("chown"), [
|
||||
["arch-chroot", str(self.target), "chown", "-R", "jeff:jeff", "/home/jeff/.ssh"],
|
||||
])
|
||||
|
||||
def test_enables_sshd(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
self.assertEqual(self.chrooted("systemctl"), [
|
||||
["arch-chroot", str(self.target), "systemctl", "enable", "sshd.service"],
|
||||
])
|
||||
|
||||
def test_allows_ssh_through_ufw_despite_chroot_exit_status(self):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
self.assertEqual(self.chrooted("ufw"), [
|
||||
["arch-chroot", str(self.target), "ufw", "allow", "ssh"],
|
||||
])
|
||||
|
||||
def test_fails_when_ufw_does_not_record_the_rule(self):
|
||||
self.ufw_writes_rule = False
|
||||
with self.assertRaisesRegex(RuntimeError, "allow rule for port 22"):
|
||||
self.configure(authorized_keys="ssh-ed25519 AAAA jeff@host\n")
|
||||
|
||||
def test_empty_file_fails_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "contains no SSH keys"):
|
||||
self.configure(authorized_keys="")
|
||||
|
||||
def test_blank_only_file_fails_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "contains no SSH keys"):
|
||||
self.configure(authorized_keys="\n \n")
|
||||
|
||||
def test_comment_only_file_fails_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "contains no SSH keys"):
|
||||
self.configure(authorized_keys="# no keys here\n")
|
||||
|
||||
|
||||
class OptionalPathTest(unittest.TestCase):
|
||||
def test_unset_and_missing_paths_are_none(self):
|
||||
from orchestrator.context import _optional_path
|
||||
|
||||
self.assertIsNone(_optional_path(None))
|
||||
self.assertIsNone(_optional_path(""))
|
||||
self.assertIsNone(_optional_path("/does/not/exist/authorized_keys"))
|
||||
|
||||
def test_existing_path_is_returned(self):
|
||||
from orchestrator.context import _optional_path
|
||||
|
||||
with tempfile.NamedTemporaryFile() as f:
|
||||
self.assertEqual(_optional_path(f.name), Path(f.name))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,164 @@
|
||||
"""Unit tests for the orchestrator's configure_tailscale phase.
|
||||
|
||||
Same harness as the SSH access tests: the phase touches the target only
|
||||
through the filesystem and arch-chroot, so it runs against a temp directory
|
||||
with subprocess.run recorded and the ufw side effect (writing user.rules)
|
||||
simulated by the fake.
|
||||
"""
|
||||
|
||||
import stat
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from subprocess import CompletedProcess
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "configs/airootfs/usr/share/omarchy-iso"))
|
||||
|
||||
# phases_impl imports the archinstall adapter at module scope, which pulls in
|
||||
# the archinstall library that only exists on the live ISO. configure_tailscale
|
||||
# never touches it, so stub the adapter out before the import.
|
||||
sys.modules["orchestrator.archinstall_adapter"] = types.ModuleType("orchestrator.archinstall_adapter")
|
||||
|
||||
from orchestrator import phases_impl # noqa: E402
|
||||
|
||||
UFW_RULE = "-A ufw-user-input -i tailscale0 -j ACCEPT\n"
|
||||
|
||||
|
||||
class ConfigureTailscaleTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name)
|
||||
self.calls = []
|
||||
|
||||
info_patch = mock.patch.object(phases_impl, "info")
|
||||
info_patch.start()
|
||||
self.addCleanup(info_patch.stop)
|
||||
|
||||
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=self.fake_run)
|
||||
run_patch.start()
|
||||
self.addCleanup(run_patch.stop)
|
||||
|
||||
self.ufw_writes_rule = True
|
||||
|
||||
def fake_run(self, cmd, **kwargs):
|
||||
self.calls.append(cmd)
|
||||
if cmd[2] == "ufw":
|
||||
if self.ufw_writes_rule:
|
||||
rules = self.target / "etc" / "ufw" / "user.rules"
|
||||
rules.parent.mkdir(parents=True, exist_ok=True)
|
||||
rules.write_text(UFW_RULE)
|
||||
# ufw cannot reach netfilter inside a chroot and exits non-zero
|
||||
# even when it recorded the rule; the phase must not trust this.
|
||||
return CompletedProcess(cmd, 1)
|
||||
return CompletedProcess(cmd, 0)
|
||||
|
||||
def ctx(self, authkey=None, tailscale_installed=True):
|
||||
authkey_path = None
|
||||
if authkey is not None:
|
||||
authkey_path = self.target / "tailscale_authkey"
|
||||
authkey_path.write_text(authkey)
|
||||
if tailscale_installed:
|
||||
binary = self.target / "usr" / "bin" / "tailscale"
|
||||
binary.parent.mkdir(parents=True, exist_ok=True)
|
||||
binary.touch()
|
||||
return types.SimpleNamespace(target=self.target, tailscale_authkey_path=authkey_path)
|
||||
|
||||
def configure(self, **kwargs):
|
||||
phases_impl.configure_tailscale(self.ctx(**kwargs))
|
||||
|
||||
def staged_key(self):
|
||||
return self.target / "etc" / "tailscale" / "authkey"
|
||||
|
||||
def unit(self):
|
||||
return self.target / "etc" / "systemd" / "system" / "omarchy-tailscale-join.service"
|
||||
|
||||
def chrooted(self, program):
|
||||
return [cmd for cmd in self.calls if cmd[:2] == ["arch-chroot", str(self.target)] and cmd[2] == program]
|
||||
|
||||
def test_no_authkey_is_a_no_op(self):
|
||||
self.configure()
|
||||
self.assertEqual(self.calls, [])
|
||||
self.assertFalse((self.target / "etc" / "tailscale").exists())
|
||||
|
||||
def test_stages_the_key(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
self.assertEqual(self.staged_key().read_text(), "tskey-auth-kFAKEKEY\n")
|
||||
|
||||
def test_drops_blank_lines_and_comments(self):
|
||||
self.configure(authkey="# reusable, tagged\n\n tskey-auth-kFAKEKEY \n")
|
||||
self.assertEqual(self.staged_key().read_text(), "tskey-auth-kFAKEKEY\n")
|
||||
|
||||
def test_key_dir_and_file_are_root_private(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
self.assertEqual(stat.S_IMODE((self.target / "etc" / "tailscale").stat().st_mode), 0o700)
|
||||
self.assertEqual(stat.S_IMODE(self.staged_key().stat().st_mode), 0o600)
|
||||
|
||||
def test_installs_the_join_unit(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
text = self.unit().read_text()
|
||||
self.assertIn("ConditionPathExists=/etc/tailscale/authkey", text)
|
||||
# The key must not outlive a successful join, the unit must not run
|
||||
# again after one, and cleanup must be sequenced after the join
|
||||
# succeeds -- inside the script, not in ExecStartPost.
|
||||
self.assertIn(
|
||||
"ExecStart=/usr/bin/sh -c 'until tailscale up --auth-key file:/etc/tailscale/authkey;"
|
||||
" do sleep 15; done; rm -f /etc/tailscale/authkey;"
|
||||
" systemctl disable omarchy-tailscale-join.service'",
|
||||
text,
|
||||
)
|
||||
|
||||
def test_join_unit_does_not_hold_up_boot(self):
|
||||
# A Type=oneshot wanted by multi-user.target holds the whole boot
|
||||
# (SDDM included) hostage until the join finishes or times out --
|
||||
# target units implicitly gain After= for their Wants=.
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
text = self.unit().read_text()
|
||||
self.assertIn("Type=simple", text)
|
||||
self.assertNotIn("Type=oneshot", text)
|
||||
self.assertNotIn("TimeoutStartSec", text)
|
||||
|
||||
def test_unit_avoids_systemd_variable_expansion(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
self.assertNotIn("$", self.unit().read_text())
|
||||
|
||||
def test_enables_tailscaled_and_the_join(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
self.assertEqual(self.chrooted("systemctl"), [
|
||||
["arch-chroot", str(self.target), "systemctl", "enable",
|
||||
"tailscaled.service", "omarchy-tailscale-join.service"],
|
||||
])
|
||||
|
||||
def test_allows_tailnet_traffic_through_ufw_despite_chroot_exit_status(self):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
self.assertEqual(self.chrooted("ufw"), [
|
||||
["arch-chroot", str(self.target), "ufw", "allow", "in", "on", "tailscale0"],
|
||||
])
|
||||
|
||||
def test_fails_when_ufw_does_not_record_the_rule(self):
|
||||
self.ufw_writes_rule = False
|
||||
with self.assertRaisesRegex(RuntimeError, "allow rule for tailscale0"):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n")
|
||||
|
||||
def test_fails_when_tailscale_is_not_on_the_target(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "not installed on the target"):
|
||||
self.configure(authkey="tskey-auth-kFAKEKEY\n", tailscale_installed=False)
|
||||
|
||||
def test_empty_file_fails_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "contains no auth key"):
|
||||
self.configure(authkey="")
|
||||
|
||||
def test_comment_only_file_fails_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "contains no auth key"):
|
||||
self.configure(authkey="# no key here\n")
|
||||
|
||||
def test_multiple_keys_fail_the_phase(self):
|
||||
with self.assertRaisesRegex(RuntimeError, "expected exactly one"):
|
||||
self.configure(authkey="tskey-auth-kONE\ntskey-auth-kTWO\n")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,86 @@
|
||||
#!/usr/bin/python
|
||||
|
||||
import importlib.util
|
||||
import re
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
MODULE_PATH = ROOT / "configs/airootfs/usr/share/omarchy-iso/orchestrator/keyboard.py"
|
||||
SPEC = importlib.util.spec_from_file_location("installer_keyboard", MODULE_PATH)
|
||||
KEYBOARD = importlib.util.module_from_spec(SPEC)
|
||||
assert SPEC.loader
|
||||
SPEC.loader.exec_module(KEYBOARD)
|
||||
|
||||
# The layout list lives in the Omarchy runtime now, shared verbatim with the
|
||||
# first-boot owner setup; build-iso.sh vendors it onto the ISO. Read it from
|
||||
# wherever this checkout can see a runtime, and skip the coverage test rather
|
||||
# than fail when none is around (a bare CI checkout of just this repo).
|
||||
SETUP_FORM_CANDIDATES = (
|
||||
Path("/omarchy-source/install/provisioning/setup-form.sh"),
|
||||
ROOT.parent / "omarchy/install/provisioning/setup-form.sh",
|
||||
Path("/usr/share/omarchy/install/provisioning/setup-form.sh"),
|
||||
)
|
||||
|
||||
|
||||
def supported_keymaps():
|
||||
for candidate in SETUP_FORM_CANDIDATES:
|
||||
if not candidate.is_file():
|
||||
continue
|
||||
block = re.search(
|
||||
r"OMARCHY_KEYBOARD_LAYOUTS=\$'(.*?)'\n", candidate.read_text(), re.DOTALL
|
||||
)
|
||||
assert block, f"no layout list in {candidate}"
|
||||
return [line.split("|", 1)[1] for line in block.group(1).splitlines()]
|
||||
return None
|
||||
|
||||
|
||||
SUPPORTED_KEYMAPS = supported_keymaps()
|
||||
|
||||
|
||||
class KeyboardConfigurationTest(unittest.TestCase):
|
||||
def target(self, directory: str) -> Path:
|
||||
target = Path(directory)
|
||||
(target / "etc").mkdir()
|
||||
(target / "etc/vconsole.conf").write_text(
|
||||
"KEYMAP=us\nFONT=default8x16\n"
|
||||
)
|
||||
return target
|
||||
|
||||
def test_writes_keymap_and_xkb_settings_and_preserves_font(self):
|
||||
# XKBLAYOUT is load-bearing: omarchy's detect-keyboard-layout.sh copies
|
||||
# it into Hyprland's kb_layout on the installed system.
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
target = self.target(directory)
|
||||
self.assertTrue(KEYBOARD.configure_keyboard(target, "us"))
|
||||
lines = (target / "etc/vconsole.conf").read_text().splitlines()
|
||||
self.assertIn("KEYMAP=us", lines)
|
||||
self.assertIn("XKBLAYOUT=us", lines)
|
||||
self.assertEqual(lines.count("FONT=default8x16"), 1)
|
||||
|
||||
def test_all_configurator_keymaps_are_known_to_localectl(self):
|
||||
if SUPPORTED_KEYMAPS is None:
|
||||
self.skipTest("no Omarchy runtime checkout to read the layout list from")
|
||||
for keymap in SUPPORTED_KEYMAPS:
|
||||
with self.subTest(keymap=keymap), tempfile.TemporaryDirectory() as directory:
|
||||
target = self.target(directory)
|
||||
self.assertTrue(KEYBOARD.configure_keyboard(target, keymap))
|
||||
lines = (target / "etc/vconsole.conf").read_text().splitlines()
|
||||
self.assertIn(f"KEYMAP={keymap}", lines)
|
||||
self.assertEqual(lines.count("FONT=default8x16"), 1)
|
||||
|
||||
def test_unknown_and_empty_keymaps_match_archinstall_behavior(self):
|
||||
for keymap, expected in (("definitely-not-a-keymap", False), ("", True)):
|
||||
with self.subTest(keymap=keymap), tempfile.TemporaryDirectory() as directory:
|
||||
target = self.target(directory)
|
||||
self.assertEqual(KEYBOARD.configure_keyboard(target, keymap), expected)
|
||||
self.assertEqual(
|
||||
(target / "etc/vconsole.conf").read_text(),
|
||||
"KEYMAP=us\nFONT=default8x16\n",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,79 @@
|
||||
#!/usr/bin/python
|
||||
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
PRUNER = ROOT / "builder/prune-offline-mirror.sh"
|
||||
|
||||
|
||||
class OfflineMirrorPruningTest(unittest.TestCase):
|
||||
def run_pruner(self, mirror: Path, required: list[str]):
|
||||
return subprocess.run(
|
||||
["bash", str(PRUNER), str(mirror)],
|
||||
input="".join(f"{filename}\n" for filename in required),
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
|
||||
def add_file(self, mirror: Path, filename: str):
|
||||
(mirror / filename).write_bytes(b"package")
|
||||
|
||||
def test_keeps_only_exactly_selected_package_archives(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
mirror = Path(directory)
|
||||
keep = "keep-2.0-1-x86_64.pkg.tar.zst"
|
||||
other = "other-1.0-1-any.pkg.tar.zst"
|
||||
stale_version = "keep-1.0-1-x86_64.pkg.tar.zst"
|
||||
removed_package = "removed-1.0-1-x86_64.pkg.tar.zst"
|
||||
orphan_signature = "interrupted-1.0-1-x86_64.pkg.tar.zst.sig"
|
||||
|
||||
for filename in (keep, other, stale_version, removed_package):
|
||||
self.add_file(mirror, filename)
|
||||
for filename in (keep, stale_version, removed_package):
|
||||
self.add_file(mirror, f"{filename}.sig")
|
||||
self.add_file(mirror, orphan_signature)
|
||||
self.add_file(mirror, "offline.db.tar.gz")
|
||||
|
||||
result = self.run_pruner(mirror, [keep, other])
|
||||
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertEqual(
|
||||
{path.name for path in mirror.iterdir()},
|
||||
{keep, f"{keep}.sig", other, "offline.db.tar.gz"},
|
||||
)
|
||||
self.assertIn(stale_version, result.stdout)
|
||||
self.assertIn(removed_package, result.stdout)
|
||||
|
||||
def test_missing_selection_aborts_before_deleting_anything(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
mirror = Path(directory)
|
||||
cached = "cached-1.0-1-x86_64.pkg.tar.zst"
|
||||
self.add_file(mirror, cached)
|
||||
|
||||
result = self.run_pruner(
|
||||
mirror, ["missing-1.0-1-x86_64.pkg.tar.zst"]
|
||||
)
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertTrue((mirror / cached).exists())
|
||||
self.assertIn("selected packages are missing", result.stderr)
|
||||
|
||||
def test_empty_selection_aborts_before_deleting_anything(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
mirror = Path(directory)
|
||||
cached = "cached-1.0-1-x86_64.pkg.tar.zst"
|
||||
self.add_file(mirror, cached)
|
||||
|
||||
result = self.run_pruner(mirror, [])
|
||||
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertTrue((mirror / cached).exists())
|
||||
self.assertIn("empty package selection", result.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,440 @@
|
||||
"""Unit tests for deferred-provisioning install support in the orchestrator.
|
||||
|
||||
Covers the InstallContext deferred-provisioning plumbing (marker file, credential-less input,
|
||||
throwaway LUKS passphrase injection) and the stage_provisioning_state /
|
||||
create_factory_snapshot / configure_login phases against a temp target with
|
||||
subprocess mocked out.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from subprocess import CompletedProcess
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "configs/airootfs/usr/share/omarchy-iso"))
|
||||
|
||||
sys.modules.setdefault(
|
||||
"orchestrator.archinstall_adapter", types.ModuleType("orchestrator.archinstall_adapter")
|
||||
)
|
||||
|
||||
from orchestrator import phases_impl # noqa: E402
|
||||
from orchestrator.context import InstallContext # noqa: E402
|
||||
|
||||
|
||||
class ContextDeferProvisioningTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.dir = Path(self.tmp.name)
|
||||
self.state_dir = self.dir / "state"
|
||||
|
||||
self.env = {
|
||||
"OMARCHY_INSTALL_CONFIG": str(self.dir / "user_configuration.json"),
|
||||
"OMARCHY_INSTALL_CREDS": str(self.dir / "user_credentials.json"),
|
||||
"OMARCHY_INSTALL_STATE_DIR": str(self.state_dir),
|
||||
}
|
||||
|
||||
def write_config(self, config):
|
||||
(self.dir / "user_configuration.json").write_text(json.dumps(config))
|
||||
|
||||
def write_creds(self, creds):
|
||||
(self.dir / "user_credentials.json").write_text(json.dumps(creds))
|
||||
|
||||
def from_env(self, **extra_env):
|
||||
env = {**self.env, **extra_env}
|
||||
with mock.patch.dict(os.environ, env, clear=False):
|
||||
for key in ("OMARCHY_INSTALL_DEFER_PROVISIONING_FILE",):
|
||||
if key not in env:
|
||||
os.environ.pop(key, None)
|
||||
return InstallContext.from_env()
|
||||
|
||||
def base_config(self, defer_provisioning=None, disk_encryption=None):
|
||||
config = {
|
||||
"disk_config": {"config_type": "default_layout"},
|
||||
"omarchy_install": {"mode": "full_disk", "target_mount": "/mnt"},
|
||||
}
|
||||
if defer_provisioning is not None:
|
||||
config["omarchy_install"]["defer_provisioning"] = defer_provisioning
|
||||
if disk_encryption is not None:
|
||||
config["disk_config"]["disk_encryption"] = disk_encryption
|
||||
return config
|
||||
|
||||
def test_defer_provisioning_flag_from_config(self):
|
||||
self.write_config(self.base_config(defer_provisioning=True))
|
||||
ctx = self.from_env()
|
||||
self.assertTrue(ctx.defer_provisioning)
|
||||
self.assertEqual(ctx.user_credentials, {"users": []})
|
||||
self.assertEqual(ctx.username, "")
|
||||
|
||||
def test_marker_file_arms_deferred_provisioning_without_credentials(self):
|
||||
self.write_config(self.base_config())
|
||||
marker = self.dir / "defer_provisioning"
|
||||
marker.touch()
|
||||
ctx = self.from_env(OMARCHY_INSTALL_DEFER_PROVISIONING_FILE=str(marker))
|
||||
self.assertTrue(ctx.defer_provisioning)
|
||||
self.assertTrue(ctx.omarchy_install["defer_provisioning"])
|
||||
|
||||
def test_absent_marker_file_is_not_deferred_provisioning(self):
|
||||
self.write_config(self.base_config())
|
||||
self.write_creds({"users": [{"username": "jeff"}]})
|
||||
ctx = self.from_env(OMARCHY_INSTALL_DEFER_PROVISIONING_FILE=str(self.dir / "missing-defer_provisioning"))
|
||||
self.assertFalse(ctx.defer_provisioning)
|
||||
self.assertEqual(ctx.username, "jeff")
|
||||
|
||||
def test_missing_credentials_without_defer_provisioning_raises(self):
|
||||
self.write_config(self.base_config())
|
||||
with self.assertRaisesRegex(RuntimeError, "credentials file missing"):
|
||||
self.from_env()
|
||||
|
||||
def test_deferred_provisioning_generates_throwaway_luks_passphrase(self):
|
||||
self.write_config(self.base_config(defer_provisioning=True, disk_encryption={
|
||||
"encryption_type": "luks", "partitions": ["x"],
|
||||
}))
|
||||
ctx = self.from_env()
|
||||
|
||||
password = ctx.user_configuration["disk_config"]["disk_encryption"]["encryption_password"]
|
||||
self.assertTrue(password)
|
||||
self.assertEqual(ctx.user_credentials["encryption_password"], password)
|
||||
|
||||
# archinstall reads both files from disk; the injected passphrase must
|
||||
# be present in each.
|
||||
arch_config = json.loads(ctx.arch_config_path.read_text())
|
||||
self.assertEqual(
|
||||
arch_config["disk_config"]["disk_encryption"]["encryption_password"], password
|
||||
)
|
||||
synthesized = json.loads(ctx.creds_path.read_text())
|
||||
self.assertEqual(synthesized["encryption_password"], password)
|
||||
|
||||
def test_deferred_provisioning_reuses_rig_supplied_passphrase(self):
|
||||
self.write_config(self.base_config(defer_provisioning=True, disk_encryption={
|
||||
"encryption_type": "luks", "partitions": ["x"],
|
||||
}))
|
||||
self.write_creds({"users": [], "encryption_password": "rig-secret"})
|
||||
ctx = self.from_env()
|
||||
self.assertEqual(
|
||||
ctx.user_configuration["disk_config"]["disk_encryption"]["encryption_password"],
|
||||
"rig-secret",
|
||||
)
|
||||
|
||||
def test_deferred_provisioning_discards_account_material(self):
|
||||
# A rig credentials file must not smuggle in users or a root password.
|
||||
self.write_config(self.base_config(defer_provisioning=True))
|
||||
self.write_creds({
|
||||
"users": [{"username": "backdoor", "enc_password": "x", "sudo": True}],
|
||||
"root_enc_password": "x",
|
||||
"encryption_password": "rig-secret",
|
||||
})
|
||||
ctx = self.from_env()
|
||||
self.assertEqual(ctx.user_credentials["users"], [])
|
||||
self.assertNotIn("root_enc_password", ctx.user_credentials)
|
||||
self.assertEqual(ctx.user_credentials["encryption_password"], "rig-secret")
|
||||
synthesized = json.loads(ctx.creds_path.read_text())
|
||||
self.assertEqual(synthesized["users"], [])
|
||||
self.assertNotIn("root_enc_password", synthesized)
|
||||
|
||||
def test_defer_provisioning_strips_account_fields(self):
|
||||
config = self.base_config(defer_provisioning=True)
|
||||
config["!users"] = [{"username": "rig"}]
|
||||
config["!root-password"] = "hunter2"
|
||||
config["root_enc_password"] = "x"
|
||||
config["auth_config"] = {"users": [{"username": "rig"}], "root_enc_password": "x"}
|
||||
self.write_config(config)
|
||||
ctx = self.from_env()
|
||||
|
||||
arch_config = json.loads(ctx.arch_config_path.read_text())
|
||||
for key in ("!users", "!root-password", "root_enc_password", "users"):
|
||||
self.assertNotIn(key, arch_config)
|
||||
self.assertEqual(arch_config["auth_config"], {})
|
||||
|
||||
def test_deferred_provisioning_unencrypted_touches_nothing(self):
|
||||
self.write_config(self.base_config(defer_provisioning=True))
|
||||
ctx = self.from_env()
|
||||
self.assertNotIn("encryption_password", ctx.user_credentials)
|
||||
|
||||
|
||||
def make_ctx(target, **overrides):
|
||||
defaults = dict(
|
||||
target=target,
|
||||
defer_provisioning=True,
|
||||
encrypt=False,
|
||||
username="",
|
||||
omarchy_install={"mode": "full_disk", "defer_provisioning": True, "storage": {}},
|
||||
user_configuration={"disk_config": {}},
|
||||
user_credentials={"users": []},
|
||||
state_dir=target / "state",
|
||||
authorized_keys_path=None,
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return types.SimpleNamespace(**defaults)
|
||||
|
||||
|
||||
class StageProvisioningStateTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name) / "mnt"
|
||||
self.target.mkdir()
|
||||
|
||||
info_patch = mock.patch.object(phases_impl, "info")
|
||||
info_patch.start()
|
||||
self.addCleanup(info_patch.stop)
|
||||
|
||||
# A fake bundled Node tarball on the "live ISO".
|
||||
self.packages = Path(self.tmp.name) / "opt-packages"
|
||||
self.packages.mkdir()
|
||||
(self.packages / "node-v24.0.0-linux-x64.tar.gz").write_bytes(b"node")
|
||||
node_patch = mock.patch.object(phases_impl, "NODE_PACKAGES_DIR", self.packages)
|
||||
node_patch.start()
|
||||
self.addCleanup(node_patch.stop)
|
||||
|
||||
def install_runtime_provisioning_support(self):
|
||||
service = self.target / "usr/share/omarchy/install/provisioning/omarchy-provision-owner.service"
|
||||
service.parent.mkdir(parents=True, exist_ok=True)
|
||||
service.write_text("[Unit]\n")
|
||||
setup_bin = self.target / "usr/bin/omarchy-provision-owner"
|
||||
setup_bin.parent.mkdir(parents=True, exist_ok=True)
|
||||
setup_bin.write_text("#!/bin/bash\n")
|
||||
|
||||
def provisioning_dir(self):
|
||||
return self.target / "var/lib/omarchy/provisioning"
|
||||
|
||||
def test_normal_install_stages_only_the_node_tarball(self):
|
||||
ctx = make_ctx(self.target, defer_provisioning=False)
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
self.assertTrue((self.provisioning_dir() / "packages/node-v24.0.0-linux-x64.tar.gz").exists())
|
||||
self.assertFalse((self.provisioning_dir() / "pending").exists())
|
||||
self.assertFalse((self.target / "etc/systemd/system/omarchy-provision-owner.service").exists())
|
||||
|
||||
def test_deferred_provisioning_against_runtime_without_support_fails_clearly(self):
|
||||
ctx = make_ctx(self.target)
|
||||
with self.assertRaisesRegex(RuntimeError, "does not ship"):
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
def test_deferred_provisioning_without_node_tarball_fails(self):
|
||||
for tarball in self.packages.glob("*.tar.gz"):
|
||||
tarball.unlink()
|
||||
self.install_runtime_provisioning_support()
|
||||
ctx = make_ctx(self.target)
|
||||
with self.assertRaisesRegex(RuntimeError, "Node tarball"):
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
def test_missing_node_tarball_fails_normal_installs_too(self):
|
||||
# The stash is what makes a later factory reset work offline.
|
||||
for tarball in self.packages.glob("*.tar.gz"):
|
||||
tarball.unlink()
|
||||
ctx = make_ctx(self.target, defer_provisioning=False)
|
||||
with self.assertRaisesRegex(RuntimeError, "Node tarball"):
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
def test_deferred_provisioning_arms_first_boot_setup(self):
|
||||
self.install_runtime_provisioning_support()
|
||||
ctx = make_ctx(self.target)
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
self.assertTrue((self.provisioning_dir() / "pending").exists())
|
||||
self.assertTrue((self.target / "etc/systemd/system/omarchy-provision-owner.service").exists())
|
||||
link = self.target / "etc/systemd/system/multi-user.target.wants/omarchy-provision-owner.service"
|
||||
self.assertTrue(link.is_symlink())
|
||||
self.assertEqual(
|
||||
os.readlink(link), "/etc/systemd/system/omarchy-provision-owner.service"
|
||||
)
|
||||
# Unencrypted: no LUKS staging.
|
||||
self.assertFalse((self.provisioning_dir() / "luks-key").exists())
|
||||
|
||||
def test_deferred_provisioning_encrypted_stages_luks_auto_unlock(self):
|
||||
self.install_runtime_provisioning_support()
|
||||
ctx = make_ctx(
|
||||
self.target,
|
||||
user_configuration={"disk_config": {"disk_encryption": {
|
||||
"encryption_type": "luks",
|
||||
"encryption_password": "throwaway-secret",
|
||||
}}},
|
||||
)
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
# Byte-for-byte the slot passphrase — no trailing newline.
|
||||
self.assertEqual((self.provisioning_dir() / "luks-key").read_text(), "throwaway-secret")
|
||||
keyfile = self.target / "etc/omarchy/provisioning.key"
|
||||
self.assertEqual(keyfile.read_text(), "throwaway-secret")
|
||||
self.assertEqual(keyfile.stat().st_mode & 0o777, 0o600)
|
||||
|
||||
cmdline = (self.target / "etc/limine-entry-tool.d/99-omarchy-provisioning-unlock.conf").read_text()
|
||||
self.assertIn("cryptkey=rootfs:/etc/omarchy/provisioning.key", cmdline)
|
||||
files = (self.target / "etc/mkinitcpio.conf.d/99-omarchy-provisioning-key.conf").read_text()
|
||||
self.assertIn("FILES+=(/etc/omarchy/provisioning.key)", files)
|
||||
|
||||
def test_deferred_provisioning_pre_encrypted_without_passphrase_fails(self):
|
||||
self.install_runtime_provisioning_support()
|
||||
ctx = make_ctx(
|
||||
self.target,
|
||||
omarchy_install={"mode": "protected", "defer_provisioning": True, "storage": {"luks_uuid": "abc"}},
|
||||
)
|
||||
with self.assertRaisesRegex(RuntimeError, "passphrase"):
|
||||
phases_impl.stage_provisioning_state(ctx)
|
||||
|
||||
|
||||
class ConfigureLoginDeferProvisioningTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name)
|
||||
self.calls = []
|
||||
|
||||
run_patch = mock.patch.object(
|
||||
phases_impl.subprocess, "run",
|
||||
side_effect=lambda cmd, **kw: (self.calls.append(cmd), CompletedProcess(cmd, 0))[1],
|
||||
)
|
||||
run_patch.start()
|
||||
self.addCleanup(run_patch.stop)
|
||||
|
||||
def test_deferred_provisioning_login_leaves_user_state_to_first_boot(self):
|
||||
ctx = make_ctx(self.target, encrypt=True)
|
||||
phases_impl.configure_login(ctx)
|
||||
|
||||
self.assertTrue((self.target / "etc/sddm.conf.d/99-omarchy-login.conf").exists())
|
||||
self.assertFalse((self.target / "etc/sddm.conf.d/autologin.conf").exists())
|
||||
self.assertFalse((self.target / "var/lib/sddm/state.conf").exists())
|
||||
self.assertTrue(any("sddm.service" in cmd for cmd in self.calls))
|
||||
|
||||
def test_normal_encrypted_login_still_autologs_in(self):
|
||||
ctx = make_ctx(self.target, defer_provisioning=False, encrypt=True, username="jeff")
|
||||
phases_impl.configure_login(ctx)
|
||||
|
||||
autologin = (self.target / "etc/sddm.conf.d/autologin.conf").read_text()
|
||||
self.assertIn("User=jeff", autologin)
|
||||
state = (self.target / "var/lib/sddm/state.conf").read_text()
|
||||
self.assertIn("User=jeff", state)
|
||||
|
||||
|
||||
class ConfigureSshAccessDeferProvisioningTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name) / "mnt"
|
||||
self.target.mkdir()
|
||||
|
||||
info_patch = mock.patch.object(phases_impl, "info")
|
||||
info_patch.start()
|
||||
self.addCleanup(info_patch.stop)
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
self.calls.append(cmd)
|
||||
if "ufw" in cmd:
|
||||
rules = self.target / "etc/ufw/user.rules"
|
||||
rules.parent.mkdir(parents=True, exist_ok=True)
|
||||
rules.write_text("-A ufw-user-input -p tcp --dport 22 -j ACCEPT\n")
|
||||
return CompletedProcess(cmd, 0)
|
||||
|
||||
self.calls = []
|
||||
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=fake_run)
|
||||
run_patch.start()
|
||||
self.addCleanup(run_patch.stop)
|
||||
|
||||
def test_deferred_provisioning_stages_keys_for_first_boot(self):
|
||||
keys = Path(self.tmp.name) / "authorized_keys"
|
||||
keys.write_text("ssh-ed25519 AAAA rig@host\n")
|
||||
ctx = make_ctx(self.target, authorized_keys_path=keys)
|
||||
|
||||
phases_impl.configure_ssh_access(ctx)
|
||||
|
||||
staged = self.target / "var/lib/omarchy/provisioning/authorized_keys"
|
||||
self.assertEqual(staged.read_text(), "ssh-ed25519 AAAA rig@host\n")
|
||||
self.assertEqual(staged.stat().st_mode & 0o777, 0o600)
|
||||
# No user yet — nothing under /home, no chown.
|
||||
self.assertFalse((self.target / "home").exists())
|
||||
self.assertFalse(any("chown" in cmd for cmd in self.calls))
|
||||
# The door is still opened for first boot.
|
||||
self.assertTrue(any("sshd.service" in cmd for cmd in self.calls))
|
||||
|
||||
|
||||
class CreateFactorySnapshotTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.target = Path(self.tmp.name) / "mnt"
|
||||
self.target.mkdir()
|
||||
self.state_dir = Path(self.tmp.name) / "state"
|
||||
self.state_dir.mkdir()
|
||||
|
||||
info_patch = mock.patch.object(phases_impl, "info")
|
||||
info_patch.start()
|
||||
self.addCleanup(info_patch.stop)
|
||||
|
||||
self.findmnt = {
|
||||
"FSTYPE": "btrfs",
|
||||
"OPTIONS": "rw,noatime,compress=zstd:3,subvol=/@",
|
||||
"SOURCE": "/dev/mapper/omarchy_root[/@]",
|
||||
}
|
||||
self.calls = []
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
self.calls.append(cmd)
|
||||
if cmd[0] == "findmnt":
|
||||
return CompletedProcess(cmd, 0, stdout=self.findmnt[cmd[2]] + "\n", stderr="")
|
||||
if cmd[0] == "mount":
|
||||
# Simulate the top level containing @.
|
||||
(Path(cmd[-1]) / "@").mkdir(parents=True, exist_ok=True)
|
||||
return CompletedProcess(cmd, 0, stdout="", stderr="")
|
||||
|
||||
run_patch = mock.patch.object(phases_impl.subprocess, "run", side_effect=fake_run)
|
||||
run_patch.start()
|
||||
self.addCleanup(run_patch.stop)
|
||||
|
||||
def ctx(self):
|
||||
return make_ctx(self.target, defer_provisioning=False, state_dir=self.state_dir)
|
||||
|
||||
def test_snapshots_root_as_factory(self):
|
||||
phases_impl.create_factory_snapshot(self.ctx())
|
||||
|
||||
top = str(self.state_dir / "factory-top")
|
||||
self.assertIn(["mount", "-o", "subvolid=5", "/dev/mapper/omarchy_root", top], self.calls)
|
||||
self.assertIn(
|
||||
["btrfs", "subvolume", "snapshot", f"{top}/@", f"{top}/@factory"],
|
||||
self.calls,
|
||||
)
|
||||
self.assertIn(
|
||||
["btrfs", "property", "set", "-ts", f"{top}/@factory", "ro", "true"],
|
||||
self.calls,
|
||||
)
|
||||
self.assertIn(["umount", top], self.calls)
|
||||
|
||||
def test_factory_snapshot_scrubs_provisioning_credentials(self):
|
||||
# Simulate the snapshot carrying deployment credentials; the scrub
|
||||
# must drop them before the snapshot goes read-only.
|
||||
top = self.state_dir / "factory-top"
|
||||
factory = top / "@factory"
|
||||
secrets = [
|
||||
factory / "var/lib/omarchy/provisioning/authorized_keys",
|
||||
factory / "etc/tailscale/authkey",
|
||||
factory / "etc/omarchy/provisioning.key",
|
||||
]
|
||||
keep = factory / "var/lib/omarchy/provisioning/groups"
|
||||
for path in [*secrets, keep]:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("secret")
|
||||
|
||||
phases_impl.create_factory_snapshot(self.ctx())
|
||||
|
||||
for path in secrets:
|
||||
self.assertFalse(path.exists(), path)
|
||||
self.assertTrue(keep.exists())
|
||||
|
||||
def test_non_btrfs_target_skips(self):
|
||||
self.findmnt["FSTYPE"] = "ext4"
|
||||
phases_impl.create_factory_snapshot(self.ctx())
|
||||
self.assertFalse(any(cmd[0] == "mount" for cmd in self.calls))
|
||||
|
||||
def test_non_subvol_root_skips(self):
|
||||
self.findmnt["OPTIONS"] = "rw,noatime"
|
||||
phases_impl.create_factory_snapshot(self.ctx())
|
||||
self.assertFalse(any(cmd[0] == "mount" for cmd in self.calls))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user