server-up/tests/test_stackupdates.py

"""Updates per app: image-parsing, digest-vergelijking en randgevallen."""
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "server-up"))

from core import registry


# ── Image-referenties ────────────────────────────────────────────────────────

@pytest.mark.parametrize("ref,verwacht", [
    ("nginx", ("registry-1.docker.io", "library/nginx", "latest")),
    ("nginx:1.27", ("registry-1.docker.io", "library/nginx", "1.27")),
    ("bes-r/app:2.0", ("registry-1.docker.io", "bes-r/app", "2.0")),
    ("ghcr.io/org/app", ("ghcr.io", "org/app", "latest")),
    ("ghcr.io/org/app:dev", ("ghcr.io", "org/app", "dev")),
    ("git.example.com:3000/bes-r/server-up:0.5.10",
     ("git.example.com:3000", "bes-r/server-up", "0.5.10")),
    ("localhost:5000/mijn/app", ("localhost:5000", "mijn/app", "latest")),
    ("vaultwarden/server:latest", ("registry-1.docker.io", "vaultwarden/server", "latest")),
])
def test_parse_image(ref, verwacht):
    assert registry.parse_image(ref) == verwacht


def test_parse_image_negeert_digest():
    host, repo, tag = registry.parse_image("nginx@sha256:" + "a" * 64)
    assert (host, repo) == ("registry-1.docker.io", "library/nginx")


def test_parse_image_leeg():
    assert registry.parse_image("") == ("", "", "")


def test_lokale_registry_gebruikt_http():
    assert registry._scheme("localhost:5000") == "http"
    assert registry._scheme("127.0.0.1:5000") == "http"
    assert registry._scheme("git.example.com") == "https"


# ── Digest-vergelijking ──────────────────────────────────────────────────────

@pytest.fixture
def su(env):
    from core import stackupdates
    stackupdates.invalidate()
    return stackupdates


@pytest.fixture
def reg(env):
    """Het registry-module-object ná de herimport door de env-fixture."""
    from core import registry as verse_registry
    return verse_registry


def _stack(env, naam="web", compose="services:\n  app:\n    image: nginx:1.27\n"):
    d = env["lib"] / naam
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(compose, encoding="utf-8")
    return d


def test_update_gedetecteerd(su, reg, env, monkeypatch):
    from core import docker
    monkeypatch.setattr(docker, "image_digest_local", lambda i: "sha256:oud")
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "sha256:nieuw")
    res = su.check_stack("web", _stack(env))
    assert res["update_available"] is True
    assert res["services"][0]["status"] == "update"


def test_geen_update_bij_gelijke_digest(su, reg, env, monkeypatch):
    from core import docker
    monkeypatch.setattr(docker, "image_digest_local", lambda i: "sha256:zelfde")
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "sha256:zelfde")
    res = su.check_stack("web", _stack(env))
    assert res["update_available"] is False
    assert res["services"][0]["status"] == "current"


def test_lokaal_gebouwd_image_is_onbekend_geen_update(su, reg, env, monkeypatch):
    """Zonder RepoDigest valt er niets te vergelijken — dat mag geen update
    melden, anders staat er permanent een badge."""
    from core import docker
    monkeypatch.setattr(docker, "image_digest_local", lambda i: "")
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "sha256:x")
    res = su.check_stack("web", _stack(env))
    assert res["update_available"] is False
    assert res["services"][0]["status"] == "unknown"


def test_onbereikbare_registry_is_onbekend_geen_update(su, reg, env, monkeypatch):
    from core import docker
    monkeypatch.setattr(docker, "image_digest_local", lambda i: "sha256:oud")
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "")
    res = su.check_stack("web", _stack(env))
    assert res["update_available"] is False
    assert res["services"][0]["status"] == "unknown"
    assert "registry" in res["services"][0]["reason"]


def test_variabele_in_image_wordt_overgeslagen(su, env):
    d = _stack(env, compose="services:\n  app:\n    image: nginx:${TAG}\n")
    assert su._images_uit_compose(d) == []


def test_meerdere_services(su, reg, env, monkeypatch):
    from core import docker
    compose = ("services:\n"
               "  web:\n    image: nginx:1.27\n"
               "  db:\n    image: postgres:16\n")
    monkeypatch.setattr(docker, "image_digest_local", lambda i: "sha256:oud")
    monkeypatch.setattr(reg, "manifest_digest",
                        lambda i, c=None: "sha256:nieuw" if "nginx" in i else "sha256:oud")
    res = su.check_stack("web", _stack(env, compose=compose))
    statussen = {d["service"]: d["status"] for d in res["services"]}
    assert statussen == {"web": "update", "db": "current"}
    assert res["update_available"] is True


def test_check_all_cachet(su, reg, env, monkeypatch):
    from core import docker
    _stack(env)
    tellers = {"n": 0}

    def _tel(image):
        tellers["n"] += 1
        return "sha256:oud"

    monkeypatch.setattr(docker, "image_digest_local", _tel)
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "sha256:oud")
    su.check_all(force=True)
    n_na_eerste = tellers["n"]
    su.check_all()
    assert tellers["n"] == n_na_eerste


def test_kapotte_stack_blokkeert_de_rest_niet(su, reg, env, monkeypatch):
    from core import docker
    _stack(env, "goed")
    _stack(env, "stuk")

    def _ontploft(image):
        raise RuntimeError("registry weg")

    monkeypatch.setattr(docker, "image_digest_local",
                        lambda i: (_ontploft(i) if False else "sha256:x"))
    monkeypatch.setattr(reg, "manifest_digest", lambda i, c=None: "sha256:x")
    res = su.check_all(force=True)
    assert set(res) == {"goed", "stuk"}


# ── Endpoints ────────────────────────────────────────────────────────────────

def test_endpoint_updates_botst_niet_met_stack_routes(client, su):
    """/api/stacks/updates mag niet als stacknaam 'updates' gelezen worden."""
    from conftest import login
    login(client)
    r = client.get("/api/stacks/updates")
    assert r.status_code == 200
    assert "stacks" in r.get_json()


def test_endpoint_credentials_alleen_bij_zelfde_registry(su, env):
    env["core"].patch({"UPDATE_REGISTRY_USER": "bes-r",
                       "UPDATE_REGISTRY_TOKEN": "geheim",
                       "UPDATE_IMAGE": "git.example.com/bes-r/server-up"})
    assert su._registry_creds("git.example.com/bes-r/iets") == ("bes-r", "geheim")
    assert su._registry_creds("nginx:1.27") is None