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


# ── Eigen mappen en apparaten ────────────────────────────────────────────────

EXTRA_BASIS = """\
services:
  app:
    image: nginx
    volumes:
      - /opt/serverup/appdata/app:/config
  db:
    image: postgres:16-alpine
"""


def test_datamap_wordt_toegevoegd():
    """Een mediamap die ergens anders staat kon alleen via het compose-bestand,
    en dat wordt bij de eerstvolgende wijziging overschreven."""
    uit = ct.add_extras(EXTRA_BASIS, {"volumes": [
        {"service": "app", "host": "/srv/media", "container": "/media"}]})
    doc = yaml.safe_load(uit)
    assert "/srv/media:/media" in doc["services"]["app"]["volumes"]
    # Wat er al stond blijft staan.
    assert "/opt/serverup/appdata/app:/config" in doc["services"]["app"]["volumes"]


def test_datamap_alleen_lezen():
    uit = ct.add_extras(EXTRA_BASIS, {"volumes": [
        {"service": "app", "host": "/srv/foto", "container": "/foto", "ro": True}]})
    assert "/srv/foto:/foto:ro" in yaml.safe_load(uit)["services"]["app"]["volumes"]


def test_apparaat_wordt_toegevoegd():
    uit = ct.add_extras(EXTRA_BASIS, {"devices": [
        {"service": "app", "host": "/dev/ttyUSB0", "container": "/dev/ttyACM0"}]})
    assert yaml.safe_load(uit)["services"]["app"]["devices"] == \
        ["/dev/ttyUSB0:/dev/ttyACM0"]


def test_apparaat_zonder_pad_binnen_de_container():
    uit = ct.add_extras(EXTRA_BASIS, {"devices": [
        {"service": "app", "host": "/dev/ttyUSB0"}]})
    assert yaml.safe_load(uit)["services"]["app"]["devices"] == \
        ["/dev/ttyUSB0:/dev/ttyUSB0"]


def test_extras_zijn_idempotent():
    """Elke wijziging rendert opnieuw en past dit opnieuw toe; twee mounts op
    hetzelfde bronpad laat compose stuklopen."""
    extras = {"volumes": [{"service": "app", "host": "/srv/media",
                           "container": "/media"}]}
    een = ct.add_extras(EXTRA_BASIS, extras)
    twee = ct.add_extras(een, extras)
    volumes = yaml.safe_load(twee)["services"]["app"]["volumes"]
    assert volumes.count("/srv/media:/media") == 1


def test_onbekende_service_wordt_overgeslagen():
    """Een hernoemd onderdeel mag geen kapot compose-bestand opleveren."""
    uit = ct.add_extras(EXTRA_BASIS, {"volumes": [
        {"service": "bestaatniet", "host": "/srv/x", "container": "/x"}]})
    assert yaml.safe_load(uit)["services"].keys() == {"app", "db"}


def test_gevaarlijke_mounts_worden_geweigerd():
    """Een container die /etc of de Docker-socket mag lezen, is de host."""
    for pad in ("/", "/etc", "/etc/ssh", "/var/run/docker.sock", "/proc/self",
                "/boot", "relatief/pad", "/srv/../etc"):
        assert ct.mount_bezwaar(pad), pad
    for pad in ("/srv/media", "/mnt/schijf/films", "/opt/data"):
        assert ct.mount_bezwaar(pad) == "", pad


def test_alleen_echte_apparaten():
    for pad in ("/dev", "/srv/media", "/dev/../etc", "ttyUSB0", ""):
        assert ct.apparaat_bezwaar(pad), pad
    for pad in ("/dev/ttyUSB0", "/dev/serial/by-id/usb-abc", "/dev/dri/renderD128"):
        assert ct.apparaat_bezwaar(pad) == "", pad


def test_geweigerd_pad_belandt_niet_in_compose():
    """De controle in de API is de eerste zeef; deze is de laatste."""
    uit = ct.add_extras(EXTRA_BASIS, {
        "volumes": [{"service": "app", "host": "/var/run/docker.sock",
                     "container": "/var/run/docker.sock"}],
        "devices": [{"service": "app", "host": "/etc/shadow"}]})
    doc = yaml.safe_load(uit)
    assert "docker.sock" not in str(doc["services"]["app"].get("volumes"))
    assert "devices" not in doc["services"]["app"]


def test_zonder_extras_verandert_er_niets():
    assert ct.add_extras(EXTRA_BASIS, {}) == EXTRA_BASIS
    assert ct.add_extras(EXTRA_BASIS, {"volumes": [], "devices": []}) == EXTRA_BASIS


# ── Versietags ───────────────────────────────────────────────────────────────

TAG_BASIS = """\
services:
  app:
    image: lscr.io/linuxserver/unifi-network-application:latest
  db:
    image: registry.local:5000/mongo
  var:
    image: ${IMAGE}
"""


def test_images_per_service_splitst_repo_en_tag():
    per_naam = {i["service"]: i for i in ct.images_per_service(TAG_BASIS)}
    assert per_naam["app"]["repo"] == "lscr.io/linuxserver/unifi-network-application"
    assert per_naam["app"]["tag"] == "latest"
    # De dubbele punt in registry.local:5000 is een poort, geen tag.
    assert per_naam["db"]["repo"] == "registry.local:5000/mongo"
    assert per_naam["db"]["tag"] == "latest"
    # Een image uit een variabele gaat over de .env, niet over dit scherm.
    assert "var" not in per_naam


def test_versietag_zetten():
    uit = ct.zet_image_tags(TAG_BASIS, {"app": "10.4.57", "db": "7.0"})
    doc = yaml.safe_load(uit)
    assert doc["services"]["app"]["image"].endswith(":10.4.57")
    assert doc["services"]["db"]["image"] == "registry.local:5000/mongo:7.0"
    assert doc["services"]["var"]["image"] == "${IMAGE}"


def test_lege_of_onbekende_tag_verandert_niets():
    assert ct.zet_image_tags(TAG_BASIS, {}) == TAG_BASIS
    assert ct.zet_image_tags(TAG_BASIS, {"app": "  "}) == TAG_BASIS
    assert ct.zet_image_tags(TAG_BASIS, {"bestaatniet": "1"}) == TAG_BASIS
    # Dezelfde tag nog eens zetten laat het bestand ongemoeid.
    assert ct.zet_image_tags(TAG_BASIS, {"app": "latest"}) == TAG_BASIS