405 lines
15 KiB
Python
405 lines
15 KiB
Python
from app import (
|
|
build_archinstall_config,
|
|
build_archinstall_creds,
|
|
validate_step1,
|
|
)
|
|
|
|
|
|
def test_validate_step1_accepts_good_input():
|
|
errors, values = validate_step1(
|
|
{
|
|
"hostname": "furtka",
|
|
"username": "daniel",
|
|
"password": "topsecretpw",
|
|
"password2": "topsecretpw",
|
|
"language": "de",
|
|
}
|
|
)
|
|
assert errors == []
|
|
assert values == {
|
|
"hostname": "furtka",
|
|
"username": "daniel",
|
|
"password": "topsecretpw",
|
|
"language": "de",
|
|
}
|
|
|
|
|
|
def test_validate_step1_collects_all_errors():
|
|
errors, _ = validate_step1(
|
|
{
|
|
"hostname": "BAD!",
|
|
"username": "1bad",
|
|
"password": "short",
|
|
"password2": "mismatch",
|
|
"language": "xx",
|
|
}
|
|
)
|
|
assert len(errors) == 5
|
|
|
|
|
|
def test_build_archinstall_config_uses_selected_locale(monkeypatch):
|
|
# build_disk_config imports archinstall lazily; archinstall isn't
|
|
# installed in CI (only runs on the live ISO), so stub it out.
|
|
import app as app_module
|
|
|
|
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"stubbed_device": d})
|
|
|
|
cfg = build_archinstall_config(
|
|
{
|
|
"hostname": "h",
|
|
"username": "u",
|
|
"password": "pw12345678",
|
|
"language": "pl",
|
|
"boot_drive": "/dev/sda",
|
|
}
|
|
)
|
|
assert cfg["disk_config"] == {"stubbed_device": "/dev/sda"}
|
|
assert cfg["hostname"] == "h"
|
|
assert cfg["locale_config"]["locale"] == "pl_PL.UTF-8"
|
|
# Users moved out of config into creds once we adopted archinstall 4.x's
|
|
# `!password` sentinel; config only carries a gpasswd in custom_commands
|
|
# so the user lands in the docker group after docker is pacstrapped.
|
|
assert "users" not in cfg
|
|
assert cfg["custom_commands"][0] == "gpasswd -a u docker"
|
|
|
|
|
|
def test_build_archinstall_config_includes_post_install_bootstrap(monkeypatch, tmp_path):
|
|
# The installed system should come up with a Furtka landing page at
|
|
# http://<hostname>.local. That means caddy + avahi pacstrapped, the
|
|
# matching services enabled, a Caddyfile written into the target rootfs,
|
|
# nss-mdns spliced into nsswitch.conf, and the resource-manager tarball
|
|
# extracted into the versioned /opt/furtka/current layout.
|
|
import app as app_module
|
|
|
|
# Fake payload so _resource_manager_commands emits its full cmd tree.
|
|
fake_payload = tmp_path / "payload.tar.gz"
|
|
fake_payload.write_bytes(b"\x1f\x8b\x08\x00fake")
|
|
monkeypatch.setattr(app_module, "RESOURCE_MANAGER_PAYLOAD", fake_payload)
|
|
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"stubbed_device": d})
|
|
|
|
cfg = build_archinstall_config(
|
|
{
|
|
"hostname": "heimserver",
|
|
"username": "u",
|
|
"password": "pw12345678",
|
|
"language": "en",
|
|
"boot_drive": "/dev/sda",
|
|
}
|
|
)
|
|
|
|
for pkg in ("caddy", "avahi", "nss-mdns", "wireguard-tools"):
|
|
assert pkg in cfg["packages"]
|
|
# Packaged units go in `services` (enabled before custom_commands runs);
|
|
# our own units don't exist at that point, so they must be enabled from
|
|
# within custom_commands after the unit files land on disk.
|
|
for svc in ("caddy", "avahi-daemon"):
|
|
assert svc in cfg["services"]
|
|
assert "furtka-welcome" not in cfg["services"]
|
|
assert "furtka-status.timer" not in cfg["services"]
|
|
|
|
joined = "\n".join(cfg["custom_commands"])
|
|
# Every furtka-* unit is systemctl-linked from the shipped asset tree and
|
|
# then enabled — no hand-written files under /etc/systemd/system/.
|
|
assert "systemctl link /opt/furtka/current/assets/systemd/" in joined
|
|
assert "systemctl enable furtka-api.service" in joined
|
|
for path in (
|
|
"/etc/caddy/Caddyfile",
|
|
"/var/lib/furtka/status.json",
|
|
"/var/lib/furtka/furtka.json",
|
|
):
|
|
assert path in joined, f"expected {path} to be written by custom_commands"
|
|
# /srv/furtka/www/ is retired — no writes should target it anymore.
|
|
assert "/srv/furtka/www" not in joined
|
|
|
|
assert "mdns_minimal" in joined
|
|
assert "nsswitch.conf" in joined
|
|
# Hostname is injected via /var/lib/furtka/furtka.json, not via sed.
|
|
assert "__HOSTNAME__" not in joined
|
|
|
|
|
|
def test_resource_manager_payload_landed_when_present(monkeypatch, tmp_path):
|
|
# When iso/build.sh has staged the resource-manager tarball, the
|
|
# post-install commands should untar it into /opt/furtka/versions/<ver>/,
|
|
# flip the /opt/furtka/current symlink, drop the `furtka` wrapper, and
|
|
# systemctl-link every Furtka unit file.
|
|
import app as app_module
|
|
|
|
fake_payload = tmp_path / "furtka-resource-manager.tar.gz"
|
|
fake_payload.write_bytes(b"\x1f\x8b\x08\x00fake-tarball-bytes")
|
|
monkeypatch.setattr(app_module, "RESOURCE_MANAGER_PAYLOAD", fake_payload)
|
|
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"stubbed_device": d})
|
|
|
|
cfg = build_archinstall_config(
|
|
{
|
|
"hostname": "heimserver",
|
|
"username": "u",
|
|
"password": "pw12345678",
|
|
"language": "en",
|
|
"boot_drive": "/dev/sda",
|
|
}
|
|
)
|
|
joined = "\n".join(cfg["custom_commands"])
|
|
|
|
# Tarball lands in the versioned slot, not flat /opt/furtka/.
|
|
assert 'tar -xzf - -C "$staging"' in joined
|
|
assert "/opt/furtka/versions/$ver" in joined
|
|
assert 'ln -sfn "/opt/furtka/versions/$ver" /opt/furtka/current' in joined
|
|
# CLI wrapper lands and references /opt/furtka/current. The wrapper body
|
|
# rides the base64 payload, so assert on the constant directly.
|
|
assert "/usr/local/bin/furtka" in joined
|
|
assert "PYTHONPATH=/opt/furtka/current" in app_module._FURTKA_WRAPPER_SH
|
|
# Every unit is linked + enabled.
|
|
for unit in app_module._FURTKA_UNITS:
|
|
assert f"/opt/furtka/current/assets/systemd/{unit}" in joined
|
|
assert unit in joined
|
|
# python is pacstrapped so the wrapper has an interpreter.
|
|
assert "python" in cfg["packages"]
|
|
|
|
|
|
def test_resource_manager_absent_without_payload(monkeypatch, tmp_path):
|
|
# Dev box / CI without an ISO build: payload doesn't exist. No tarball
|
|
# extract, no unit link, no enable — and no stale references to furtka-*
|
|
# units in custom_commands.
|
|
import app as app_module
|
|
|
|
monkeypatch.setattr(app_module, "RESOURCE_MANAGER_PAYLOAD", tmp_path / "does-not-exist.tar.gz")
|
|
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"stubbed_device": d})
|
|
|
|
cfg = build_archinstall_config(
|
|
{
|
|
"hostname": "heimserver",
|
|
"username": "u",
|
|
"password": "pw12345678",
|
|
"language": "en",
|
|
"boot_drive": "/dev/sda",
|
|
}
|
|
)
|
|
joined = "\n".join(cfg["custom_commands"])
|
|
|
|
assert "tar -xzf" not in joined
|
|
assert "systemctl link" not in joined
|
|
# The base system bootstrap (caddy, status.json placeholder, furtka.json)
|
|
# is unaffected.
|
|
assert "/etc/caddy/Caddyfile" in joined
|
|
assert "/var/lib/furtka/furtka.json" in joined
|
|
|
|
|
|
def test_build_archinstall_creds_uses_archinstall_sentinel_keys():
|
|
creds = build_archinstall_creds({"username": "u", "password": "pw12345678"})
|
|
assert creds["!root-password"] == "pw12345678"
|
|
assert creds["users"] == [
|
|
{
|
|
"username": "u",
|
|
"!password": "pw12345678",
|
|
"sudo": True,
|
|
"groups": [],
|
|
}
|
|
]
|
|
|
|
|
|
def test_log_page_done_state_shows_usb_step_before_restart(monkeypatch, tmp_path):
|
|
"""The 'pull the USB stick' step must be visible *before* the restart
|
|
button — after the click the box reboots within seconds (first hardware
|
|
bench booted the stick again)."""
|
|
import app as app_module
|
|
|
|
log_path = tmp_path / "install.log"
|
|
log_path.write_text("Installation completed without any errors.\n")
|
|
monkeypatch.setattr(app_module, "INSTALL_LOG", log_path)
|
|
app_module.settings["hostname"] = "bench"
|
|
|
|
html = app_module.app.test_client().get("/install/log").get_data(as_text=True)
|
|
|
|
assert '<ol class="next-steps">' in html
|
|
assert "http://bench.local" in html
|
|
assert "USB stick is out" in html
|
|
assert html.index("next-steps") < html.index("USB stick is out")
|
|
|
|
|
|
def test_log_page_running_state_has_no_restart_button(monkeypatch, tmp_path):
|
|
import app as app_module
|
|
|
|
log_path = tmp_path / "install.log"
|
|
log_path.write_text("Creating partitions: /dev/sda\n")
|
|
monkeypatch.setattr(app_module, "INSTALL_LOG", log_path)
|
|
|
|
html = app_module.app.test_client().get("/install/log").get_data(as_text=True)
|
|
assert "next-steps" not in html
|
|
assert "install_reboot" not in html and "restart now" not in html
|
|
|
|
|
|
def test_nonzero_exit_without_traceback_is_an_error():
|
|
import app as app_module
|
|
|
|
# OOM kill / signal: archinstall dies mid-pacstrap and prints nothing.
|
|
log = "Installing packages: ['base'\n"
|
|
assert app_module.parse_install_progress(log)["status"] == "running"
|
|
assert app_module.parse_install_progress(log, exit_code=-9)["status"] == "error"
|
|
assert app_module.parse_install_progress(log, exit_code=0)["status"] == "running"
|
|
|
|
|
|
class _FakeProc:
|
|
def __init__(self, code=None):
|
|
self.code = code
|
|
|
|
def poll(self):
|
|
return self.code
|
|
|
|
|
|
def _install_env(monkeypatch, tmp_path):
|
|
import app as app_module
|
|
|
|
monkeypatch.setattr(app_module, "STATE_DIR", tmp_path)
|
|
monkeypatch.setattr(app_module, "INSTALL_LOG", tmp_path / "install.log")
|
|
monkeypatch.setattr(app_module, "_install_proc", None)
|
|
monkeypatch.delenv("FURTKA_DRY_RUN", raising=False)
|
|
monkeypatch.setitem(app_module.settings, "boot_drive", "/dev/sda")
|
|
monkeypatch.setitem(app_module.settings, "username", "u")
|
|
monkeypatch.setitem(app_module.settings, "password", "pw12345678")
|
|
monkeypatch.setattr(app_module, "release_disk", lambda d: [f"released {d}"])
|
|
monkeypatch.setattr(app_module, "build_disk_config", lambda d: {"device": d})
|
|
return app_module
|
|
|
|
|
|
def test_install_run_refuses_second_concurrent_install(monkeypatch, tmp_path):
|
|
app_module = _install_env(monkeypatch, tmp_path)
|
|
spawned = []
|
|
monkeypatch.setattr(
|
|
app_module, "spawn_archinstall", lambda *a: spawned.append(a) or _FakeProc()
|
|
)
|
|
client = app_module.app.test_client()
|
|
assert client.post("/install/run").status_code == 302
|
|
assert client.post("/install/run").status_code == 302
|
|
assert len(spawned) == 1
|
|
|
|
|
|
def test_install_run_disk_probe_failure_lands_in_log(monkeypatch, tmp_path):
|
|
app_module = _install_env(monkeypatch, tmp_path)
|
|
|
|
def boom(_):
|
|
raise RuntimeError("parted could not read /dev/sda")
|
|
|
|
monkeypatch.setattr(app_module, "build_disk_config", boom)
|
|
resp = app_module.app.test_client().post("/install/run")
|
|
assert resp.status_code == 302
|
|
log = (tmp_path / "install.log").read_text()
|
|
assert "Traceback (most recent call last)" in log
|
|
assert app_module.parse_install_progress(log)["status"] == "error"
|
|
|
|
|
|
def test_install_log_json_reports_crashed_archinstall(monkeypatch, tmp_path):
|
|
app_module = _install_env(monkeypatch, tmp_path)
|
|
(tmp_path / "install.log").write_text("Installing packages: ['base'\n")
|
|
monkeypatch.setattr(app_module, "_install_proc", _FakeProc(code=137))
|
|
data = app_module.app.test_client().get("/install/log.json").get_json()
|
|
assert data["progress"]["status"] == "error"
|
|
assert "exit code 137" in data["log"]
|
|
|
|
|
|
def test_step2_rejects_disk_not_offered(monkeypatch):
|
|
import app as app_module
|
|
|
|
offered = [
|
|
{
|
|
"name": "/dev/sda",
|
|
"usable": True,
|
|
"size": "500 GB",
|
|
"type_label": "SSD",
|
|
"health_label": "Healthy",
|
|
"score": 27,
|
|
"reason": "",
|
|
"warning": "",
|
|
},
|
|
{
|
|
"name": "/dev/sdb",
|
|
"usable": False,
|
|
"size": "8 GB",
|
|
"type_label": "USB",
|
|
"health_label": "Status unknown",
|
|
"score": 10,
|
|
"reason": "Too small",
|
|
"warning": "",
|
|
},
|
|
]
|
|
monkeypatch.setattr(app_module, "list_scored_devices", lambda: offered)
|
|
monkeypatch.setitem(app_module.settings, "boot_drive", "")
|
|
client = app_module.app.test_client()
|
|
for bad in ("/dev/sdb", "/dev/nope", "/etc/passwd"):
|
|
resp = client.post("/install/step2", data={"boot_drive": bad})
|
|
assert resp.status_code == 200
|
|
assert app_module.settings["boot_drive"] == ""
|
|
resp = client.post("/install/step2", data={"boot_drive": "/dev/sda"})
|
|
assert resp.status_code == 302
|
|
assert app_module.settings["boot_drive"] == "/dev/sda"
|
|
|
|
|
|
def test_validate_step1_lowercases_names():
|
|
errors, values = validate_step1(
|
|
{
|
|
"hostname": "Furtka",
|
|
"username": "Daniel",
|
|
"password": "topsecretpw",
|
|
"password2": "topsecretpw",
|
|
"language": "de",
|
|
}
|
|
)
|
|
assert errors == []
|
|
assert values["hostname"] == "furtka"
|
|
assert values["username"] == "daniel"
|
|
|
|
|
|
def test_validate_step1_rejects_reserved_usernames():
|
|
for name in ("root", "caddy", "docker", "nobody"):
|
|
errors, _ = validate_step1(
|
|
{
|
|
"hostname": "furtka",
|
|
"username": name,
|
|
"password": "topsecretpw",
|
|
"password2": "topsecretpw",
|
|
"language": "en",
|
|
}
|
|
)
|
|
assert len(errors) == 1 and "reserved" in errors[0], name
|
|
|
|
|
|
def test_post_install_sets_wait_online_any(monkeypatch):
|
|
import base64
|
|
|
|
import app as app_module
|
|
|
|
cmds = app_module._post_install_commands("h", "u", "pw12345678")
|
|
hits = [c for c in cmds if "systemd-networkd-wait-online.service.d" in c]
|
|
assert len(hits) == 1
|
|
payload = hits[0].split("printf %s ")[1].split(" |")[0]
|
|
assert "--any" in base64.b64decode(payload).decode()
|
|
|
|
|
|
def test_wizard_pages_redirect_to_progress_while_installing(monkeypatch, tmp_path):
|
|
app_module = _install_env(monkeypatch, tmp_path)
|
|
monkeypatch.setattr(app_module, "_install_proc", _FakeProc(code=None))
|
|
client = app_module.app.test_client()
|
|
for path in ("/install/step1", "/install/step2", "/install/overview"):
|
|
resp = client.get(path)
|
|
assert resp.status_code == 302
|
|
assert resp.headers["Location"].endswith("/install/log")
|
|
|
|
|
|
def test_error_page_offers_retry(monkeypatch, tmp_path):
|
|
app_module = _install_env(monkeypatch, tmp_path)
|
|
(tmp_path / "install.log").write_text("Traceback (most recent call last)\nboom\n")
|
|
html = app_module.app.test_client().get("/install/log").get_data(as_text=True)
|
|
assert "Try again" in html
|
|
assert "next-steps" not in html
|
|
|
|
|
|
def test_detect_bootloader_by_firmware(tmp_path):
|
|
import app as app_module
|
|
|
|
efi = tmp_path / "efi"
|
|
assert app_module._detect_bootloader(efi) == "Grub" # BIOS / legacy
|
|
efi.mkdir()
|
|
assert app_module._detect_bootloader(efi) == "Systemd-boot" # no bitness file
|
|
(efi / "fw_platform_size").write_text("64\n")
|
|
assert app_module._detect_bootloader(efi) == "Systemd-boot"
|
|
(efi / "fw_platform_size").write_text("32\n")
|
|
assert app_module._detect_bootloader(efi) == "Limine"
|