Files
omarchy-iso/test/unit/test_region.py
T
ZacharyZhang-NY 02ee068504 Merge omacom/omarchy-iso#236 regional ISO builds, as an install-time choice
Every ISO bundles the cn profile; the first-screen pick decides. --region cn
only preselects China and names the artifact. The region value is cn, the cn
timezone is preselected when China is chosen, and the profile packages and
ArchLinuxCN keyring reach the target only for China.
2026-10-07 11:32:14 -04:00

430 lines
22 KiB
Python

"""Region: the install-time choice, the bundled cn profile, and regional builds."""
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest import mock
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "configs/airootfs/usr/share/omarchy-iso"))
sys.modules.setdefault(
"orchestrator.archinstall_adapter",
types.ModuleType("orchestrator.archinstall_adapter"),
)
from orchestrator import context, phases_impl # noqa: E402
def load_context(root: Path, omarchy_install: dict | None) -> context.InstallContext:
config = {"kernels": ["linux-omarchy"]}
if omarchy_install is not None:
config["omarchy_install"] = omarchy_install
config_path = root / "config.json"
creds_path = root / "creds.json"
config_path.write_text(json.dumps(config))
creds_path.write_text(json.dumps({"users": [{"username": "test"}]}))
with mock.patch.dict(os.environ, {
"OMARCHY_INSTALL_CONFIG": str(config_path),
"OMARCHY_INSTALL_CREDS": str(creds_path),
"OMARCHY_INSTALL_STATE_DIR": str(root / "state"),
}, clear=True):
return context.InstallContext.from_env()
class RegionTest(unittest.TestCase):
def test_context_reads_region(self):
for omarchy_install, expected in [
(None, "global"),
({"mode": "full_disk"}, "global"),
({"mode": "full_disk", "region": "global"}, "global"),
({"mode": "full_disk", "region": "cn"}, "cn"),
]:
with self.subTest(omarchy_install=omarchy_install), tempfile.TemporaryDirectory() as tmp:
ctx = load_context(Path(tmp), omarchy_install)
self.assertEqual(ctx.omarchy_install["region"], expected)
def test_context_rejects_unknown_region(self):
for region in ("China", "mars", ""):
with self.subTest(region=region), tempfile.TemporaryDirectory() as tmp:
with self.assertRaisesRegex(RuntimeError, "omarchy_install.region must be global or cn"):
load_context(Path(tmp), {"mode": "full_disk", "region": region})
def test_target_setup_receives_region(self):
for region in ("global", "cn"):
with self.subTest(region=region), tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
ctx = types.SimpleNamespace(
target=root / "mnt",
log_path=root / "install.log",
omarchy_install={"region": region},
username="test",
full_name="",
email="",
state={"omarchy_start_time": "now", "omarchy_start_epoch": 0,
"omarchy_finalizer_header_written": True},
)
with mock.patch.object(phases_impl, "_prepare_target_setup"), \
mock.patch.object(phases_impl.subprocess, "run") as run:
phases_impl._run_target_setup_command(ctx, ["/usr/bin/omarchy-apply-system"])
chroot = next(call.args[0] for call in run.call_args_list if call.args[0][0] == "arch-chroot")
self.assertIn(f"OMARCHY_REGION={region}", chroot)
def run_keyboard_form(self, script: str, preselect: str = "global", **env: str) -> subprocess.CompletedProcess:
configurator = (ROOT / "configs/airootfs/root/configurator").read_text()
language = re.search(r"^apply_region_language\(\) \{\n.*?^\}", configurator, re.M | re.S)
form = re.search(r"^keyboard_form\(\) \{\n.*?^\}", configurator, re.M | re.S)
init = re.search(r'^region="\$\{OMARCHY_INSTALL_REGION:-.*\}"$', configurator, re.M)
self.assertIsNotNone(language)
self.assertIsNotNone(form)
self.assertIsNotNone(init)
harness = "\n".join([
"set -euo pipefail",
"OMARCHY_FORM_BACK=1; OMARCHY_FORM_SIGNAL=130; defer_provisioning=false",
"clear_logo() { :; }; say() { :; }; abort() { exit 99; }; tty() { echo /dev/pts/0; }",
"omarchy_text() { printf '%s' \"$1\"; }",
"answers=(" + script + "); step=0",
"answer() { local a=${answers[step]}; step=$((step + 1)); [[ $a == ok:* ]] && eval \"${a#ok:}\" && return 0; return ${a#rc:}; }",
"omarchy_prompt_region() { echo \"region-preselect=$region\"; answer; }",
"omarchy_prompt_keyboard() { echo keyboard-prompt; answer; }",
"confirm_prepare_for_another_owner() { answer; }",
init.group(),
language.group(),
form.group(),
"keyboard_form",
'echo "region=$region defer=$defer_provisioning keyboard=${keyboard:-} lang=${OMARCHY_INSTALL_LANG:-}"',
])
with tempfile.TemporaryDirectory() as tmp:
iso_region = Path(tmp) / "omarchy_region"
iso_region.write_text(preselect + "\n")
return subprocess.run(["bash", "-c", harness], capture_output=True, text=True,
env={"PATH": os.environ["PATH"], "ISO_REGION_FILE": str(iso_region), **env})
def lines(self, result: subprocess.CompletedProcess) -> list[str]:
return [line for line in result.stdout.splitlines() if line]
def test_ctrl_c_before_choosing_defers_as_global(self):
out = self.run_keyboard_form("rc:130 ok:true")
self.assertIn("region=global defer=true", out.stdout)
def test_back_from_keyboard_keeps_the_chosen_region(self):
out = self.run_keyboard_form("ok:region=cn rc:1 ok:true ok:keyboard=us")
self.assertEqual(self.lines(out), [
"region-preselect=global",
"keyboard-prompt",
"region-preselect=cn",
"keyboard-prompt",
"region=cn defer=false keyboard=us lang=",
])
def test_a_cn_iso_preselects_china(self):
out = self.run_keyboard_form("ok:true", preselect="cn", OMARCHY_CJK_RELAUNCH="10")
self.assertEqual(out.returncode, 10)
self.assertEqual(self.lines(out), ["region-preselect=cn"])
def test_china_on_the_vt_hands_over_to_the_cjk_console(self):
out = self.run_keyboard_form("ok:region=cn", OMARCHY_CJK_RELAUNCH="10")
self.assertEqual(out.returncode, 10)
self.assertEqual(self.lines(out), ["region-preselect=global"])
def test_cjk_console_reads_chinese_with_a_us_keyboard(self):
out = self.run_keyboard_form("ok:true", OMARCHY_CJK_CONSOLE="1", OMARCHY_VT_RETURN="11",
OMARCHY_INSTALL_REGION="cn")
self.assertEqual(out.returncode, 0)
self.assertEqual(self.lines(out), [
"region-preselect=cn",
"region=cn defer=false keyboard=us lang=zh",
])
def test_global_in_the_cjk_console_returns_to_the_vt(self):
out = self.run_keyboard_form("ok:region=global", OMARCHY_CJK_CONSOLE="1", OMARCHY_VT_RETURN="11",
OMARCHY_INSTALL_REGION="cn")
self.assertEqual(out.returncode, 11)
self.assertNotIn("keyboard-prompt", out.stdout)
class RegionTargetTest(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
self.target = self.root / "target"
self.payload = self.root / "payload"
patch = mock.patch.object(phases_impl, "ISO_REGION_PAYLOAD", self.payload)
patch.start()
self.addCleanup(patch.stop)
def ctx(self, region, deferred=False):
return types.SimpleNamespace(target=self.target, username="alice", defer_provisioning=deferred,
omarchy_install={"region": region})
def test_global_does_not_change_user_defaults(self):
(self.payload / "skel").mkdir(parents=True)
(self.payload / "skel/.regional").write_text("cn only\n")
phases_impl._stage_region_defaults(self.ctx("global"))
self.assertEqual((self.target / "etc/omarchy/region").read_text(), "global\n")
self.assertFalse((self.target / "etc/skel").exists())
def test_cn_seeds_skel_without_touching_existing_users_or_locale(self):
profile = self.payload / "skel/.config/example/defaults"
profile.parent.mkdir(parents=True)
profile.write_text("regional defaults\n")
existing = self.target / "home/alice/.config/example/defaults"
existing.parent.mkdir(parents=True)
existing.write_text("personal settings\n")
skel = self.target / "etc/skel"
skel.mkdir(parents=True)
(skel / ".bashrc").write_text("original\n")
(self.target / "etc/locale.conf").write_text("LANG=en_US.UTF-8\n")
phases_impl._stage_region_defaults(self.ctx("cn"))
self.assertEqual((skel / ".config/example/defaults").read_text(), "regional defaults\n")
self.assertEqual((skel / ".bashrc").read_text(), "original\n")
self.assertEqual(existing.read_text(), "personal settings\n")
self.assertEqual((self.target / "etc/locale.conf").read_text(), "LANG=en_US.UTF-8\n")
self.assertEqual((self.target / "etc/omarchy/region").read_text(), "cn\n")
def test_cn_without_user_defaults_only_records_the_region(self):
phases_impl._stage_region_defaults(self.ctx("cn"))
self.assertEqual((self.target / "etc/omarchy/region").read_text(), "cn\n")
self.assertFalse((self.target / "etc/skel").exists())
def test_cn_keyring_is_initialized_before_finalization_for_both_install_modes(self):
for deferred in (False, True):
ctx = self.ctx("cn", deferred)
with mock.patch.object(phases_impl, "_run_target_setup_command") as run, \
mock.patch.object(phases_impl, "_mask_mkinitcpio_pacman_hooks"), \
mock.patch.object(phases_impl, "_unmask_mkinitcpio_pacman_hooks"):
phases_impl.run_system_finalizer(ctx)
commands = [call.args[1] for call in run.call_args_list]
self.assertEqual(commands[:2], [["pacman-key", "--init"], ["pacman-key", "--populate", "archlinux", "archlinuxcn"]])
expected = ["--defer-provisioning"] if deferred else ["--install-user", "alice"]
self.assertEqual(commands[2], ["/usr/bin/omarchy-apply-system", *expected, "--first-install"])
def test_keyring_failure_prevents_enabling_online_repositories(self):
ctx = self.ctx("cn")
with mock.patch.object(phases_impl, "_run_target_setup_command", side_effect=RuntimeError("keyring failed")) as run:
with self.assertRaisesRegex(RuntimeError, "keyring failed"):
phases_impl.run_system_finalizer(ctx)
self.assertEqual(run.call_count, 1)
def test_global_trusts_no_regional_keyring(self):
with mock.patch.object(phases_impl, "_run_target_setup_command") as run, \
mock.patch.object(phases_impl, "_mask_mkinitcpio_pacman_hooks"), \
mock.patch.object(phases_impl, "_unmask_mkinitcpio_pacman_hooks"):
phases_impl.run_system_finalizer(self.ctx("global"))
self.assertNotIn("pacman-key", [call.args[1][0] for call in run.call_args_list])
class RegionBuildTest(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
self.runtime = self.root / "runtime"
self.profile = self.runtime / "default/regions/cn"
(self.profile / "pacman").mkdir(parents=True)
(self.runtime / "bin").mkdir()
(self.runtime / "bin/omarchy-apply-pacman").touch()
(self.profile / "packages").write_text("archlinuxcn-keyring\nexample-regional-package\n")
(self.profile / "pacman/pacman.conf.append").write_text("[archlinuxcn]\nServer = https://mirrors.ustc.edu.cn/archlinuxcn/$arch\n")
(self.profile / "pacman/mirrorlist.append").write_text("Server = https://mirrors.ustc.edu.cn/archlinux/$repo/os/$arch\n")
(self.profile / "timezone").write_text("Asia/Shanghai\n")
self.iso = self.root / "iso"
self.payload = self.iso / "usr/share/omarchy-iso"
self.payload.mkdir(parents=True)
self.packages = self.payload / "omarchy-base.packages"
self.packages.write_text("base\n")
self.other = self.payload / "omarchy-other.packages"
self.other.write_text("other\n")
self.config = self.root / "pacman.conf"
shutil.copyfile(ROOT / "configs/pacman-online-stable.conf", self.config)
self.original_config = self.config.read_bytes()
self.bin = self.root / "bin"
self.bin.mkdir()
self.log = self.root / "commands"
for command in ("pacman", "pacman-key"):
stub = self.bin / command
stub.write_text('#!/bin/bash\nprintf "%s %s\\n" "${0##*/}" "$*" >> "$TEST_LOG"\nif [[ ${0##*/} == pacman ]]; then exit "${TEST_PACMAN_STATUS:-0}"; fi\n')
stub.chmod(0o755)
def prepare(self, region, **env):
return subprocess.run(
["bash", str(ROOT / "builder/prepare-region.sh"), region, str(self.runtime), str(self.iso), str(self.config)],
env={**os.environ, "PATH": f"{self.bin}:{os.environ['PATH']}", "TEST_LOG": str(self.log), **env},
capture_output=True, text=True,
)
def test_cn_bundles_its_packages_installs_them_only_for_china_and_verifies_keyring_first(self):
result = self.prepare("cn")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.packages.read_text(), "base\n")
self.assertEqual(self.other.read_text(), "other\n\narchlinuxcn-keyring\nexample-regional-package\n")
self.assertEqual(self.config.read_bytes(), self.original_config + b"\n[archlinuxcn]\nServer = https://mirrors.ustc.edu.cn/archlinuxcn/$arch\n")
self.assertEqual((self.payload / "region/packages").read_text(), "archlinuxcn-keyring\nexample-regional-package\n")
self.assertEqual((self.payload / "region/timezone").read_text(), "Asia/Shanghai\n")
self.assertEqual(self.log.read_text().splitlines(), [f"pacman --config {self.config} --noconfirm -Sy --needed archlinuxcn-keyring", "pacman-key --populate archlinuxcn"])
with mock.patch.object(phases_impl, "Path", return_value=self.packages), \
mock.patch.object(phases_impl, "ISO_REGION_PAYLOAD", self.payload / "region"), \
mock.patch.object(phases_impl, "_package_targets", return_value={"runtime": "omarchy-dev", "settings": "omarchy-settings-dev", "nvim": "omarchy-nvim"}):
for region, extra in (("global", []), ("cn", ["archlinuxcn-keyring", "example-regional-package"])):
ctx = types.SimpleNamespace(omarchy_install={"region": region})
self.assertEqual(phases_impl._runtime_package_list(ctx), ["omarchy-dev", "base", *extra])
def test_global_is_unchanged_without_region_support_in_runtime(self):
shutil.rmtree(self.runtime)
result = self.prepare("global")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(self.other.read_text(), "other\n")
self.assertEqual(self.config.read_bytes(), self.original_config)
self.assertFalse(self.log.exists())
self.assertFalse((self.payload / "region").exists())
def test_missing_runtime_support_fails_before_package_operations(self):
(self.runtime / "bin/omarchy-apply-pacman").unlink()
result = self.prepare("cn")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--local-source", result.stderr)
self.assertFalse(self.log.exists())
self.assertEqual(self.config.read_bytes(), self.original_config)
def test_profile_without_user_defaults_is_supported(self):
result = self.prepare("cn")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertFalse((self.payload / "region/skel").exists())
def test_untrusted_keyring_aborts_without_importing_keys(self):
result = self.prepare("cn", TEST_PACMAN_STATUS="1")
self.assertNotEqual(result.returncode, 0)
self.assertNotIn("pacman-key", self.log.read_text())
class RegionMakeTest(unittest.TestCase):
def setUp(self):
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.root = Path(tmp.name)
(self.root / "bin").mkdir()
shutil.copyfile(ROOT / "bin/omarchy-iso-make", self.root / "bin/omarchy-iso-make")
for directory in ("configs/airootfs/root", "configs/airootfs/usr/local/bin", "stubs"):
(self.root / directory).mkdir(parents=True)
(self.root / "configs/profiledef.sh").touch()
self.log = self.root / "docker-args"
stubs = {
"git": '#!/bin/bash\nexit 0\n',
"sudo": '#!/bin/bash\necho "unexpected sudo" >&2\nexit 99\n',
"docker": '''#!/bin/bash
if [[ $1 == version ]]; then exit 0; fi
printf '%s\\n' "$@" > "$TEST_LOG"
for arg in "$@"; do
if [[ $arg == *:/out/ ]]; then touch "${arg%:/out/}/omarchy-test-x86_64.iso"; fi
done
''',
}
for name, content in stubs.items():
stub = self.root / "stubs" / name
stub.write_text(content)
stub.chmod(0o755)
def make(self, *args):
env = {key: value for key, value in os.environ.items() if not key.startswith("OMARCHY_")}
return subprocess.run(
["bash", "bin/omarchy-iso-make", "--keep-pkg-cache", "--no-boot-offer", *args],
cwd=self.root, env={**env, "HOME": str(self.root), "PATH": f"{self.root / 'stubs'}:{env['PATH']}", "TEST_LOG": str(self.log), "TEST_ROOT": str(self.root)},
capture_output=True, text=True,
)
def test_region_is_independent_of_channel_and_names_cache_and_artifact(self):
for flags, channel, ref in (([], "stable", "quattro"), (["--rc"], "rc", "rc"), (["--edge"], "edge", "edge"), (["--dev"], "edge", "dev")):
with self.subTest(channel=ref):
release = self.root / "release"
if release.exists():
shutil.rmtree(release)
result = self.make(*flags, "--region", "cn")
self.assertEqual(result.returncode, 0, result.stderr)
args = self.log.read_text().splitlines()
self.assertIn("OMARCHY_REGION=cn", args)
self.assertIn(f"OMARCHY_MIRROR={channel}", args)
self.assertIn(f"OMARCHY_ISO_REF={ref}", args)
self.assertIn(f"{self.root}/.cache/omarchy/iso_{channel}-cn/airootfs/var/cache/omarchy:/var/cache/airootfs/var/cache/omarchy", args)
self.assertTrue((release / f"omarchy-test-x86_64-{ref}-cn.iso").exists())
def test_default_preserves_existing_global_cache_and_name(self):
result = self.make()
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("OMARCHY_REGION=global", self.log.read_text().splitlines())
self.assertIn("/iso_stable/", self.log.read_text())
self.assertTrue((self.root / "release/omarchy-test-x86_64-quattro.iso").exists())
def test_build_renames_only_its_own_iso(self):
release = self.root / "release"
release.mkdir()
other = release / "omarchy-other-x86_64.iso"
other.touch()
result = self.make("--region", "cn")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue(other.exists())
self.assertTrue((release / "omarchy-test-x86_64-quattro-cn.iso").exists())
self.assertEqual(list(release.glob(".build-*")), [])
def test_local_source_keeps_region_and_mounts_both_checkouts(self):
for name in ("runtime", "packages"):
(self.root / name).mkdir()
result = self.make("--region", "cn", "--local-source", str(self.root / "runtime"), str(self.root / "packages"))
self.assertEqual(result.returncode, 0, result.stderr)
args = self.log.read_text().splitlines()
self.assertIn("OMARCHY_REGION=cn", args)
self.assertIn("OMARCHY_ISO_REF=local", args)
self.assertIn(f"{self.root}/runtime:/omarchy-source:ro", args)
self.assertIn(f"{self.root}/packages:/omarchy-pkgs:ro", args)
self.assertTrue((self.root / "release/omarchy-test-x86_64-local-cn.iso").exists())
def test_invalid_regions_fail_without_starting_build(self):
for flags in (("--region",), ("--region", "chn"), ("--region", "CN"), ("--region", "zz"), ("--region", "../cn")):
with self.subTest(flags=flags):
result = self.make(*flags)
self.assertNotEqual(result.returncode, 0)
self.assertFalse(self.log.exists())
self.assertFalse((self.root / "release").exists())
class ConfiguratorRegionTimezoneTest(unittest.TestCase):
"""The configurator hands the cn timezone to the setup form when China is chosen."""
def setUp(self):
configurator = (ROOT / "configs/airootfs/root/configurator").read_text()
helper = re.search(r"^region_timezone\(\) \{\n.*?^\}", configurator, re.M | re.S)
self.assertIsNotNone(helper)
self.helper = helper.group(0)
self.configurator = configurator
tmp = tempfile.TemporaryDirectory()
self.addCleanup(tmp.cleanup)
self.file = Path(tmp.name) / "timezone"
def region_timezone(self, region):
self.file.write_text("Asia/Shanghai\n")
return subprocess.run(
["bash", "-c", f"set -euo pipefail\nregion={region}\n{self.helper}\nregion_timezone"],
env={**os.environ, "REGION_TIMEZONE": str(self.file)},
capture_output=True, text=True, check=True,
).stdout
def test_global_offers_no_timezone(self):
self.assertEqual(self.region_timezone("global"), "")
def test_china_offers_the_profile_timezone(self):
self.assertEqual(self.region_timezone("cn"), "Asia/Shanghai\n")
def test_user_form_passes_it_to_the_timezone_prompt(self):
self.assertIn('omarchy_prompt_timezone "$(region_timezone)"', self.configurator)
if __name__ == "__main__":
unittest.main()