furtka/tests/test_updater.py

"""Tests for the Phase-2 self-update logic.

These tests exercise the pure logic in furtka/updater.py: version compare,
sha256 verify, tarball extract, symlink swap, state writes. The service-
restart + health-check paths are stubbed so tests don't talk to systemd or
hit the network.

FURTKA_ROOT + FURTKA_STATE_DIR + FURTKA_LOCK_PATH all override to tmp_path
so each test gets a clean filesystem.
"""

import hashlib
import io
import tarfile
from pathlib import Path

import pytest


@pytest.fixture
def updater(tmp_path, monkeypatch):
    monkeypatch.setenv("FURTKA_ROOT", str(tmp_path / "opt_furtka"))
    monkeypatch.setenv("FURTKA_STATE_DIR", str(tmp_path / "var_lib_furtka"))
    monkeypatch.setenv("FURTKA_LOCK_PATH", str(tmp_path / "update.lock"))
    monkeypatch.setenv("FURTKA_CADDYFILE_PATH", str(tmp_path / "etc_caddy" / "Caddyfile"))
    monkeypatch.setenv("FURTKA_SYSTEMD_DIR", str(tmp_path / "etc_systemd_system"))
    (tmp_path / "etc_systemd_system").mkdir()
    # Reload the module so the path constants pick up the env vars.
    import importlib

    import furtka.updater as u

    importlib.reload(u)
    return u


def _make_tarball(path: Path, version: str):
    """Build a minimal valid Furtka release tarball at `path`."""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        for name, content in [
            ("VERSION", f"{version}\n"),
            ("furtka/__init__.py", ""),
            ("apps/fileshare/manifest.json", "{}"),
        ]:
            data = content.encode()
            info = tarfile.TarInfo(name=name)
            info.size = len(data)
            tf.addfile(info, io.BytesIO(data))
    path.write_bytes(buf.getvalue())


def test_version_tuple_orders_prereleases_before_stable(updater):
    vt = updater._version_tuple
    assert vt("26.0-alpha") < vt("26.0-beta")
    assert vt("26.0-beta") < vt("26.0-rc1")
    assert vt("26.0-rc1") < vt("26.0")
    assert vt("26.0") < vt("26.1-alpha")
    assert vt("26.1-alpha") < vt("27.0-alpha")


def test_verify_tarball_accepts_matching_hash(tmp_path, updater):
    tar = tmp_path / "t.tar.gz"
    tar.write_bytes(b"hello world")
    sha = hashlib.sha256(b"hello world").hexdigest()
    updater.verify_tarball(tar, sha)  # no raise


def test_verify_tarball_rejects_mismatch(tmp_path, updater):
    tar = tmp_path / "t.tar.gz"
    tar.write_bytes(b"hello world")
    with pytest.raises(updater.UpdateError, match="sha256 mismatch"):
        updater.verify_tarball(tar, "0" * 64)


def test_parse_sha256_sidecar_strips_filename(updater):
    line = "abc123  furtka-26.1-alpha.tar.gz\n"
    assert updater._parse_sha256_sidecar(line) == "abc123"


def test_parse_sha256_sidecar_rejects_empty(updater):
    with pytest.raises(updater.UpdateError):
        updater._parse_sha256_sidecar("")


def test_extract_tarball_returns_version_and_refuses_unsafe_paths(tmp_path, updater):
    tar = tmp_path / "t.tar.gz"
    _make_tarball(tar, "26.2-alpha")
    dest = tmp_path / "dest"
    assert updater._extract_tarball(tar, dest) == "26.2-alpha"
    assert (dest / "VERSION").read_text().strip() == "26.2-alpha"

    # Build a malicious tarball with a traversal path; must refuse.
    evil = tmp_path / "evil.tar.gz"
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        info = tarfile.TarInfo(name="../escape")
        info.size = 0
        tf.addfile(info, io.BytesIO(b""))
    evil.write_bytes(buf.getvalue())
    with pytest.raises(updater.UpdateError, match="refusing"):
        updater._extract_tarball(evil, tmp_path / "dest2")


def test_write_and_read_state_round_trip(updater):
    updater.write_state("downloading", latest="26.2-alpha")
    s = updater.read_state()
    assert s["stage"] == "downloading"
    assert s["latest"] == "26.2-alpha"
    assert "updated_at" in s


