server-up/tests/test_netwerk_advies.py

"""Tijdzone, vrije poorten, vrije IP-adressen en gedeelde netwerken.

Allemaal dingen waar Server Up de gebruiker eerder mee liet gokken: de tijdzone
stond in elk sjabloon hard op Europe/Amsterdam, poortcontrole keek alleen naar
Docker, en er was één gedeeld netwerk waar élke app aan hing.
"""
import json
import sys
from pathlib import Path

import pytest

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

from conftest import login

LAN = {"name": "lan", "driver": "macvlan", "parent": "eth0",
       "subnet": "192.168.1.0/24", "gateway": "192.168.1.1",
       "ip_range": "192.168.1.240/28"}


# ── Tijdzone ─────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("tz,geldig", [
    ("Europe/Amsterdam", True), ("UTC", True), ("America/Argentina/Salta", True),
    ("Etc/GMT+2", True),
    ("", False), ("geen; tijdzone", False), ("../../etc/passwd", False),
    ("a" * 80, False), ("Europe/Amsterdam; rm -rf /", False),
])
def test_tijdzonevorm(env, tz, geldig):
    assert env["core"].geldige_tijdzone(tz) is geldig


def test_host_tijdzone_uit_de_omgeving(env, monkeypatch):
    monkeypatch.setenv("TZ", "Asia/Tokyo")
    assert env["core"].host_timezone() == "Asia/Tokyo"


def test_host_tijdzone_negeert_onzin(env, monkeypatch):
    """Een kapotte TZ mag niet doorsijpelen naar elk compose-bestand."""
    monkeypatch.setenv("TZ", "niet; goed")
    assert env["core"].host_timezone() != "niet; goed"


def test_ingestelde_tijdzone_wint(env, monkeypatch):
    monkeypatch.setenv("TZ", "Asia/Tokyo")
    env["core"].patch({"TIMEZONE": "Europe/Berlin"})
    assert env["core"].tijdzone() == "Europe/Berlin"


def test_zonder_instelling_die_van_de_server(env, monkeypatch):
    monkeypatch.setenv("TZ", "Asia/Tokyo")
    env["core"].patch({"TIMEZONE": ""})
    assert env["core"].tijdzone() == "Asia/Tokyo"


