furtka/tests/test_diskprep.py

"""release_disk() against a recorded `lsblk -P` picture of the first
hardware bench: an SSD with an old Proxmox VE `pve` volume group stacked on
sda3 (swap, root, thin pool). Commands are captured through the injected
runner — nothing touches real devices."""

import subprocess
from pathlib import Path

import pytest
from diskprep import release_disk

# What the bench looked like before cleanup (lsblk -lnP -o NAME,TYPE,MOUNTPOINTS /dev/sda).
BENCH_BEFORE = """NAME="sda" TYPE="disk" MOUNTPOINTS=""
NAME="sda1" TYPE="part" MOUNTPOINTS=""
NAME="sda3" TYPE="part" MOUNTPOINTS=""
NAME="pve-swap" TYPE="lvm" MOUNTPOINTS="[SWAP]"
NAME="pve-root" TYPE="lvm" MOUNTPOINTS="/mnt/old"
NAME="pve-data_tmeta" TYPE="lvm" MOUNTPOINTS=""
NAME="pve-data-tpool" TYPE="lvm" MOUNTPOINTS=""
NAME="pve-data" TYPE="lvm" MOUNTPOINTS=""
NAME="pve-data_tdata" TYPE="lvm" MOUNTPOINTS=""
NAME="pve-data-tpool" TYPE="lvm" MOUNTPOINTS=""
NAME="pve-data" TYPE="lvm" MOUNTPOINTS=""
"""

BENCH_CLEAN = """NAME="sda" TYPE="disk" MOUNTPOINTS=""
NAME="sda1" TYPE="part" MOUNTPOINTS=""
NAME="sda3" TYPE="part" MOUNTPOINTS=""
"""

EMPTY = """NAME="sda" TYPE="disk" MOUNTPOINTS=""
"""


class FakeRunner:
    """Answers lsblk from a script of snapshots (one per call, last one
    sticks) and records everything else."""

    def __init__(self, lsblk_outputs):
        self.lsblk_outputs = list(lsblk_outputs)
        self.calls = []

    def __call__(self, cmd, capture_output=True, text=True):
        if cmd[0] == "lsblk":
            if len(self.lsblk_outputs) > 1:
                out = self.lsblk_outputs.pop(0)
            else:
                out = self.lsblk_outputs[0]
            return subprocess.CompletedProcess(cmd, 0, out, "")
        self.calls.append(cmd)
        return subprocess.CompletedProcess(cmd, 0, "", "")


def test_release_tears_down_lvm_stack_leaves_first():
    # lsblk #1 (inventory) and #2 (first removal pass) see the full stack;
    # everything after that is clean — the dm removals "worked".
    runner = FakeRunner([BENCH_BEFORE, BENCH_BEFORE, BENCH_CLEAN])
    lines = release_disk("/dev/sda", run=runner)
    calls = runner.calls

    assert ["swapoff", "/dev/pve-swap"] in calls
    assert ["umount", "--all-targets", "/dev/pve-root"] in calls
    removed = [c[2] for c in calls if c[:2] == ["dmsetup", "remove"]]
    # Thin LV before its pool, pool before tmeta/tdata.
    assert removed.index("pve-data") < removed.index("pve-data-tpool")
    assert removed.index("pve-data-tpool") < removed.index("pve-data_tmeta")
    assert set(removed) == {
        "pve-swap",
        "pve-root",
        "pve-data_tmeta",
        "pve-data-tpool",
        "pve-data",
        "pve-data_tdata",
    }
    assert ["wipefs", "-a", "/dev/sda3"] in calls
    assert ["wipefs", "-a", "/dev/sda1"] in calls
    assert ["wipefs", "-a", "/dev/sda"] in calls
    assert calls[-1] == ["blockdev", "--rereadpt", "/dev/sda"]
    assert not any("WARNING" in line for line in lines)
    assert lines[0].startswith("Releasing /dev/sda")


def test_release_warns_when_stack_survives():
    runner = FakeRunner([BENCH_BEFORE])  # never becomes clean
    lines = release_disk("/dev/sda", run=runner)
    assert any("WARNING: still in use" in line for line in lines)
    # Still tries to give archinstall a chance.
    assert ["blockdev", "--rereadpt", "/dev/sda"] in runner.calls


def test_release_empty_disk_only_wipes_and_rereads():
    runner = FakeRunner([EMPTY])
    lines = release_disk("sda", run=runner)  # bare name is accepted too
    assert runner.calls == [["wipefs", "-a", "/dev/sda"], ["blockdev", "--rereadpt", "/dev/sda"]]
    assert any("nothing stacked" in line for line in lines)


def test_release_survives_missing_tools():
    def runner(cmd, capture_output=True, text=True):
        if cmd[0] == "lsblk":
            return subprocess.CompletedProcess(cmd, 0, BENCH_BEFORE, "")
        raise FileNotFoundError(cmd[0])

    lines = release_disk("/dev/sda", run=runner)
    assert any("command not found" in line for line in lines)


@pytest.fixture
def app_module(monkeypatch, tmp_path):
    import app as app_module

    monkeypatch.setattr(app_module, "INSTALL_LOG", tmp_path / "install.log")
    monkeypatch.setattr(app_module, "STATE_DIR", tmp_path / "state")
    return app_module


def test_install_run_releases_disk_before_spawning(app_module, monkeypatch):
    order = []

    def fake_release(disk):
        order.append(("release", disk))
        return ["Releasing X"]

    monkeypatch.setattr(app_module, "release_disk", fake_release)
    monkeypatch.setattr(app_module, "spawn_archinstall", lambda *a: order.append(("spawn",)))
    monkeypatch.setattr(app_module, "build_archinstall_config", lambda s: {})
    monkeypatch.setattr(app_module, "build_archinstall_creds", lambda s: {})
    monkeypatch.delenv("FURTKA_DRY_RUN", raising=False)
    app_module.settings.update(boot_drive="/dev/sda", username="u", password="pw12345678")

    client = app_module.app.test_client()
    resp = client.post("/install/run")

    assert resp.status_code == 302
    assert order == [("release", "/dev/sda"), ("spawn",)]
    assert Path(app_module.INSTALL_LOG).read_text().startswith("Releasing X")


def test_progress_names_disk_in_use_failure(app_module):
    log = (
        "Traceback (most recent call last):\n"
        "_ped.IOException: ... unable to inform the kernel of the change ..."
    )
    progress = app_module.parse_install_progress(log)
    assert progress["status"] == "error"
    assert "still in use" in progress["phase"]