def _make_release_tarball(path: Path, version: str, caddyfile_body: str = "# caddy\n"):
    """Richer tarball with assets/Caddyfile + assets/systemd/ — enough for
    apply_update's post-swap integration (caddy refresh, unit linking)."""
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        for name, content in [
            ("VERSION", f"{version}\n"),
            ("furtka/__init__.py", ""),
            ("apps/fileshare/manifest.json", "{}"),
            ("assets/Caddyfile", caddyfile_body),
            ("assets/systemd/furtka-api.service", "[Service]\nExecStart=/bin/true\n"),
        ]:
            data = content.encode()
            info = tarfile.TarInfo(name=name)
            info.size = len(data)
            tf.addfile(info, io.BytesIO(data))
    path.write_bytes(buf.getvalue())


def test_apply_update_happy_path(tmp_path, updater, monkeypatch):
    # Set up an existing "26.0-alpha" current symlink so apply_update has
    # something to swap out.
    versions = updater.versions_dir()
    versions.mkdir(parents=True)
    (versions / "26.0-alpha").mkdir()
    (versions / "26.0-alpha" / "VERSION").write_text("26.0-alpha\n")
    current = updater.current_symlink()
    current.symlink_to(versions / "26.0-alpha")

    tar = tmp_path / "t.tar.gz"
    _make_release_tarball(tar, "26.1-alpha", caddyfile_body="# new caddy config\n")

    # Stub the shell-out + health check — both succeed.
    monkeypatch.setattr(updater, "_run", lambda cmd: None)
    monkeypatch.setattr(updater, "_health_check", lambda url, deadline_s=30.0: True)

    updater.apply_update(tar, "26.1-alpha")

    assert current.resolve() == (versions / "26.1-alpha").resolve()
    assert (versions / "26.1-alpha" / "VERSION").read_text().strip() == "26.1-alpha"
    # P1-2: Caddyfile was copied into /etc/caddy/ from the new version.
    assert updater._CADDYFILE_LIVE.read_text() == "# new caddy config\n"
    state = updater.read_state()
    assert state["stage"] == "done"
    assert state["version"] == "26.1-alpha"


def test_apply_update_rolls_back_on_health_check_failure(tmp_path, updater, monkeypatch):
    versions = updater.versions_dir()
    versions.mkdir(parents=True)
    (versions / "26.0-alpha").mkdir()
    (versions / "26.0-alpha" / "VERSION").write_text("26.0-alpha\n")
    current = updater.current_symlink()
    current.symlink_to(versions / "26.0-alpha")

    tar = tmp_path / "t.tar.gz"
    _make_tarball(tar, "26.1-alpha")

    # _run succeeds (restart "works"), but the API never comes back healthy.
    monkeypatch.setattr(updater, "_run", lambda cmd: None)
    monkeypatch.setattr(updater, "_health_check", lambda url, deadline_s=30.0: False)
    # The rollback path calls subprocess.run directly — stub that too so we
    # don't actually try to restart a real service.
    import subprocess

    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: None)

    with pytest.raises(updater.UpdateError, match="rolled back"):
        updater.apply_update(tar, "26.1-alpha")

    # Symlink should point back at 26.0-alpha.
    assert current.resolve() == (versions / "26.0-alpha").resolve()
    state = updater.read_state()
    assert state["stage"] == "rolled_back"
    assert state["failed_version"] == "26.1-alpha"


def test_refresh_caddyfile_copies_when_different(updater, tmp_path):
    # Fresh /etc/caddy/ — source wins.
    src = tmp_path / "src"
    src.write_text("# new\n")
    assert updater._refresh_caddyfile(src) is True
    assert updater._CADDYFILE_LIVE.read_text() == "# new\n"

    # Same content — no-op.
    assert updater._refresh_caddyfile(src) is False


def test_refresh_caddyfile_noops_if_source_missing(updater, tmp_path):
    assert updater._refresh_caddyfile(tmp_path / "does-not-exist") is False


def test_link_new_units_only_links_missing(updater, tmp_path, monkeypatch):
    unit_dir = tmp_path / "assets_systemd"
    unit_dir.mkdir()
    (unit_dir / "furtka-foo.service").write_text("[Service]\nExecStart=/bin/true\n")
    (unit_dir / "furtka-bar.timer").write_text("[Timer]\nOnBootSec=1s\n")
    (unit_dir / "ignored.txt").write_text("not a unit")
    # Pretend furtka-foo is already linked — it must be skipped.
    (updater._SYSTEMD_DIR / "furtka-foo.service").symlink_to("/dev/null")

    seen = []
    monkeypatch.setattr(updater, "_run", lambda cmd: seen.append(cmd))

    linked = updater._link_new_units(unit_dir)
    assert linked == ["furtka-bar.timer"]
    # Only one systemctl link call — for the new timer, not the existing service.
    assert len(seen) == 1
    assert seen[0][:2] == ["systemctl", "link"]
    assert seen[0][2].endswith("furtka-bar.timer")


