furtka/tests/test_app.py

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"):
        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"