server-up/tests/test_compose_transform.py

"""Poorten → eigen IP-adres (macvlan/ipvlan)."""
import sys
from pathlib import Path

import pytest
import yaml

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

from core import compose_transform as ct

BASIS = """\
services:
  vaultwarden:
    image: vaultwarden/server:latest
    ports:
      - "8222:80"
    volumes:
      - /data:/data
"""


def test_poorten_verdwijnen_en_ip_wordt_toegekend():
    doc = yaml.safe_load(ct.to_dedicated_ip(BASIS, "lan", "192.168.1.240"))
    svc = doc["services"]["vaultwarden"]
    assert "ports" not in svc
    assert svc["networks"]["lan"] == {"ipv4_address": "192.168.1.240"}
    assert doc["networks"] == {"lan": {"external": True}}


def test_eigen_adres_verbreekt_de_stack_niet():
    """Een `networks`-blok haalt een service uit het impliciete default-netwerk.
    Zonder `default` erbij vindt de container zijn eigen database niet meer."""
    doc = yaml.safe_load(ct.to_dedicated_ip(BASIS, "lan", "192.168.1.240"))
    assert "default" in doc["services"]["vaultwarden"]["networks"]


def test_overige_instellingen_blijven_intact():
    doc = yaml.safe_load(ct.to_dedicated_ip(BASIS, "lan", "192.168.1.240"))
    svc = doc["services"]["vaultwarden"]
    assert svc["image"] == "vaultwarden/server:latest"
    assert svc["volumes"] == ["/data:/data"]


def test_alleen_de_publicerende_service_krijgt_het_ip():
    tekst = """\
services:
  db:
    image: postgres
  web:
    image: nginx
    ports:
      - "8080:80"
"""
    doc = yaml.safe_load(ct.to_dedicated_ip(tekst, "lan", "10.0.0.5"))
    assert doc["services"]["web"]["networks"]["lan"] == {"ipv4_address": "10.0.0.5"}
    # De database blijft ongemoeid en is via het default-netwerk bereikbaar.
    assert "networks" not in doc["services"]["db"]


def test_expliciete_service_kan_gekozen_worden():
    tekst = "services:\n  a:\n    image: x\n  b:\n    image: y\n    ports: ['1:1']\n"
    doc = yaml.safe_load(ct.to_dedicated_ip(tekst, "lan", "10.0.0.9", only_service="a"))
    assert doc["services"]["a"]["networks"]["lan"] == {"ipv4_address": "10.0.0.9"}
    # b krijgt geen adres en houdt dus zijn poortmapping.
    assert doc["services"]["b"]["ports"] == ["1:1"]


# ── Meerdere containers, elk een eigen adres ─────────────────────────────────

MEERDERE = """\
services:
  sonarr:
    image: sonarr
    ports: ["8989:8989"]
  radarr:
    image: radarr
    ports: ["7878:7878"]
  bazarr:
    image: bazarr
    ports: ["6767:6767"]
  qbittorrent:
    image: qbittorrent
    network_mode: "service:gluetun"
"""


def test_meerdere_adressen_naast_elkaar():
    doc = yaml.safe_load(ct.to_dedicated_ips(
        MEERDERE, "lan", {"sonarr": "10.0.0.5", "radarr": "10.0.0.6"}))
    assert doc["services"]["sonarr"]["networks"]["lan"] == {"ipv4_address": "10.0.0.5"}
    assert doc["services"]["radarr"]["networks"]["lan"] == {"ipv4_address": "10.0.0.6"}
    assert "ports" not in doc["services"]["sonarr"]
    assert "ports" not in doc["services"]["radarr"]


def test_service_zonder_adres_houdt_zijn_poorten():
    """De bug waar dit voor gemaakt is: bij een stack met twintig webinterfaces
    verloor élke container zijn poorten zodra er één een eigen adres kreeg."""
    doc = yaml.safe_load(ct.to_dedicated_ips(MEERDERE, "lan", {"sonarr": "10.0.0.5"}))
    assert doc["services"]["bazarr"]["ports"] == ["6767:6767"]
    assert doc["services"]["radarr"]["ports"] == ["7878:7878"]


def test_geen_enkele_service_blijft_onbereikbaar_achter():
    """Poorten kwijt én geen adres is de combinatie die nergens meer te bereiken
    is. Voor elke service moet minstens één van de twee overblijven."""
    doc = yaml.safe_load(ct.to_dedicated_ips(MEERDERE, "lan", {"radarr": "10.0.0.6"}))
    for naam, svc in doc["services"].items():
        if svc.get("network_mode"):
            continue
        heeft_adres = isinstance(svc.get("networks"), dict) and "lan" in svc["networks"]
        assert svc.get("ports") or heeft_adres, f"{naam} is nergens bereikbaar"


def test_network_mode_krijgt_geen_adres():
    """Compose weigert network_mode en networks samen."""
    doc = yaml.safe_load(ct.to_dedicated_ips(
        MEERDERE, "lan", {"sonarr": "10.0.0.5", "qbittorrent": "10.0.0.7"}))
    assert "networks" not in doc["services"]["qbittorrent"]
    assert doc["services"]["qbittorrent"]["network_mode"] == "service:gluetun"


def test_onbekende_service_in_de_mapping_geeft_fout():
    with pytest.raises(ct.TransformError) as e:
        ct.to_dedicated_ips(MEERDERE, "lan", {"bestaat-niet": "10.0.0.5"})
    assert "bestaat-niet" in str(e.value)


def test_lege_mapping_laat_alles_staan():
    doc = yaml.safe_load(ct.to_dedicated_ips(MEERDERE, "lan", {}))
    assert doc["services"]["sonarr"]["ports"] == ["8989:8989"]
    assert "networks" not in doc


def test_zonder_poorten_krijgt_de_eerste_service_het_ip():
    doc = yaml.safe_load(ct.to_dedicated_ip(
        "services:\n  solo:\n    image: x\n", "lan", "10.0.0.2"))
    assert doc["services"]["solo"]["networks"]["lan"] == {"ipv4_address": "10.0.0.2"}


@pytest.mark.parametrize("tekst,fout", [
    ("dit: [is geen: geldige yaml", "geldige YAML"),
    ("version: '3'\n", "geen services"),
    ("- een\n- lijst\n", "mapping"),
])
def test_foutmeldingen(tekst, fout):
    with pytest.raises(ct.TransformError) as e:
        ct.to_dedicated_ip(tekst, "lan", "10.0.0.1")
    assert fout in str(e.value)


def test_onbekende_service_geeft_fout():
    with pytest.raises(ct.TransformError):
        ct.to_dedicated_ip(BASIS, "lan", "10.0.0.1", only_service="bestaat-niet")


def test_published_ports():
    assert ct.published_ports(BASIS) == ["8222:80"]
    assert ct.published_ports("services:\n  a:\n    image: x\n") == []