daynivo-kiosk-os/tests/test_helper.py

import json
import subprocess

import pytest

from daynivo_kiosk import helper as h
from daynivo_kiosk.config import load_config


class Recorder:
    def __init__(self, outputs=None):
        self.commands = []
        self.outputs = outputs or {}

    def __call__(self, args, timeout=30, check=True, input=None):
        self.commands.append(list(args))
        key = " ".join(args[:3])
        return subprocess.CompletedProcess(args, 0, stdout=self.outputs.get(key, ""), stderr="")

    def ran(self, *prefix):
        return [c for c in self.commands if c[: len(prefix)] == list(prefix)]


@pytest.fixture
def boot(tmp_path):
    d = tmp_path / "boot"
    d.mkdir()
    (d / "config.txt").write_text("")
    return d


@pytest.fixture
def helper(paths, tmp_path, boot):
    etc = tmp_path / "sysetc"
    etc.mkdir()
    (etc / "hosts").write_text("127.0.0.1\tlocalhost\n127.0.1.1\tkiosk\n")
    run = Recorder({"nmcli -t -f": "aaaa-1111\n"})
    return h.Helper(paths=paths, run=run, etc=etc, boot_dirs=(boot,), helper_path="/usr/lib/daynivo-kiosk/daynivo-kiosk-helper")


def run_main(helper, argv, env=None, capsys=None):
    code = h.main(argv, helper=helper, environ=env or {})
    out = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
    return code, out


def test_internal_verbs_refused_via_sudo(helper, capsys):
    code, out = run_main(helper, ["network-restore"], {"SUDO_USER": "daynivo"}, capsys)
    assert code == 1 and out["ok"] is False
    code, out = run_main(helper, ["boot-import"], {"SUDO_USER": "daynivo"}, capsys)
    assert code == 1
    assert helper.run.commands == []


def test_public_verb_via_sudo(helper, capsys):
    code, out = run_main(helper, ["restart-browser"], {"SUDO_USER": "daynivo"}, capsys)
    assert code == 0 and out["ok"] is True
    assert helper.run.ran("systemctl", "restart", "--no-block", "daynivo-kiosk-browser.service")


def test_unknown_verb_and_bad_arity(helper, capsys):
    assert run_main(helper, ["rm", "-rf", "/"], capsys=capsys)[0] == 1
    assert run_main(helper, ["reboot", "now"], capsys=capsys)[0] == 1
    assert run_main(helper, ["set-hostname"], capsys=capsys)[0] == 1
    assert helper.run.commands == []


def test_requires_root_without_injected_helper(capsys, monkeypatch):
    monkeypatch.setattr(h.os, "geteuid", lambda: 1000)
    assert h.main(["reboot"], environ={}) == 1
    assert "root" in capsys.readouterr().out


def test_set_hostname(helper, capsys):
    code, out = run_main(helper, ["set-hostname", "Hal-BG"], capsys=capsys)
    assert code == 0 and out["hostname"] == "hal-bg"
    assert (helper.etc / "hostname").read_text() == "hal-bg\n"
    hosts = (helper.etc / "hosts").read_text()
    assert "127.0.1.1\thal-bg" in hosts and "127.0.1.1\tkiosk" not in hosts and "localhost" in hosts
    assert helper.run.ran("hostname", "hal-bg")


@pytest.mark.parametrize("bad", ["kiosk; reboot", "../etc", "$(id)", "-x", ""])
def test_set_hostname_rejects_injection(helper, capsys, bad):
    code, out = run_main(helper, ["set-hostname", bad], capsys=capsys)
    assert code == 1
    assert helper.run.commands == []


def test_reboot_is_delayed_through_systemd_run(helper):
    helper.reboot()
    assert helper.run.commands[0][:2] == ["systemd-run", "--on-active=3"]
    assert helper.run.commands[0][-2:] == ["/bin/systemctl", "reboot"]


