server-up/tests/test_stack_api.py

"""Routes rond stacks: containerbeheer en de netwerkkeuze bij installeren."""
from conftest import login


def test_containers_van_onbekende_stack(client):
    login(client)
    r = client.get("/api/stacks/bestaatniet/containers")
    assert r.status_code == 404


def test_containeractie_alleen_binnen_eigen_stack(client):
    """De containernaam wordt getoetst aan compose_ps van díé stack, zodat deze
    route nooit een willekeurige container op de host kan raken."""
    csrf = login(client)
    r = client.post("/api/stacks/mijnstack/containers/traefik/stop",
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 404
    assert "hoort niet bij deze stack" in r.get_json()["msg"]


def test_onbekende_containeractie(client):
    csrf = login(client)
    r = client.post("/api/stacks/mijnstack/containers/web/verwijder",
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400


def test_containerlogs_alleen_binnen_eigen_stack(client):
    login(client)
    r = client.get("/api/stacks/mijnstack/containers/vreemde/logs")
    assert r.status_code == 404


def test_install_weigert_onbekend_netwerk(client):
    csrf = login(client)
    r = client.post("/api/store/install",
                    json={"stack": "vaultwarden", "repo_id": "server-up",
                          "instance": "vw", "network": "bestaatniet"},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "niet geconfigureerd" in r.get_json()["msg"]


def test_install_weigert_ip_buiten_subnet(client, env):
    csrf = login(client)
    env["core"].patch({"NETWORKS": [{
        "name": "lan", "driver": "macvlan", "parent": "eth0",
        "subnet": "192.168.1.0/24", "gateway": "192.168.1.1"}]})
    r = client.post("/api/store/install",
                    json={"stack": "vaultwarden", "repo_id": "server-up",
                          "instance": "vw", "network": "lan", "ip": "10.9.9.9"},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "buiten het subnet" in r.get_json()["msg"]


def test_netwerk_toevoegen_valideert(client):
    csrf = login(client)
    r = client.post("/api/networks",
                    json={"name": "lan", "driver": "macvlan",
                          "parent": "eth0; id", "subnet": "192.168.1.0/24"},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400


def test_netwerk_verwijderen_onbekend(client):
    csrf = login(client)
    r = client.delete("/api/networks/bestaatniet", headers={"X-CSRF-Token": csrf})
    assert r.status_code == 404


def test_gereserveerde_ips_worden_gelezen_uit_stacks(client, env):
    """Een geïnstalleerde stack houdt zijn adres, ook als hij gestopt is."""
    import json
    login(client)
    stack = env["lib"] / "vaultwarden"
    stack.mkdir()
    (stack / ".serverup.json").write_text(json.dumps({"ip": "192.168.1.241"}))
    import app as app_module
    with app_module.app.test_request_context():
        assert "192.168.1.241" in app_module._reserved_ips()
        assert "192.168.1.241" not in app_module._reserved_ips(exclude_stack="vaultwarden")


def test_stack_meldt_wat_er_niet_meer_draait(client, env, monkeypatch):
    """De melding bij het installeren was eenmalig. Haal je Mosquitto later weg,
    dan valt Zigbee2MQTT stil zonder dat iets zegt waarom."""
    import json as _json
    from core import docker, git
    login(client)

    bron = git.cache_dir("server-up") / "apps" / "z2m"
    (bron / "files").mkdir(parents=True)
    (bron / "template.json").write_text(_json.dumps({
        "kind": "compose",
        "metadata": {"name": "Zigbee2MQTT", "description": "d", "tags": ["mqtt"],
                     "depends_on": [{"app": "mosquitto", "reason": "broker nodig"}]},
        "variables": [],
    }), encoding="utf-8")
    (bron / "files" / "compose.yaml").write_text("services: {}\n", encoding="utf-8")

    d = env["lib"] / "z2m"
    d.mkdir(parents=True)
    (d / "docker-compose.yml").write_text("services:\n  a:\n    image: x\n")
    (d / ".serverup.json").write_text(_json.dumps(
        {"source": "z2m", "repo_id": "server-up"}))
    monkeypatch.setattr(docker, "compose_ps", lambda *a, **k: [])

    st = {s["name"]: s for s in client.get("/api/stacks").get_json()}
    assert st["z2m"]["depends_missing"] == [
        {"app": "mosquitto", "reason": "broker nodig"}]

    # Zodra de broker er staat, is de melding weg.
    mq = env["lib"] / "mqtt"
    mq.mkdir()
    (mq / "docker-compose.yml").write_text("services:\n  a:\n    image: x\n")
    (mq / ".serverup.json").write_text(_json.dumps({"source": "mosquitto"}))
    st = {s["name"]: s for s in client.get("/api/stacks").get_json()}
    assert st["z2m"]["depends_missing"] == []


def test_stacklijst_geeft_het_webadres_van_de_app(client, env):
    """Het adres stond alleen in de modal met inloggegevens; op de kaart hoort
    een knop te staan die de app opent — dus moet de lijst het adres kennen."""
    import json
    d = env["lib"] / "unifi-network"
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(
        "services:\n"
        "  unifi-mongo:\n"
        "    image: mongo:4.4.29\n"
        "  unifi:\n"
        "    image: lscr.io/linuxserver/unifi-network-application:latest\n"
        "    ports:\n"
        '      - "8443:8443"\n'
        '      - "8080:8080"\n')
    (d / ".serverup.json").write_text(json.dumps({
        "source": "unifi-network", "values": {"port_ui": 8443},
        "web": {"scheme": "https", "port": 8443}}))
    login(client)

    per_naam = {s["name"]: s for s in client.get("/api/stacks").get_json()}
    assert per_naam["unifi-network"]["web_url"].startswith("https://")
    assert per_naam["unifi-network"]["web_url"].endswith(":8443")


def test_stack_zonder_gepubliceerde_poort_heeft_geen_webadres(client, env):
    import json
    d = env["lib"] / "worker"
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(
        "services:\n  worker:\n    image: busybox\n")
    (d / ".serverup.json").write_text(json.dumps({"source": "worker"}))
    login(client)

    per_naam = {s["name"]: s for s in client.get("/api/stacks").get_json()}
    assert per_naam["worker"]["web_url"] == ""