furtka/tests/test_catalog.py

"""Tests for the apps-catalog sync flow.

Same shape as ``tests/test_updater.py``: fixture reloads the module with
env-overridden paths, fake tarballs land in tmp_path, Forgejo API is
stubbed via ``urllib.request.urlopen`` monkeypatching so nothing talks
to the network.

Asserts end-to-end atomicity: on any failure path — bad sha256, broken
tarball, invalid manifest — the live catalog dir is either left
untouched (if one existed) or absent (if it didn't).
"""

from __future__ import annotations

import io
import json
import tarfile
from pathlib import Path

import pytest


@pytest.fixture
def catalog(tmp_path, monkeypatch):
    monkeypatch.setenv("FURTKA_CATALOG_DIR", str(tmp_path / "var_lib_furtka_catalog"))
    monkeypatch.setenv("FURTKA_CATALOG_STATE", str(tmp_path / "var_lib_furtka_catalog-state.json"))
    monkeypatch.setenv("FURTKA_CATALOG_LOCK", str(tmp_path / "catalog.lock"))
    monkeypatch.setenv("FURTKA_FORGEJO_HOST", "forgejo.test.local")
    monkeypatch.setenv("FURTKA_CATALOG_REPO", "daniel/furtka-apps")

    import importlib

    from furtka import catalog as c
    from furtka import paths as p

    importlib.reload(p)
    importlib.reload(c)
    return c


def _manifest(name: str = "fileshare") -> dict:
    return {
        "name": name,
        "display_name": "Fileshare",
        "version": "0.1.0",
        "description": "Test fixture app",
        "volumes": ["files"],
        "ports": [445],
        "icon": "icon.svg",
    }


def _make_catalog_tarball(
    path: Path,
    version: str,
    *,
    apps: list[tuple[str, dict]] | None = None,
    extra_entries: list[tuple[str, bytes]] | None = None,
) -> None:
    """Build a minimal valid catalog tarball.

    `apps` is a list of (folder_name, manifest_dict). Each app folder gets
    a `manifest.json` + a stub `docker-compose.yaml` + `icon.svg`.
    `extra_entries` lets tests inject malformed content (path-traversal,
    missing VERSION, ...) without rebuilding the helper.
    """
    apps = apps if apps is not None else [("fileshare", _manifest())]
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tf:
        entries: list[tuple[str, bytes]] = [("VERSION", f"{version}\n".encode())]
        for folder, m in apps:
            entries.append((f"apps/{folder}/manifest.json", json.dumps(m).encode()))
            entries.append(
                (f"apps/{folder}/docker-compose.yaml", b"services:\n  app:\n    image: scratch\n")
            )
            entries.append((f"apps/{folder}/icon.svg", b"<svg/>"))
        if extra_entries:
            entries.extend(extra_entries)
        for name, data in entries:
            info = tarfile.TarInfo(name=name)
            info.size = len(data)
            tf.addfile(info, io.BytesIO(data))
    path.write_bytes(buf.getvalue())


def _stub_forgejo_release(
    monkeypatch,
    catalog,
    *,
    tag: str,
    tarball_url: str = "https://forgejo.test.local/t.tar.gz",
    sha_url: str = "https://forgejo.test.local/t.tar.gz.sha256",
    releases: list | None = None,
):
    """Patch ``_rc.forgejo_api`` so check_catalog sees a canned release list."""
    if releases is None:
        releases = [
            {
                "tag_name": tag,
                "assets": [
                    {"name": f"furtka-apps-{tag}.tar.gz", "browser_download_url": tarball_url},
                    {
                        "name": f"furtka-apps-{tag}.tar.gz.sha256",
                        "browser_download_url": sha_url,
                    },
                ],
            }
        ]

    def fake_api(host, repo, path, *, error_cls=RuntimeError):
        return releases

    from furtka import _release_common as _rc

    monkeypatch.setattr(_rc, "forgejo_api", fake_api)


