server-up/tests/test_pangolin.py

"""Pangolin: publieke URL's voor je apps.

Getest tegen een nagebouwde API, niet tegen een echte Pangolin — die heb ik
hier niet. Wat hier vastligt is dus dat wij de juiste verzoeken sturen en met
alle antwoorden overweg kunnen, niet dat de echte server ze accepteert.

De API veranderde in 1.9: een resource hing eerst onder een site en staat nu
los. Beide routes worden geprobeerd; zie fosrl/pangolin#1344.
"""
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


@pytest.fixture
def pg(env):
    from core import pangolin
    env["core"].patch({"PANGOLIN_URL": "https://pangolin.test",
                       "PANGOLIN_API_KEY": "sleutel123",
                       "PANGOLIN_ORG_ID": "mijnorg"})
    return pangolin


@pytest.fixture
def nep(pg, monkeypatch):
    """Vangt de verzoeken op en beantwoordt ze volgens een script."""
    verzonden = []
    antwoorden = {}

    def _verzoek(methode, pad, body=None):
        verzonden.append({"methode": methode, "pad": pad, "body": body})
        for sleutel, waarde in antwoorden.items():
            if sleutel[0] == methode and pad.startswith(sleutel[1]):
                return waarde
        return 404, {"message": "niet gevonden"}

    monkeypatch.setattr(pg, "_verzoek", _verzoek)
    return {"verzonden": verzonden, "antwoorden": antwoorden}


# ── Instellingen ─────────────────────────────────────────────────────────────

def test_zonder_instellingen_uitgeschakeld(env):
    from core import pangolin
    assert pangolin.ingesteld() is False


def test_met_instellingen_ingeschakeld(pg):
    assert pg.ingesteld() is True


def test_sleutel_lekt_niet_via_de_api(client, env):
    csrf = login(client)
    client.put("/api/settings",
               json={"PANGOLIN_URL": "https://pangolin.test",
                     "PANGOLIN_API_KEY": "zeergeheim", "PANGOLIN_ORG_ID": "o"},
               headers={"X-CSRF-Token": csrf})
    body = client.get("/api/pangolin").get_data(as_text=True)
    assert "zeergeheim" not in body
    assert '"has_key":true' in body.replace(" ", "")
    assert "zeergeheim" not in client.get("/api/settings").get_data(as_text=True)
    assert env["core"].load()["PANGOLIN_API_KEY"] == "zeergeheim"


# ── Verbinding testen ────────────────────────────────────────────────────────

def test_test_zonder_organisatie(pg, env):
    env["core"].patch({"PANGOLIN_ORG_ID": ""})
    ok, msg = pg.test()
    assert ok is False
    assert "organisatie" in msg


def test_test_meldt_een_geweigerde_sleutel(pg, nep):
    nep["antwoorden"][("GET", "/org/")] = (401, {"message": "unauthorized"})
    ok, msg = pg.test()
    assert ok is False
    assert "401" in msg


def test_test_slaagt(pg, nep):
    nep["antwoorden"][("GET", "/org/")] = (200, {"sites": [{"siteId": 1, "name": "thuis"}]})
    ok, msg = pg.test()
    assert ok is True
    assert "1 site" in msg


def test_netwerkfout_wordt_gemeld(pg, monkeypatch):
    def _stuk(*a, **kw):
        raise pg.PangolinError("TimeoutError: te traag")
    monkeypatch.setattr(pg, "_verzoek", _stuk)
    ok, msg = pg.test()
    assert ok is False
    assert "Timeout" in msg


# ── Lijsten uitpakken ────────────────────────────────────────────────────────

@pytest.mark.parametrize("antwoord", [
    {"sites": [{"siteId": 7, "name": "thuis"}]},
    {"data": {"sites": [{"siteId": 7, "name": "thuis"}]}},
    {"data": [{"siteId": 7, "name": "thuis"}]},
    [{"siteId": 7, "name": "thuis"}],
])
def test_sites_uit_verschillende_verpakkingen(pg, nep, antwoord):
    """Pangolin verpakt lijsten wisselend in `data`; allebei aankunnen."""
    nep["antwoorden"][("GET", "/org/")] = (200, antwoord)
    assert pg.sites() == [{"id": 7, "name": "thuis"}]


def test_domeinen(pg, nep):
    nep["antwoorden"][("GET", "/org/")] = (
        200, {"domains": [{"domainId": "d1", "baseDomain": "voorbeeld.nl"}]})
    assert pg.domeinen() == [{"id": "d1", "domain": "voorbeeld.nl"}]


# ── Publiceren ───────────────────────────────────────────────────────────────

def _standaard(nep):
    nep["antwoorden"][("PUT", "/org/mijnorg/resource")] = (
        200, {"resourceId": 42, "fullDomain": "app.voorbeeld.nl"})
    nep["antwoorden"][("PUT", "/resource/42/target")] = (200, {"targetId": 1})


def test_publiceren_maakt_resource_en_target(pg, nep):
    _standaard(nep)
    res = pg.publiceer("mijnapp", "mijnapp", "192.168.1.50", 8080,
                       site_id=3, domein_id="d1")
    assert res["url"] == "https://app.voorbeeld.nl"
    assert res["resource_id"] == 42

    resource = nep["verzonden"][0]
    assert resource["methode"] == "PUT"
    assert resource["body"]["name"] == "mijnapp"
    assert resource["body"]["http"] is True
    assert resource["body"]["domainId"] == "d1"

    target = nep["verzonden"][1]
    assert target["pad"] == "/resource/42/target"
    assert target["body"] == {"siteId": 3, "ip": "192.168.1.50", "port": 8080,
                              "method": "http", "enabled": True}


