server-up/tests/test_connect.py

"""Apps onderling koppelen, geavanceerde velden en de uitleg per veld."""
import sys
import tempfile
from pathlib import Path

import pytest
import yaml

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

from core import boilerplates as bp
from core import compose_transform as ct

from conftest import login

BASIS = """\
services:
  web:
    image: nginx
    ports:
      - "8080:80"
  db:
    image: postgres
"""


# ── Gedeeld netwerk ──────────────────────────────────────────────────────────

def test_gedeeld_netwerk_komt_bij_elke_service():
    """Zonder gedeeld netwerk zit elke stack in zijn eigen compose-netwerk en
    kan geen enkele app een andere op naam bereiken."""
    doc = yaml.safe_load(ct.add_shared_network(BASIS, "serverup"))
    for naam, svc in doc["services"].items():
        assert "serverup" in svc["networks"], naam
    assert doc["networks"]["serverup"] == {"external": True}


def test_gedeeld_netwerk_naast_een_eigen_ip():
    """Een stack met een macvlan-adres moet het gedeelde netwerk erbij krijgen,
    niet in plaats daarvan."""
    met_ip = ct.to_dedicated_ip(BASIS, "lan", "192.168.1.240")
    doc = yaml.safe_load(ct.add_shared_network(met_ip, "serverup"))
    web = doc["services"]["web"]["networks"]
    assert "lan" in web and "serverup" in web
    assert web["lan"] == {"ipv4_address": "192.168.1.240"}
    assert doc["networks"]["lan"]["external"] is True
    assert doc["networks"]["serverup"]["external"] is True


def test_gedeeld_netwerk_is_idempotent():
    een = ct.add_shared_network(BASIS, "serverup")
    twee = ct.add_shared_network(een, "serverup")
    doc = yaml.safe_load(twee)
    assert doc["services"]["web"]["networks"].count("serverup") == 1


def test_gedeeld_netwerk_bij_ongeldige_yaml():
    with pytest.raises(ct.TransformError):
        ct.add_shared_network("dit: [is geen: yaml", "serverup")


# ── Veldopties ───────────────────────────────────────────────────────────────

def _veld(app: str, naam: str) -> dict:
    return next(f for f in bp.fields(WORTEL / "apps" / app) if f["name"] == naam)


def test_koppelveld_wordt_doorgegeven():
    f = _veld("zigbee2mqtt", "mqtt_server")
    assert f["connect"] == {"app": "mosquitto", "scheme": "mqtt", "port": 1883}
    assert f["help"], "koppelveld hoort uitleg te hebben"


def test_open_webui_wijst_naar_ollama():
    f = _veld("open-webui", "ollama_url")
    assert f["connect"]["app"] == "ollama"
    assert f["connect"]["port"] == 11434


def test_geavanceerd_veld_wordt_gemarkeerd():
    assert _veld("zigbee2mqtt", "usb_device")["advanced"] is True
    assert _veld("zigbee2mqtt", "mqtt_server")["advanced"] is False


def test_gewone_velden_blijven_zichtbaar():
    """De basisvelden mogen niet per ongeluk achter het uitklapmenu verdwijnen."""
    for app in ("zigbee2mqtt", "immich", "frigate", "ollama"):
        velden = {f["name"]: f for f in bp.fields(WORTEL / "apps" / app)}
        assert velden["service_name"]["advanced"] is False
        if "port" in velden:
            assert velden["port"]["advanced"] is False


def test_elke_app_houdt_zichtbare_velden_over():
    """Een app waarvan alles geavanceerd is, zou een leeg formulier tonen."""
    for d in sorted((WORTEL / "apps").iterdir()):
        if not d.is_dir():
            continue
        velden = [f for f in bp.fields(d) if not f.get("is_group_toggle")]
        zichtbaar = [f for f in velden if not f["advanced"]]
        assert zichtbaar, f"{d.name}: alle velden staan als geavanceerd gemarkeerd"


def test_koppelvelden_verwijzen_naar_bestaande_apps():
    """Een connect-veld dat naar een app verwijst die niet bestaat, levert een
    lege keuzelijst op zonder dat iemand snapt waarom."""
    apps = {d.name for d in (WORTEL / "apps").iterdir() if d.is_dir()}
    for d in sorted((WORTEL / "apps").iterdir()):
        if not d.is_dir():
            continue
        for f in bp.fields(d):
            doel = (f.get("connect") or {}).get("app")
            if doel:
                assert doel in apps, f"{d.name}/{f['name']} verwijst naar '{doel}'"