ETH_STATIC = {"interface": "eth0", "method": "static", "address": "192.168.5.10/24", "gateway": "192.168.5.1", "dns": ["192.168.5.1"]}


def test_network_apply_snapshot_pending_and_timers(helper, paths):
    (paths.nm_connections_dir / "Wired connection 1.nmconnection").write_text("[connection]\nid=Wired connection 1\n")
    result = helper.network_apply(ETH_STATIC)
    keyfile = paths.nm_connections_dir / "daynivo-eth0.nmconnection"
    assert keyfile.exists() and oct(keyfile.stat().st_mode & 0o777) == "0o600"
    assert "address1=192.168.5.10/24,192.168.5.1" in keyfile.read_text()
    assert (paths.network_backup_dir / "Wired connection 1.nmconnection").exists()
    pending = json.loads(paths.network_pending_file.read_text())
    assert pending["active_before"] == ["aaaa-1111"]
    assert pending["deadline"] - pending["created"] == 180
    assert result["pending"]["interface"] == "eth0"
    runs = [c for c in helper.run.commands if c[0] == "systemd-run"]
    assert runs[0][1:3] == ["--unit=daynivo-network-rollback", "--on-active=180"] and runs[0][-1] == "network-restore"
    assert runs[1][1:3] == ["--unit=daynivo-network-activate", "--on-active=2"] and runs[1][-2:] == ["network-activate", "eth0"]


def test_second_apply_refused_while_pending(helper):
    helper.network_apply(ETH_STATIC)
    with pytest.raises(h.HelperError, match="open"):
        helper.network_apply(ETH_STATIC)


def test_confirm_clears_pending(helper, paths):
    helper.network_apply(ETH_STATIC)
    helper.network_confirm()
    assert not paths.network_pending_file.exists()
    assert not paths.network_backup_dir.exists()
    assert (paths.nm_connections_dir / "daynivo-eth0.nmconnection").exists()
    assert helper.run.ran("systemctl", "stop", "daynivo-network-rollback.timer")


def test_restore_puts_old_profiles_back(helper, paths):
    old = paths.nm_connections_dir / "Wired connection 1.nmconnection"
    old.write_text("[connection]\nid=Wired connection 1\n")
    helper.network_apply(ETH_STATIC)
    old.unlink()
    helper.network_restore()
    assert old.exists()
    assert not (paths.nm_connections_dir / "daynivo-eth0.nmconnection").exists()
    assert not paths.network_pending_file.exists()
    assert helper.run.ran("nmcli", "--wait", "30", "connection", "up", "uuid", "aaaa-1111")


def test_public_rollback_is_asynchronous(helper):
    helper.network_apply(ETH_STATIC)
    helper.run.commands.clear()
    helper.network_rollback()
    runs = [c for c in helper.run.commands if c[0] == "systemd-run"]
    assert runs and runs[0][2] == "--on-active=1" and runs[0][-1] == "network-restore"


def test_wifi_keeps_existing_password_for_same_ssid(helper, paths):
    helper.network_apply({"interface": "wlan0", "method": "dhcp", "wifi": {"ssid": "Kantoor", "psk": "geheim-wachtwoord"}})
    helper.network_confirm()
    helper.network_apply({"interface": "wlan0", "method": "static", "address": "10.0.0.9/24", "gateway": "10.0.0.1", "dns": "10.0.0.1",
                          "wifi": {"ssid": "Kantoor", "psk": ""}})
    assert "psk=geheim-wachtwoord" in (paths.nm_connections_dir / "daynivo-wlan0.nmconnection").read_text()
    helper.network_confirm()
    with pytest.raises(h.HelperError, match="wachtwoord"):
        helper.network_apply({"interface": "wlan0", "method": "dhcp", "wifi": {"ssid": "Ander", "psk": ""}})