def test_valt_terug_op_de_oude_route(pg, nep):
    """Tot 1.9 hing een resource onder een site en moest siteId in de body."""
    nep["antwoorden"][("PUT", "/org/mijnorg/resource")] = (
        400, {"message": "Validation error: Unrecognized key(s) in object: 'siteId'"})
    nep["antwoorden"][("PUT", "/org/mijnorg/site/3/resource")] = (
        200, {"resourceId": 42, "fullDomain": "app.voorbeeld.nl"})
    nep["antwoorden"][("PUT", "/resource/42/target")] = (200, {})

    res = pg.publiceer("mijnapp", "mijnapp", "192.168.1.50", 8080,
                       site_id=3, domein_id="d1")
    assert res["resource_id"] == 42
    assert nep["verzonden"][1]["pad"] == "/org/mijnorg/site/3/resource"
    assert nep["verzonden"][1]["body"]["siteId"] == 3


def test_beide_routes_mislukken_geeft_beide_meldingen(pg, nep):
    """Anders sta je met een foutmelding die maar de helft van het verhaal is."""
    nep["antwoorden"][("PUT", "/org/mijnorg/resource")] = (400, {"message": "nieuw stuk"})
    nep["antwoorden"][("PUT", "/org/mijnorg/site/3/resource")] = (
        500, {"message": "oud ook stuk"})
    with pytest.raises(pg.PangolinError) as exc:
        pg.publiceer("a", "a", "1.2.3.4", 80, site_id=3, domein_id="d1")
    assert "nieuw stuk" in str(exc.value)
    assert "oud ook stuk" in str(exc.value)


def test_zonder_resource_id_stopt_hij(pg, nep):
    nep["antwoorden"][("PUT", "/org/mijnorg/resource")] = (200, {"ok": True})
    with pytest.raises(pg.PangolinError) as exc:
        pg.publiceer("a", "a", "1.2.3.4", 80, site_id=3, domein_id="d1")
    assert "resource-id" in str(exc.value)


def test_domein_wordt_opgehaald_als_het_niet_is_ingesteld(pg, nep):
    nep["antwoorden"][("GET", "/org/mijnorg/domains")] = (
        200, {"domains": [{"domainId": "auto", "baseDomain": "x.nl"}]})
    _standaard(nep)
    pg.publiceer("mijnapp", "mijnapp", "1.2.3.4", 80, site_id=3)
    resource = next(v for v in nep["verzonden"] if v["pad"].endswith("/resource"))
    assert resource["body"]["domainId"] == "auto"


def test_zonder_domein_nette_fout(pg, nep):
    nep["antwoorden"][("GET", "/org/mijnorg/domains")] = (200, {"domains": []})
    with pytest.raises(pg.PangolinError) as exc:
        pg.publiceer("a", "a", "1.2.3.4", 80, site_id=3)
    assert "domein" in str(exc.value).lower()


def test_zonder_site_wijst_hij_naar_newt(pg, nep):
    nep["antwoorden"][("GET", "/org/mijnorg/sites")] = (200, {"sites": []})
    with pytest.raises(pg.PangolinError) as exc:
        pg.publiceer("a", "a", "1.2.3.4", 80, domein_id="d1")
    assert "Newt" in str(exc.value)


def test_mislukt_target_wordt_gemeld(pg, nep):
    nep["antwoorden"][("PUT", "/org/mijnorg/resource")] = (200, {"resourceId": 42})
    nep["antwoorden"][("PUT", "/resource/42/target")] = (400, {"message": "poort bezet"})
    with pytest.raises(pg.PangolinError) as exc:
        pg.publiceer("a", "a", "1.2.3.4", 80, site_id=3, domein_id="d1")
    assert "poort bezet" in str(exc.value)


# ── Via de API ───────────────────────────────────────────────────────────────

def test_publiceren_zonder_instellingen(client, env):
    csrf = login(client)
    (env["lib"] / "web").mkdir(parents=True, exist_ok=True)
    (env["lib"] / "web" / "docker-compose.yml").write_text(
        "services:\n  web:\n    image: nginx\n", encoding="utf-8")
    r = client.post("/api/stacks/web/pangolin", json={},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "niet ingesteld" in r.get_json()["msg"]


def test_publiceren_van_onbekende_stack(client, env):
    csrf = login(client)
    r = client.post("/api/stacks/bestaatniet/pangolin", json={},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 404


def test_poort_wordt_uit_compose_gehaald(client, env, monkeypatch):
    from core import pangolin
    csrf = login(client)
    env["core"].patch({"PANGOLIN_URL": "https://p.test", "PANGOLIN_API_KEY": "k",
                       "PANGOLIN_ORG_ID": "o"})
    d = env["lib"] / "web"
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(
        'services:\n  web:\n    image: nginx\n    ports:\n      - "8123:80"\n',
        encoding="utf-8")
    (d / ".serverup.json").write_text(json.dumps({"ip": "192.168.1.60"}),
                                      encoding="utf-8")

    gezien = {}

    def _publiceer(naam, subdomein, ip, poort, site_id=None, domein_id=""):
        gezien.update({"ip": ip, "poort": poort, "sub": subdomein})
        return {"url": "https://web.voorbeeld.nl", "resource_id": 1}

    monkeypatch.setattr(pangolin, "publiceer", _publiceer)
    r = client.post("/api/stacks/web/pangolin", json={},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200, r.get_json()
    assert gezien["ip"] == "192.168.1.60"       # eigen IP gaat voor
    assert gezien["poort"] == 8123
    # En de URL wordt bewaard bij de stack.
    meta = json.loads((d / ".serverup.json").read_text())
    assert meta["pangolin_url"] == "https://web.voorbeeld.nl"


def test_alleen_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/pangolin/test", headers={"X-CSRF-Token": op})
    assert r.status_code == 403