# ── Afhankelijkheden tussen apps ─────────────────────────────────────────────
# Een `connect`-veld wijst naar een andere app, maar zegt niet dát je die nodig
# hebt. Zigbee2MQTT zonder MQTT-broker start prima en doet niets; `depends_on`
# maakt dat zichtbaar vóór het installeren.

def _apps():
    return sorted(d for d in (WORTEL / "apps").iterdir() if d.is_dir())


def _deps(d):
    return bp.metadata(d).get("depends_on") or []


def test_afhankelijkheden_verwijzen_naar_bestaande_apps():
    namen = {d.name for d in _apps()}
    for d in _apps():
        for dep in _deps(d):
            assert dep["app"] in namen, f"{d.name} hangt af van '{dep['app']}'"


def test_elke_afhankelijkheid_heeft_een_reden():
    """Een waarschuwing zonder uitleg is een waarschuwing die je wegklikt."""
    for d in _apps():
        for dep in _deps(d):
            assert dep["reason"], f"{d.name} → {dep['app']}: geen reden opgegeven"


def test_geen_app_hangt_van_zichzelf_af():
    for d in _apps():
        assert d.name not in [x["app"] for x in _deps(d)], d.name


def test_geen_kringetjes():
    """A heeft B nodig en B heeft A nodig levert een melding op die je nooit
    kunt wegwerken."""
    graaf = {d.name: [x["app"] for x in _deps(d)] for d in _apps()}
    bezocht: set[str] = set()

    def loop(naam, pad):
        if naam in pad:
            raise AssertionError(f"kringetje: {' → '.join(pad + [naam])}")
        if naam in bezocht:
            return
        bezocht.add(naam)
        for volgende in graaf.get(naam, []):
            loop(volgende, pad + [naam])

    for naam in graaf:
        loop(naam, [])


def test_korte_schrijfwijze_wordt_uitgeklapt(tmp_path):
    """Externe repo's hoeven niets van deze sleutel te weten; `["mosquitto"]`
    moet net zo goed werken als de volledige vorm."""
    import json as _json
    d = tmp_path / "app"
    (d / "files").mkdir(parents=True)
    (d / "template.json").write_text(_json.dumps({
        "kind": "compose",
        "metadata": {"name": "x", "description": "y", "tags": ["z"],
                     "depends_on": ["mosquitto", {"app": "loki", "reason": "omdat"}]},
        "variables": [],
    }), encoding="utf-8")
    (d / "files" / "compose.yaml").write_text("services: {}\n", encoding="utf-8")
    assert bp.metadata(d)["depends_on"] == [
        {"app": "mosquitto", "reason": ""},
        {"app": "loki", "reason": "omdat"},
    ]


def test_zonder_depends_on_een_lege_lijst():
    assert bp.metadata(WORTEL / "apps" / "vaultwarden")["depends_on"] == []


# ── Endpoint ─────────────────────────────────────────────────────────────────

def test_connectable_endpoint(client, env, monkeypatch):
    from core import docker
    import json as _json
    login(client)
    stack = env["lib"] / "mosquitto"
    stack.mkdir(parents=True)
    (stack / "docker-compose.yml").write_text("services:\n  a:\n    image: x\n")
    (stack / ".serverup.json").write_text(_json.dumps(
        {"source": "mosquitto", "name": "Mosquitto MQTT"}))
    monkeypatch.setattr(docker, "compose_ps",
                        lambda d, name=None: [{"name": "mosquitto", "service": "a",
                                               "ports": "1883/tcp", "running": True}])

    d = client.get("/api/apps/connectable").get_json()
    assert d["shared_network"] == "serverup"
    app = next(a for a in d["apps"] if a["stack"] == "mosquitto")
    assert app["source"] == "mosquitto"
    assert app["containers"][0]["name"] == "mosquitto"


def test_connectable_zonder_stacks(client):
    login(client)
    assert client.get("/api/apps/connectable").get_json()["apps"] == []