def _stub_download(monkeypatch, catalog, mapping: dict[str, bytes]):
    """Patch ``_rc.download`` so sync_catalog pulls from an in-memory map."""
    from furtka import _release_common as _rc

    def fake_download(url, dest, *, error_cls=RuntimeError):
        if url not in mapping:
            raise error_cls(f"test: no fake content for {url}")
        dest.parent.mkdir(parents=True, exist_ok=True)
        dest.write_bytes(mapping[url])

    monkeypatch.setattr(_rc, "download", fake_download)


# --------------------------------------------------------------------------- #
# check_catalog
# --------------------------------------------------------------------------- #


def test_check_catalog_reports_update_when_versions_differ(catalog, monkeypatch, tmp_path):
    # Pretend we already have catalog version 26.5 on disk; Forgejo reports 26.6.
    catalog.catalog_dir().mkdir(parents=True)
    (catalog.catalog_dir() / "VERSION").write_text("26.5\n")
    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")

    check = catalog.check_catalog()
    assert check.current == "26.5"
    assert check.latest == "26.6"
    assert check.update_available is True
    assert check.tarball_url.endswith(".tar.gz")
    assert check.sha256_url.endswith(".sha256")


def test_check_catalog_reports_up_to_date_when_same_version(catalog, monkeypatch):
    catalog.catalog_dir().mkdir(parents=True)
    (catalog.catalog_dir() / "VERSION").write_text("26.5\n")
    _stub_forgejo_release(monkeypatch, catalog, tag="26.5")

    check = catalog.check_catalog()
    assert check.current == "26.5"
    assert check.latest == "26.5"
    assert check.update_available is False


def test_check_catalog_treats_missing_current_as_installable(catalog, monkeypatch):
    # Fresh box, no catalog ever synced — any release is an update.
    _stub_forgejo_release(monkeypatch, catalog, tag="26.5")

    check = catalog.check_catalog()
    assert check.current is None
    assert check.update_available is True


def test_check_catalog_raises_when_no_releases_published(catalog, monkeypatch):
    _stub_forgejo_release(monkeypatch, catalog, tag="x", releases=[])
    with pytest.raises(catalog.CatalogError, match="no catalog releases"):
        catalog.check_catalog()


# --------------------------------------------------------------------------- #
# sync_catalog — happy + error paths
# --------------------------------------------------------------------------- #


def test_sync_catalog_happy_path(catalog, monkeypatch, tmp_path):
    import hashlib

    tarball_path = tmp_path / "tarball.tar.gz"
    _make_catalog_tarball(tarball_path, "26.6")
    tarball_bytes = tarball_path.read_bytes()
    sha = hashlib.sha256(tarball_bytes).hexdigest()

    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")
    _stub_download(
        monkeypatch,
        catalog,
        {
            "https://forgejo.test.local/t.tar.gz": tarball_bytes,
            "https://forgejo.test.local/t.tar.gz.sha256": (
                f"{sha}  furtka-apps-26.6.tar.gz\n".encode()
            ),
        },
    )

    check = catalog.sync_catalog()
    assert check.latest == "26.6"
    assert (catalog.catalog_dir() / "VERSION").read_text().strip() == "26.6"
    assert (catalog.catalog_dir() / "apps" / "fileshare" / "manifest.json").is_file()
    state = catalog.read_state()
    assert state["stage"] == "done"
    assert state["version"] == "26.6"


def test_sync_catalog_noop_when_already_current(catalog, monkeypatch, tmp_path):
    catalog.catalog_dir().mkdir(parents=True)
    (catalog.catalog_dir() / "VERSION").write_text("26.5\n")
    _stub_forgejo_release(monkeypatch, catalog, tag="26.5")

    check = catalog.sync_catalog()
    assert check.update_available is False
    assert catalog.read_state()["stage"] == "done"


