server-up/tests/test_reconfigure.py

"""Instellingen wijzigen na installatie, koppelen, en het audit-filter."""
import json
import sys
import time
from pathlib import Path

import pytest

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

from conftest import login


def _stack(env, naam="web", meta=None, compose=None):
    d = env["lib"] / naam
    d.mkdir(parents=True, exist_ok=True)
    (d / "docker-compose.yml").write_text(
        compose or "services:\n  web:\n    image: nginx\n    ports:\n      - \"8080:80\"\n")
    if meta is not None:
        (d / ".serverup.json").write_text(json.dumps(meta), encoding="utf-8")
    return d


# ── Herconfigureren ──────────────────────────────────────────────────────────

def test_config_zonder_opgeslagen_waarden(client, env):
    """Stacks van vóór deze versie hebben geen keuzes bewaard; dan moet de
    interface uitleggen waarom de knop niet werkt in plaats van te falen."""
    login(client)
    _stack(env, "oud", {"source": "oud"})
    d = client.get("/api/stacks/oud/config").get_json()
    assert d["ok"] is False and d["editable"] is False
    assert "compose bewerken" in d["msg"]


def test_config_van_onbekende_stack(client):
    login(client)
    assert client.get("/api/stacks/bestaatniet/config").status_code == 404


def test_stacklijst_meldt_of_wijzigen_kan(client, env):
    login(client)
    _stack(env, "nieuw", {"source": "x", "values": {"port": 80}})
    _stack(env, "oud", {"source": "x"})
    per_naam = {s["name"]: s for s in client.get("/api/stacks").get_json()}
    assert per_naam["nieuw"]["editable"] is True
    assert per_naam["oud"]["editable"] is False


def test_reconfigure_zonder_bronsjabloon(client, env):
    csrf = login(client)
    _stack(env, "web", {"source": "bestaatniet", "repo_id": "x", "values": {"a": 1}})
    r = client.post("/api/stacks/web/reconfigure", json={"values": {"a": 2}},
                    headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "Bronsjabloon" in r.get_json()["msg"]


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


def test_installatie_bewaart_de_gemaakte_keuzes(client, env, monkeypatch):
    """Zonder opgeslagen waarden valt het formulier later niet te heropenen."""
    import app as app_module
    assert '"values": bp_values if is_bp else {}' in \
        (WORTEL / "server-up" / "app.py").read_text(encoding="utf-8")


# ── Koppelen ─────────────────────────────────────────────────────────────────

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


def test_connect_start_een_job(client, env):
    csrf = login(client)
    _stack(env, "web", {"source": "x"})
    d = client.post("/api/stacks/web/connect", json={"connect": True},
                    headers={"X-CSRF-Token": csrf}).get_json()
    assert d["ok"] and d["job_id"]


def test_viewer_mag_niet_herconfigureren(client, env):
    csrf = login(client)
    _stack(env, "web", {"source": "x", "values": {}})
    client.post("/api/auth/users",
                json={"username": "kijker", "password": "eenlangwachtwoord",
                      "role": "viewer"}, headers={"X-CSRF-Token": csrf})
    kijker = client.post("/api/auth/login",
                         json={"username": "kijker", "password": "eenlangwachtwoord"}
                         ).get_json()["csrf_token"]
    assert client.post("/api/stacks/web/reconfigure", json={"values": {}},
                       headers={"X-CSRF-Token": kijker}).status_code == 403


# ── Audit-filter ─────────────────────────────────────────────────────────────

@pytest.fixture
def gevuld(env):
    from core import audit
    audit.init()
    audit.log("auth", "login", "ok", ref="ramon", ip="10.0.0.1")
    audit.log("auth", "login", "error", ref="indringer", ip="10.9.9.9")
    audit.log("stacks", "start", "ok", ref="jellyfin")
    audit.log("backups", "create", "ok", ref="jellyfin")
    return audit


def test_filter_op_bron(gevuld):
    assert len(gevuld.query(100, src="auth")) == 2
    assert gevuld.count(src="auth") == 2


def test_filter_op_actie_en_status(gevuld):
    r = gevuld.query(100, src="auth", action="login", status="error")
    assert len(r) == 1 and r[0]["ref"] == "indringer"


def test_zoeken_in_referentie_en_ip(gevuld):
    assert len(gevuld.query(100, zoek="jellyfin")) == 2
    assert len(gevuld.query(100, zoek="10.9.9.9")) == 1


def test_filter_op_periode(gevuld):
    assert gevuld.count(sinds=time.time() - 60) == 4
    assert gevuld.count(sinds=time.time() + 60) == 0


def test_doorbladeren(gevuld):
    eerste = gevuld.query(2, 0)
    tweede = gevuld.query(2, 2)
    assert len(eerste) == 2 and len(tweede) == 2
    assert {r["id"] for r in eerste}.isdisjoint({r["id"] for r in tweede})


def test_facetten_voor_de_keuzelijsten(gevuld):
    f = gevuld.facets()
    assert set(f["src"]) == {"auth", "stacks", "backups"}
    assert "login" in f["action"]
    assert set(f["status"]) == {"ok", "error"}


def test_audit_endpoint_filtert(client, env, gevuld):
    login(client)
    d = client.get("/api/audit?src=auth&status=error").get_json()
    assert d["total"] == 1
    assert d["entries"][0]["ref"] == "indringer"
    assert "auth" in d["facets"]["src"]
    assert d["grand_total"] >= d["total"]


def test_audit_endpoint_limiet_wordt_begrensd(client, env, gevuld):
    login(client)
    d = client.get("/api/audit?limit=99999").get_json()
    assert d["limit"] <= 500