def test_boot_guard_rolls_back_and_honours_reset(helper, paths, boot):
    old = paths.nm_connections_dir / "Wired connection 1.nmconnection"
    old.write_text("x")
    helper.network_apply(ETH_STATIC)
    helper.run.commands.clear()
    result = helper.network_boot_guard()
    assert result == {"rolled_back": True}
    assert not (paths.nm_connections_dir / "daynivo-eth0.nmconnection").exists()
    assert not any(c[0] == "nmcli" for c in helper.run.commands), "NetworkManager is not running yet at boot"

    (paths.nm_connections_dir / "daynivo-wlan0.nmconnection").write_text("x")
    (boot / "daynivo-network-reset").write_text("")
    assert helper.network_boot_guard() == {"reset": True}
    assert not (paths.nm_connections_dir / "daynivo-wlan0.nmconnection").exists()
    assert not (boot / "daynivo-network-reset").exists()
    assert old.exists()


def test_boot_guard_still_honours_the_old_reset_name(helper, paths, boot):
    (paths.nm_connections_dir / "daynivo-eth0.nmconnection").write_text("x")
    (boot / "voxip-network-reset").write_text("")
    assert helper.network_boot_guard() == {"reset": True}
    assert not (paths.nm_connections_dir / "daynivo-eth0.nmconnection").exists()
    assert not (boot / "voxip-network-reset").exists()


def test_boot_import_applies_config(helper, paths, boot):
    (boot / "daynivo-config.json").write_text(json.dumps({"device_name": "Hal", "url": "https://klant.example.nl/display/hal?token=x",
                                                           "hostname": "hal", "rotation": 90}))
    assert helper.boot_import() == {"imported": ["daynivo-config.json"]}
    conf = load_config(paths.config_file)
    assert conf["device_name"] == "Hal" and conf["rotation"] == 90 and conf["url"].startswith("https://klant")
    assert (helper.etc / "hostname").read_text() == "hal\n"
    assert (boot / "daynivo-config.json.applied").exists()


def test_boot_import_still_accepts_the_old_names(helper, paths, boot):
    (boot / "voxip-config.json").write_text(json.dumps({"device_name": "Oud", "rotation": 180}))
    (boot / "daynivo-config.json").write_text(json.dumps({"device_name": "Nieuw"}))
    assert helper.boot_import() == {"imported": ["voxip-config.json", "daynivo-config.json"]}
    conf = load_config(paths.config_file)
    assert conf["device_name"] == "Nieuw" and conf["rotation"] == 180  # the Daynivo file is applied last
    assert (boot / "voxip-config.json.applied").exists()


def test_boot_import_rejects_invalid_config(helper, paths, boot):
    before = paths.config_file.read_text()
    (boot / "daynivo-config.json").write_text(json.dumps({"url": "http://onveilig.example"}))
    helper.boot_import()
    assert paths.config_file.read_text() == before
    assert (boot / "daynivo-config.json.rejected").exists()
    assert "HTTPS" in (boot / "daynivo-config.error.txt").read_text()


def test_boot_import_rejects_non_key_lines(helper, boot):
    (boot / "daynivo-authorized_keys").write_text("dit is geen sleutel\n")
    (boot / "voxip-authorized_keys").write_text("ook geen sleutel\n")
    helper.boot_import()
    assert (boot / "daynivo-authorized_keys.rejected").exists()
    assert (boot / "voxip-authorized_keys.rejected").exists()


def test_update_verbs_are_public_and_install_runs_in_its_own_unit(helper):
    assert {"update-check", "update-install"} <= h.PUBLIC_VERBS
    result = helper.dispatch("update-install", [])
    assert "gestart" in result["message"]
    command = helper.run.ran("systemd-run")[-1]
    assert f"--unit={h.UPDATE_UNIT}" in command and command[-2:] == [h.UPDATER_PATH, "--install"]


def test_update_install_reports_a_running_update(helper):
    def busy(args, timeout=30, check=True, input=None):
        return subprocess.CompletedProcess(args, 1, stdout="", stderr="Unit daynivo-kiosk-update-now.service already exists.")

    helper.run = busy
    with pytest.raises(h.HelperError, match="al een update"):
        helper.dispatch("update-install", [])