daynivo-kiosk-os/tests/test_updater.py

import hashlib
import io
import json
import subprocess
import tarfile

import pytest

from daynivo_kiosk import updater as u
from daynivo_kiosk.config import save_config

API = u.RELEASE_API
BASE = "https://git.example.nl/rothman/daynivo-kiosk-os/releases/download"


def make_tarball(version, files=None, top=None):
    """A release archive like `git archive --prefix=daynivo-kiosk-os-<v>/` makes."""
    top = top or f"daynivo-kiosk-os-{version}"
    files = files or {"VERSION": f"{version}\n", "update.sh": "#!/bin/sh\n"}
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        for name, content in files.items():
            data = content.encode()
            info = tarfile.TarInfo(f"{top}/{name}" if not name.startswith(("/", "..")) else name)
            info.size = len(data)
            info.mode = 0o755
            tar.addfile(info, io.BytesIO(data))
    return buf.getvalue()


def release_json(version, assets=True):
    names = [f"daynivo-kiosk-os-{version}.tar.gz", f"daynivo-kiosk-os-{version}.tar.gz.sha256"] if assets else []
    return {
        "tag_name": f"v{version}",
        "html_url": f"https://git.example.nl/rothman/daynivo-kiosk-os/releases/tag/v{version}",
        "assets": [{"name": n, "browser_download_url": f"{BASE}/v{version}/{n}"} for n in names],
    }


class Server:
    """Stands in for the Gitea API and asset downloads."""

    def __init__(self, version="1.2.0", tarball=None, checksum=None):
        self.version = version
        self.tarball = tarball if tarball is not None else make_tarball(version)
        self.checksum = checksum or hashlib.sha256(self.tarball).hexdigest()
        self.release = release_json(version)
        self.down = False

    def __call__(self, url, limit=u.MAX_DOWNLOAD, timeout=30):
        if self.down:
            raise u.UpdateError("Downloaden mislukt: geen verbinding")
        if url == API:
            return json.dumps(self.release).encode()
        if url.endswith(".tar.gz"):
            return self.tarball
        if url.endswith(".sha256"):
            return f"{self.checksum}  daynivo-kiosk-os-{self.version}.tar.gz\n".encode()
        raise AssertionError(url)


class Run:
    def __init__(self, returncode=0, stderr=""):
        self.calls = []
        self.returncode = returncode
        self.stderr = stderr

    def __call__(self, args, **kwargs):
        self.calls.append(args)
        return subprocess.CompletedProcess(args, self.returncode, stdout="", stderr=self.stderr)


@pytest.fixture
def make(paths, tmp_path):
    (paths.share_dir / "VERSION").write_text("1.1.0\n")

    def factory(server=None, run=None):
        server = server or Server()
        run = run or Run()
        updater = u.Updater(paths, get=server, run=run, release_dir=tmp_path / "opt", cache_dir=tmp_path / "cache",
                            now=lambda: 1_800_000_000)
        return updater, server, run

    return factory


def test_versions():
    assert u.parse_version("v1.2.3") == (1, 2, 3)
    assert u.parse_version("1.2") is None
    assert u.is_newer("1.10.0", "1.9.9")
    assert not u.is_newer("1.1.0", "1.1.0")
    assert not u.is_newer("1.0.9", "1.1.0")
    assert u.is_newer("1.1.0", "1.0.0-dev")


def test_release_needs_both_assets():
    with pytest.raises(u.UpdateError, match="mist"):
        u.parse_release(release_json("1.2.0", assets=False))
    with pytest.raises(u.UpdateError, match="versienummer"):
        u.parse_release({**release_json("1.2.0"), "tag_name": "nightly"})


def test_check_records_an_available_update(make, paths):
    updater, _, run = make()
    state, release = updater.check()
    assert release.version == "1.2.0"
    assert state["available"] is True and state["current"] == "1.1.0" and state["latest"] == "1.2.0"
    assert json.loads(paths.update_state_file.read_text())["checked_at"] == 1_800_000_000
    assert run.calls == []


def test_check_when_up_to_date_or_offline(make):
    updater, server, _ = make(Server(version="1.1.0"))
    state, release = updater.check()
    assert release is None and state["available"] is False

    server.down = True
    state, release = updater.check()
    assert release is None and "geen verbinding" in state["check_error"]
    assert state["latest"] == "1.1.0"  # the last known answer stays visible


def test_install_unpacks_and_runs_update_sh(make, tmp_path, paths):
    updater, _, run = make()
    _, release = updater.check()
    state = updater.install(release)
    assert state["status"] == "done" and state["current"] == "1.2.0" and state["available"] is False
    folder = tmp_path / "opt" / "1.2.0"
    assert run.calls == [[str(folder / "update.sh"), "--no-pull"]]
    assert (folder / "VERSION").read_text().strip() == "1.2.0"


def test_install_refuses_a_bad_checksum(make, tmp_path):
    updater, _, run = make(Server(checksum="0" * 64))
    _, release = updater.check()
    state = updater.install(release)
    assert state["status"] == "failed" and "Checksum" in state["install_error"]
    assert run.calls == []
    assert not (tmp_path / "opt" / "1.2.0").exists()


def test_install_refuses_paths_outside_the_release(make):
    tarball = make_tarball("1.2.0", {"VERSION": "1.2.0\n", "../../etc/evil": "x"})
    updater, _, run = make(Server(tarball=tarball))
    _, release = updater.check()
    state = updater.install(release)
    assert state["status"] == "failed" and "Onveilig" in state["install_error"]
    assert run.calls == []


def test_install_refuses_a_mislabelled_release(make):
    updater, _, run = make(Server(tarball=make_tarball("1.2.0", {"VERSION": "9.0.0\n"})))
    _, release = updater.check()
    assert updater.install(release)["status"] == "failed"
    assert run.calls == []


def test_install_reports_a_failing_installer(make):
    updater, _, _ = make(run=Run(returncode=1, stderr="apt-get: kon niet verbinden"))
    _, release = updater.check()
    state = updater.install(release)
    assert state["status"] == "failed" and "apt-get" in state["install_error"]


def test_never_installs_an_older_version(make):
    updater, _, _ = make()
    with pytest.raises(u.UpdateError):
        updater.install(u.Release("1.0.0", f"{BASE}/a.tar.gz", f"{BASE}/a.sha256", ""))


def test_keeps_only_the_newest_releases(make, tmp_path):
    updater, _, _ = make()
    for old in ("1.0.0", "1.0.1", "1.1.0"):
        (tmp_path / "opt" / old).mkdir(parents=True)
    _, release = updater.check()
    updater.install(release)
    assert sorted(p.name for p in (tmp_path / "opt").iterdir()) == ["1.1.0", "1.2.0"]


def test_auto_mode_installs_only_when_switched_on(make, paths):
    updater, _, run = make()
    assert u.main(["--auto"], updater=updater) == 0
    assert run.calls == []

    save_config(paths.config_file, {"auto_update": True})
    (paths.health_run_dir / "maintenance").touch()
    assert u.main(["--auto"], updater=updater) == 0
    assert run.calls == []

    (paths.health_run_dir / "maintenance").unlink()
    assert u.main(["--auto"], updater=updater) == 0
    assert len(run.calls) == 1


def test_check_mode_never_installs(make):
    updater, _, run = make()
    assert u.main(["--check"], updater=updater) == 0
    assert run.calls == []