def test_ongeldige_tijdzone_wordt_geweigerd(client):
    csrf = login(client)
    r = client.put("/api/settings", json={"TIMEZONE": "geen; tijdzone"},
                   headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "tijdzone" in r.get_json()["msg"].lower()


def test_instellingen_tonen_de_tijdzone_van_de_server(client, env):
    login(client)
    d = client.get("/api/settings").get_json()
    assert d["host_timezone"]


def test_tijdzone_wordt_toegevoegd_waar_die_ontbreekt(env):
    """Negentien sjablonen hebben geen tijdzoneveld; die containers draaiden in
    UTC en al hun tijdstempels stonden er een paar uur naast."""
    from core import compose_transform as ct
    compose = ("services:\n  a:\n    image: nginx\n"
               "  b:\n    image: redis\n    environment:\n      - FOO=bar\n")
    uit = ct.zorg_voor_tijdzone(compose, "Europe/Amsterdam")
    import yaml
    doc = yaml.safe_load(uit)
    assert "TZ=Europe/Amsterdam" in doc["services"]["a"]["environment"]
    assert "TZ=Europe/Amsterdam" in doc["services"]["b"]["environment"]
    assert "FOO=bar" in doc["services"]["b"]["environment"]


def test_bestaande_tijdzone_blijft_staan(env):
    """Heeft de app zelf een tijdzoneveld, dan is dat de keuze van de gebruiker."""
    from core import compose_transform as ct
    compose = ("services:\n  a:\n    image: nginx\n"
               "    environment:\n      - TZ=Asia/Tokyo\n")
    assert ct.zorg_voor_tijdzone(compose, "Europe/Amsterdam") == compose


def test_tijdzone_als_mapping(env):
    from core import compose_transform as ct
    import yaml
    uit = ct.zorg_voor_tijdzone(
        "services:\n  a:\n    image: nginx\n    environment:\n      FOO: bar\n",
        "Europe/Berlin")
    assert yaml.safe_load(uit)["services"]["a"]["environment"]["TZ"] == "Europe/Berlin"


# ── Poorten ──────────────────────────────────────────────────────────────────

def test_poortcontrole_meldt_vrij(client, env):
    login(client)
    d = client.get("/api/ports/check?port=54321").get_json()
    assert d["ok"] is True
    assert d["free"] is True
    assert d["suggested"] == 54321


def test_poortcontrole_stelt_een_alternatief_voor(client, env, monkeypatch):
    import app as app_module
    monkeypatch.setattr(app_module, "_bezette_poorten", lambda: {8080, 8081})
    login(client)
    d = client.get("/api/ports/check?port=8080").get_json()
    assert d["free"] is False
    assert d["suggested"] == 8082


@pytest.mark.parametrize("waarde", ["0", "70000", "abc", "-1"])
def test_ongeldige_poort(client, env, waarde):
    login(client)
    assert client.get(f"/api/ports/check?port={waarde}").status_code == 400


def test_luisterpoorten_uit_proc(env):
    """/proc/net/tcp: kolom 4 is de status, 0A betekent LISTEN."""
    from core import networks
    tekst = ("  sl  local_address rem_address   st tx_queue\n"
             "   0: 0100007F:1F90 00000000:0000 0A 00000000\n"   # 8080, LISTEN
             "   1: 0100007F:0050 0100007F:C001 01 00000000\n"   # 80, ESTABLISHED
             "   2: 00000000:0035 00000000:0000 0A 00000000\n")  # 53, LISTEN
    assert networks._poorten_uit_proc(tekst) == {8080, 53}


def test_poorten_van_de_host_tellen_mee(env, monkeypatch):
    """Een nginx die rechtstreeks op de host draait ziet `docker ps` niet, maar
    je botst er wel mee: 'port is already allocated'."""
    import app as app_module
    from core import docker, networks
    monkeypatch.setattr(docker, "used_ports", lambda: {8080})
    monkeypatch.setattr(networks, "host_luisterpoorten", lambda *a, **k: {443})
    assert app_module._bezette_poorten() == {8080, 443}


# ── IP-adressen ──────────────────────────────────────────────────────────────

@pytest.fixture
def lan(client, env):
    env["core"].patch({"NETWORKS": [dict(LAN)]})
    return client


def test_ip_buiten_het_subnet(lan, env):
    login(lan)
    d = lan.get("/api/networks/lan/ip-check?ip=10.9.9.9").get_json()
    assert d["free"] is False
    assert d["in_subnet"] is False
    assert "buiten het subnet" in d["reason"]


def test_gateway_is_geen_vrij_adres(lan, env):
    login(lan)
    d = lan.get("/api/networks/lan/ip-check?ip=192.168.1.1").get_json()
    assert d["free"] is False
    assert "gateway" in d["reason"]


def test_buiten_het_ingestelde_bereik(lan, env):
    """Binnen het subnet maar buiten je range: dan zit je in het gebied waar je
    router DHCP uitdeelt, en botst het vroeg of laat."""
    login(lan)
    d = lan.get("/api/networks/lan/ip-check?ip=192.168.1.50").get_json()
    assert d["free"] is False
    assert "bereik" in d["reason"]


def test_vrij_adres(lan, env):
    login(lan)
    d = lan.get("/api/networks/lan/ip-check?ip=192.168.1.242").get_json()
    assert d["free"] is True
    assert d["reason"] == ""


def test_advies_zonder_adres(lan, env):
    login(lan)
    d = lan.get("/api/networks/lan/ip-check").get_json()
    assert d["suggested"].startswith("192.168.1.24")


def test_apparaat_op_het_lan_blokkeert(lan, env, monkeypatch):
    """Docker kent alleen zijn eigen containers; een fysiek apparaat met een
    DHCP-adres is precies de botsing die pas weken later opvalt."""
    from core import networks
    monkeypatch.setattr(networks, "buren_op_het_lan", lambda: {"192.168.1.242"})
    login(lan)
    d = lan.get("/api/networks/lan/ip-check?ip=192.168.1.242").get_json()
    assert d["free"] is False
    assert "apparaat" in d["reason"]


def test_arp_tabel_negeert_onvolledige_regels(env):
    from core import networks
    tekst = ("IP address  HW type  Flags  HW address        Mask  Device\n"
             "192.168.1.5 0x1      0x2    aa:bb:cc:dd:ee:ff *     eth0\n"
             "192.168.1.9 0x1      0x0    00:00:00:00:00:00 *     eth0\n")
    assert networks._arp_tabel(tekst) == {"192.168.1.5"}


def test_onbekend_netwerk(client, env):
    login(client)
    assert client.get("/api/networks/bestaatniet/ip-check?ip=1.2.3.4").status_code == 404


# ── Gedeelde netwerken ───────────────────────────────────────────────────────

def test_standaardnetwerk_staat_in_de_lijst(client, env):
    login(client)
    d = client.get("/api/appnetworks").get_json()
    namen = [n["name"] for n in d["networks"]]
    assert "serverup" in namen
    assert next(n for n in d["networks"] if n["name"] == "serverup")["default"]


def test_voorstel_voor_een_nieuwe_naam(client, env):
    login(client)
    d = client.get("/api/appnetworks?for=jellyfin").get_json()
    assert d["suggested"] == "su-jellyfin"


def test_voorstel_wijkt_uit_bij_een_bestaande_naam(env):
    from core import networks
    env["core"].patch({"APP_NETWORKS": ["su-media"]})
    assert networks.voorstel_netwerknaam("media") == "su-media-2"


def test_netwerk_toevoegen_onthoudt_het(env, monkeypatch):
    from core import networks
    monkeypatch.setattr(networks, "ensure_shared", lambda n: (True, ""))
    ok, _ = networks.app_netwerk_toevoegen("media")
    assert ok is True
    assert "media" in env["core"].load()["APP_NETWORKS"]


@pytest.mark.parametrize("naam", ["../weg", "met spatie", "", "-begint-fout"])
def test_ongeldige_netwerknaam(env, naam):
    from core import networks
    ok, msg = networks.app_netwerk_toevoegen(naam)
    assert ok is False
    assert "naam" in msg.lower()


def test_standaardnetwerk_kan_niet_weg(env):
    from core import networks
    ok, msg = networks.app_netwerk_verwijderen("serverup")
    assert ok is False
    assert "standaard" in msg


def test_netwerk_met_containers_kan_niet_weg(env, monkeypatch):
    from core import networks
    env["core"].patch({"APP_NETWORKS": ["media"]})
    monkeypatch.setattr(networks, "_containers_in", lambda n: {"x": {}})
    ok, msg = networks.app_netwerk_verwijderen("media")
    assert ok is False
    assert "containers" in msg


def test_netwerkbeheer_is_voor_beheerders(client, env):
    csrf = login(client)
    client.post("/api/auth/users",
                json={"username": "op", "password": "eenlangwachtwoord",
                      "role": "operator"},
                headers={"X-CSRF-Token": csrf})
    op = client.post("/api/auth/login",
                     json={"username": "op", "password": "eenlangwachtwoord"}
                     ).get_json()["csrf_token"]
    r = client.post("/api/appnetworks", json={"name": "media"},
                    headers={"X-CSRF-Token": op})
    assert r.status_code == 403


# ── Meerdere netwerken in compose ────────────────────────────────────────────

def test_stack_kan_aan_meerdere_netwerken(env):
    from core import compose_transform as ct
    import yaml
    uit = ct.add_shared_network("services:\n  a:\n    image: nginx\n",
                                ["serverup", "media"])
    doc = yaml.safe_load(uit)
    assert doc["services"]["a"]["networks"] == ["serverup", "media"]
    assert set(doc["networks"]) == {"serverup", "media"}


def test_een_naam_blijft_werken(env):
    """De oude aanroep met één string mag niet stukgaan."""
    from core import compose_transform as ct
    import yaml
    doc = yaml.safe_load(ct.add_shared_network(
        "services:\n  a:\n    image: nginx\n", "serverup"))
    assert doc["services"]["a"]["networks"] == ["serverup"]


def test_lege_lijst_verandert_niets(env):
    from core import compose_transform as ct
    compose = "services:\n  a:\n    image: nginx\n"
    assert ct.add_shared_network(compose, []) == compose


def test_network_mode_wordt_nog_steeds_overgeslagen(env):
    """Compose weigert network_mode én networks in dezelfde service."""
    from core import compose_transform as ct
    import yaml
    doc = yaml.safe_load(ct.add_shared_network(
        "services:\n  a:\n    image: nginx\n    network_mode: host\n",
        ["serverup", "media"]))
    assert "networks" not in doc["services"]["a"]