diff --git a/test/all b/test/all new file mode 100755 index 0000000..fd91768 --- /dev/null +++ b/test/all @@ -0,0 +1,13 @@ +#!/bin/bash + +set -euo pipefail + +ROOT=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd) + +for test in "$ROOT"/test/*-test.sh; do + printf '==> %s\n' "${test#$ROOT/}" + bash "$test" +done + +printf '==> test/test_*.py\n' +python3 -m unittest discover -t "$ROOT" -s "$ROOT/test" diff --git a/test/cidata-load-test.sh b/test/cidata-load-test.sh new file mode 100644 index 0000000..ede60fa --- /dev/null +++ b/test/cidata-load-test.sh @@ -0,0 +1,170 @@ +#!/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/ssh.json" +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 ssh.json; 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/ssh.json ]] || 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/ssh.json" +run_load || fail "required pair plus ssh.json loads" +[[ -f $sandbox/root/ssh.json ]] || fail "ssh.json is copied when present" +[[ ! -e $sandbox/root/user_full_name.txt ]] || fail "absent optional files are not copied" +pass "present optional files are copied, absent ones skipped" + +# 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/ssh.json" +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/ssh.json ]] || fail "partial copy leaves no ssh.json behind" +pass "partial copy cleans up what it copied" diff --git a/test/test_configure_ssh_access.py b/test/test_configure_ssh_access.py new file mode 100644 index 0000000..71159af --- /dev/null +++ b/test/test_configure_ssh_access.py @@ -0,0 +1,157 @@ +"""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 json +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().parent.parent / "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, ssh_keys=None, ssh_keys_text=None): + ssh_keys_path = None + if ssh_keys is not None or ssh_keys_text is not None: + ssh_keys_path = self.target / "ssh.json" + ssh_keys_path.write_text(ssh_keys_text if ssh_keys_text is not None else json.dumps(ssh_keys)) + return types.SimpleNamespace(target=self.target, username="jeff", ssh_keys_path=ssh_keys_path) + + 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_ssh_json_is_a_no_op(self): + self.configure() + self.assertEqual(self.calls, []) + self.assertFalse((self.target / "home" / "jeff" / ".ssh").exists()) + + def test_writes_keys_one_per_line(self): + self.configure(ssh_keys=["ssh-ed25519 AAAA jeff@host", "ssh-rsa BBBB jeff@work"]) + self.assertEqual(self.authorized_keys().read_text(), "ssh-ed25519 AAAA jeff@host\nssh-rsa BBBB jeff@work\n") + + def test_strips_whitespace_and_drops_blank_entries(self): + self.configure(ssh_keys=[" ssh-ed25519 AAAA jeff@host\n", "", " "]) + self.assertEqual(self.authorized_keys().read_text(), "ssh-ed25519 AAAA jeff@host\n") + + def test_ssh_dir_is_private(self): + self.configure(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + 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(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + self.assertEqual(stat.S_IMODE(self.authorized_keys().stat().st_mode), 0o600) + + def test_chowns_ssh_dir_to_the_user(self): + self.configure(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + self.assertEqual(self.chrooted("chown"), [ + ["arch-chroot", str(self.target), "chown", "-R", "jeff:jeff", "/home/jeff/.ssh"], + ]) + + def test_enables_sshd(self): + self.configure(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + 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(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + 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(ssh_keys=["ssh-ed25519 AAAA jeff@host"]) + + def test_malformed_json_fails_the_phase(self): + with self.assertRaisesRegex(RuntimeError, "not readable JSON"): + self.configure(ssh_keys_text="not json") + + def test_non_array_json_fails_the_phase(self): + with self.assertRaisesRegex(RuntimeError, "JSON array of public key strings"): + self.configure(ssh_keys={"key": "ssh-ed25519 AAAA"}) + + def test_non_string_entries_fail_the_phase(self): + with self.assertRaisesRegex(RuntimeError, "JSON array of public key strings"): + self.configure(ssh_keys=["ssh-ed25519 AAAA jeff@host", 42]) + + def test_empty_key_list_fails_the_phase(self): + with self.assertRaisesRegex(RuntimeError, "contains no SSH keys"): + self.configure(ssh_keys=[]) + + def test_blank_only_key_list_fails_the_phase(self): + with self.assertRaisesRegex(RuntimeError, "contains no SSH keys"): + self.configure(ssh_keys=["", " "]) + + +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/ssh.json")) + + 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()