def test_preview_meldt_wat_er_nog_ontbreekt(client, env, monkeypatch):
    """De installatiemodal moet vóór het installeren kunnen tonen wat er mist."""
    import json as _json
    from core import git
    csrf = login(client)
    bron = git.cache_dir("server-up") / "apps" / "z2m"
    (bron / "files").mkdir(parents=True)
    (bron / "template.json").write_text(_json.dumps({
        "kind": "compose",
        "metadata": {"name": "Zigbee2MQTT", "description": "test", "tags": ["mqtt"],
                     "depends_on": [{"app": "mosquitto", "reason": "broker nodig"},
                                    {"app": "loki", "reason": "logs"}]},
        "variables": [{"title": "Algemeen", "items": [
            {"name": "service_name", "type": "str", "title": "Naam",
             "default": "z2m", "required": True}]}],
    }), encoding="utf-8")
    (bron / "files" / "compose.yaml").write_text(
        "services:\n  << service_name >>:\n    image: x\n", encoding="utf-8")

    # Alleen Mosquitto draait al.
    stack = env["lib"] / "mqtt"
    stack.mkdir(parents=True)
    (stack / "docker-compose.yml").write_text("services:\n  a:\n    image: x\n")
    (stack / ".serverup.json").write_text(_json.dumps({"source": "mosquitto"}))

    d = client.post("/api/store/preview", json={"stack": "z2m", "repo_id": "server-up"},
                    headers={"X-CSRF-Token": csrf}).get_json()
    deps = {x["app"]: x for x in d["depends_on"]}
    assert deps["mosquitto"]["installed"] is True
    assert deps["mosquitto"]["stacks"] == ["mqtt"]
    assert deps["loki"]["installed"] is False
    assert deps["loki"]["reason"] == "logs"


def test_store_herkent_een_installatie_aan_zijn_bron(env, tmp_path):
    """De store keek naar de mapnaam: een stack telde als installatie van
    `sonarr` zodra hij `sonarr` of `sonarr-…` heette. Wie zijn stack `media-tv`
    noemde, kreeg te zien dat Sonarr nog niet geïnstalleerd was — terwijl hij
    draaide. `.serverup.json` weet wél uit welk sjabloon hij komt.
    """
    import json as _json
    from core import git

    sjablonen = tmp_path / "sjablonen"
    (sjablonen / "sonarr" / "files").mkdir(parents=True)
    (sjablonen / "sonarr" / "template.json").write_text(_json.dumps({
        "kind": "compose",
        "metadata": {"name": "Sonarr", "description": "d", "tags": ["media"]},
        "variables": [],
    }), encoding="utf-8")
    (sjablonen / "sonarr" / "files" / "compose.yaml").write_text(
        "services: {}\n", encoding="utf-8")

    lib = env["lib"]
    # Een installatie met een eigen naam …
    eigen = lib / "media-tv"
    eigen.mkdir(parents=True)
    (eigen / ".serverup.json").write_text(_json.dumps({"source": "sonarr"}))
    # … en eentje die toevallig wel zo heet, zonder metadata.
    (lib / "sonarr-2").mkdir()

    stacks = {s["dir"]: s for s in git._scan_compose_dirs(sjablonen, lib)}
    assert stacks["sonarr"]["instances"] == ["media-tv", "sonarr-2"]


def test_store_ziet_geen_installatie_van_een_andere_app(env, tmp_path):
    import json as _json
    from core import git

    sjablonen = tmp_path / "sjablonen"
    (sjablonen / "radarr" / "files").mkdir(parents=True)
    (sjablonen / "radarr" / "template.json").write_text(_json.dumps({
        "kind": "compose",
        "metadata": {"name": "Radarr", "description": "d", "tags": ["media"]},
        "variables": [],
    }), encoding="utf-8")
    (sjablonen / "radarr" / "files" / "compose.yaml").write_text(
        "services: {}\n", encoding="utf-8")

    lib = env["lib"]
    d = lib / "media-tv"
    d.mkdir(parents=True)
    (d / ".serverup.json").write_text(_json.dumps({"source": "sonarr"}))

    stacks = {s["dir"]: s for s in git._scan_compose_dirs(sjablonen, lib)}
    assert stacks["radarr"]["instances"] == []