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}'"


# ── 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"] == []