def test_sync_catalog_refuses_sha256_mismatch(catalog, monkeypatch, tmp_path):
    tarball_path = tmp_path / "tarball.tar.gz"
    _make_catalog_tarball(tarball_path, "26.6")

    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")
    _stub_download(
        monkeypatch,
        catalog,
        {
            "https://forgejo.test.local/t.tar.gz": tarball_path.read_bytes(),
            # Hash for some OTHER content — will mismatch.
            "https://forgejo.test.local/t.tar.gz.sha256": (b"0" * 64 + b"  wrong.tar.gz\n"),
        },
    )

    with pytest.raises(catalog.CatalogError, match="sha256 mismatch"):
        catalog.sync_catalog()
    # Live catalog never existed, must still not exist after the failed sync.
    assert not catalog.catalog_dir().exists()


def test_sync_catalog_refuses_tarball_with_invalid_manifest(catalog, monkeypatch, tmp_path):
    import hashlib

    bad_manifest = {"name": "broken"}  # missing required fields

    tarball_path = tmp_path / "tarball.tar.gz"
    _make_catalog_tarball(tarball_path, "26.6", apps=[("broken", bad_manifest)])
    tarball_bytes = tarball_path.read_bytes()
    sha = hashlib.sha256(tarball_bytes).hexdigest()

    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")
    _stub_download(
        monkeypatch,
        catalog,
        {
            "https://forgejo.test.local/t.tar.gz": tarball_bytes,
            "https://forgejo.test.local/t.tar.gz.sha256": (
                f"{sha}  furtka-apps-26.6.tar.gz\n".encode()
            ),
        },
    )

    with pytest.raises(catalog.CatalogError, match="invalid manifest"):
        catalog.sync_catalog()
    # Staging was cleaned; live catalog never materialised.
    assert not catalog.catalog_dir().exists()


def test_sync_catalog_preserves_existing_catalog_on_failure(catalog, monkeypatch, tmp_path):
    """A failed sync must leave the previous live catalog intact so boxes
    keep working until the next successful sync."""
    import hashlib

    # Seed a live catalog that represents a previous successful sync.
    live = catalog.catalog_dir()
    live.mkdir(parents=True)
    (live / "VERSION").write_text("26.5\n")
    (live / "apps").mkdir()

    bad_manifest = {"name": "broken"}  # invalid
    tarball_path = tmp_path / "tarball.tar.gz"
    _make_catalog_tarball(tarball_path, "26.6", apps=[("broken", bad_manifest)])
    sha = hashlib.sha256(tarball_path.read_bytes()).hexdigest()

    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")
    _stub_download(
        monkeypatch,
        catalog,
        {
            "https://forgejo.test.local/t.tar.gz": tarball_path.read_bytes(),
            "https://forgejo.test.local/t.tar.gz.sha256": f"{sha}  x\n".encode(),
        },
    )

    with pytest.raises(catalog.CatalogError):
        catalog.sync_catalog()
    # The 26.5 live catalog survives the failed 26.6 sync.
    assert (live / "VERSION").read_text().strip() == "26.5"


def test_sync_catalog_lock_contention(catalog, monkeypatch):
    _stub_forgejo_release(monkeypatch, catalog, tag="26.6")

    # Hold the lock from outside; the real sync_catalog call must refuse.
    first = catalog.acquire_lock()
    try:
        with pytest.raises(catalog.CatalogError, match="already in progress"):
            catalog.sync_catalog()
    finally:
        first.close()


# --------------------------------------------------------------------------- #
# state + current-version helpers
# --------------------------------------------------------------------------- #


def test_read_current_catalog_version_absent(catalog):
    assert catalog.read_current_catalog_version() is None


def test_read_current_catalog_version_empty_file(catalog):
    catalog.catalog_dir().mkdir(parents=True)
    (catalog.catalog_dir() / "VERSION").write_text("\n")
    assert catalog.read_current_catalog_version() is None


def test_write_and_read_state_round_trip(catalog):
    catalog.write_state("downloading", latest="26.6")
    s = catalog.read_state()
    assert s["stage"] == "downloading"
    assert s["latest"] == "26.6"
    assert "updated_at" in s