def test_extract_tarball_uses_data_filter_when_available(tmp_path, updater, monkeypatch):
    # Confirm we pass filter='data' to extractall on Python 3.12+; fall back
    # cleanly on older runtimes. Capture the kwarg via a stub.
    calls = []
    real_open = updater.tarfile.open  # capture before monkeypatching

    class _Recorder:
        def __init__(self, tarball):
            self._tb = real_open(tarball, "r:gz")

        def __enter__(self):
            return self

        def __exit__(self, *a):
            self._tb.close()

        def getmembers(self):
            return self._tb.getmembers()

        def extractall(self, *args, **kwargs):
            calls.append(("extractall", args, kwargs))
            # Force the TypeError branch when filter is passed, then re-run
            # without — matches the older-Python fallback.
            if "filter" in kwargs:
                raise TypeError("old python")
            return self._tb.extractall(*args)

    tar = tmp_path / "t.tar.gz"
    _make_release_tarball(tar, "26.9-alpha")
    monkeypatch.setattr(updater.tarfile, "open", lambda *a, **kw: _Recorder(tar))

    dest = tmp_path / "dest"
    updater._extract_tarball(tar, dest)
    # First call had filter=, second (fallback) didn't.
    assert len(calls) == 2
    assert calls[0][2] == {"filter": "data"}
    assert calls[1][2] == {}


def test_apply_update_rejects_version_mismatch(tmp_path, updater, monkeypatch):
    versions = updater.versions_dir()
    versions.mkdir(parents=True)
    tar = tmp_path / "t.tar.gz"
    _make_tarball(tar, "26.1-alpha")

    with pytest.raises(updater.UpdateError, match="doesn't match expected"):
        updater.apply_update(tar, "26.2-alpha")


def test_acquire_lock_prevents_concurrent_runs(tmp_path, updater):
    first = updater.acquire_lock()
    try:
        with pytest.raises(updater.UpdateError, match="already in progress"):
            updater.acquire_lock()
    finally:
        first.close()


def test_read_current_version_falls_back_to_dev(updater):
    # No symlink, no VERSION — should be "dev" not raise.
    assert updater.read_current_version() == "dev"


def test_rollback_flips_to_previous_slot(tmp_path, updater, monkeypatch):
    versions = updater.versions_dir()
    versions.mkdir(parents=True)
    for v in ("26.0-alpha", "26.1-alpha"):
        (versions / v).mkdir()
        (versions / v / "VERSION").write_text(f"{v}\n")
    current = updater.current_symlink()
    current.symlink_to(versions / "26.1-alpha")

    import subprocess

    monkeypatch.setattr(subprocess, "run", lambda *a, **kw: None)

    restored = updater.rollback()
    assert restored == "26.0-alpha"
    assert current.resolve() == (versions / "26.0-alpha").resolve()


def test_check_update_queries_forgejo_and_compares(updater, monkeypatch):
    # Stub the API and the current-version read.
    monkeypatch.setattr(updater, "read_current_version", lambda: "26.0-alpha")
    monkeypatch.setattr(
        updater,
        "_forgejo_api",
        lambda path: {
            "tag_name": "26.1-alpha",
            "assets": [
                {
                    "name": "furtka-26.1-alpha.tar.gz",
                    "browser_download_url": "https://x/t.tar.gz",
                },
                {
                    "name": "furtka-26.1-alpha.tar.gz.sha256",
                    "browser_download_url": "https://x/t.tar.gz.sha256",
                },
            ],
        },
    )
    check = updater.check_update()
    assert check.current == "26.0-alpha"
    assert check.latest == "26.1-alpha"
    assert check.update_available is True
    assert check.tarball_url == "https://x/t.tar.gz"
    assert check.sha256_url == "https://x/t.tar.gz.sha256"


def test_check_update_reports_up_to_date_when_same_version(updater, monkeypatch):
    monkeypatch.setattr(updater, "read_current_version", lambda: "26.1-alpha")
    monkeypatch.setattr(
        updater,
        "_forgejo_api",
        lambda path: {"tag_name": "26.1-alpha", "assets": []},
    )
    check = updater.check_update()
    assert check.update_available is False