server-up/tests/test_notify.py

"""Meldingen: kanalen, gebeurtenisfilter en de belofte dat send() nooit gooit.

Het punt van dit onderdeel is dat je hoort dat er iets misging. Een melding die
zelf een uitzondering veroorzaakt zou de taak die hem afvuurde alsnog laten
omvallen — dan is de kuur erger dan de kwaal.
"""
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 nt(env):
    from core import notify
    return notify


@pytest.fixture
def verzonden(nt, monkeypatch):
    """Vangt uitgaande HTTP-verzoeken op in plaats van ze te versturen."""
    opgevangen = []

    def _nep(url, data, headers):
        opgevangen.append({"url": url, "data": data, "headers": headers})
        return True, "HTTP 200"

    monkeypatch.setattr(nt, "_post", _nep)
    return opgevangen


# ── Kanaalkeuze ──────────────────────────────────────────────────────────────

def test_zonder_kanaal_gebeurt_er_niets(nt, env, verzonden):
    assert nt.ingeschakeld("backup_failed") is False
    assert nt.send("backup_failed", "x", "y") is False
    assert verzonden == []


def test_ntfy_stuurt_titel_als_header(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy",
                       "NOTIFY_URL": "https://ntfy.sh/mijn-topic"})
    ok, _ = nt.verstuur("Backup mislukt", "de schijf is vol")
    assert ok is True
    v = verzonden[0]
    assert v["url"] == "https://ntfy.sh/mijn-topic"
    assert v["headers"]["Title"] == "Backup mislukt"
    assert v["data"] == "de schijf is vol".encode()


def test_ntfy_zet_hoge_prioriteit(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t"})
    nt.verstuur("x", "y", prioriteit="hoog")
    assert verzonden[0]["headers"]["Priority"] == "high"


def test_token_gaat_mee_als_bearer(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t",
                       "NOTIFY_TOKEN": "tk_geheim"})
    nt.verstuur("x", "y")
    assert verzonden[0]["headers"]["Authorization"] == "Bearer tk_geheim"


def test_webhook_bedient_discord_en_slack_tegelijk(nt, env, verzonden):
    """Dezelfde instelling moet voor allebei werken zonder extra keuzemenu."""
    import json
    env["core"].patch({"NOTIFY_CHANNEL": "webhook",
                       "NOTIFY_URL": "https://voorbeeld.test/hook"})
    nt.verstuur("Titel", "Tekst")
    body = json.loads(verzonden[0]["data"])
    assert "Titel" in body["content"]      # Discord
    assert "Titel" in body["text"]         # Slack
    assert body["message"] == "Tekst"


def test_kanaal_zonder_url_meldt_dat(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": ""})
    ok, msg = nt.verstuur("x", "y")
    assert ok is False
    assert "URL" in msg
    assert verzonden == []


def test_email_zonder_server_meldt_dat(nt, env):
    env["core"].patch({"NOTIFY_CHANNEL": "email"})
    ok, msg = nt.verstuur("x", "y")
    assert ok is False
    assert "SMTP" in msg


def test_email_gaat_via_smtp(nt, env, monkeypatch):
    verstuurd = {}

    class NepSMTP:
        def __init__(self, host, port, timeout=None):
            verstuurd["host"], verstuurd["port"] = host, port
        def __enter__(self):
            return self
        def __exit__(self, *a):
            return False
        def starttls(self, context=None):
            verstuurd["tls"] = True
        def login(self, u, p):
            verstuurd["login"] = (u, p)
        def send_message(self, m):
            verstuurd["subject"] = m["Subject"]
            verstuurd["to"] = m["To"]
            verstuurd["body"] = m.get_content()

    monkeypatch.setattr(nt.smtplib, "SMTP", NepSMTP)
    env["core"].patch({"NOTIFY_CHANNEL": "email", "NOTIFY_EMAIL_HOST": "smtp.test",
                       "NOTIFY_EMAIL_PORT": 587, "NOTIFY_EMAIL_USER": "ik",
                       "NOTIFY_EMAIL_PASSWORD": "geheim",
                       "NOTIFY_EMAIL_TO": "ik@voorbeeld.test"})
    ok, _ = nt.verstuur("Backup mislukt", "uitleg")
    assert ok is True
    assert verstuurd["host"] == "smtp.test"
    assert verstuurd["tls"] is True
    assert verstuurd["login"] == ("ik", "geheim")
    assert "Backup mislukt" in verstuurd["subject"]
    assert "uitleg" in verstuurd["body"]


# ── Gebeurtenisfilter ────────────────────────────────────────────────────────

def test_alleen_gekozen_gebeurtenissen(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t",
                       "NOTIFY_EVENTS": ["disk_low"]})
    assert nt.send("backup_failed", "x", "y") is False
    assert nt.send("disk_low", "x", "y") is True
    assert len(verzonden) == 1


def test_testbericht_negeert_het_filter(nt, env, verzonden):
    """De testknop moet werken ook als je nog niets hebt aangevinkt."""
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t",
                       "NOTIFY_EVENTS": []})
    assert nt.verstuur("test", "test")[0] is True


def test_onbekende_gebeurtenis_wordt_genegeerd(nt, env, verzonden):
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t",
                       "NOTIFY_EVENTS": list(nt.EVENTS)})
    assert nt.send("bestaat_niet", "x", "y") is False


# ── Nooit doorgeven ──────────────────────────────────────────────────────────

def test_send_gooit_nooit(nt, env, monkeypatch):
    """Een kapot meldingskanaal mag de taak niet alsnog laten omvallen."""
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t"})

    def _ontploft(*a, **kw):
        raise RuntimeError("netwerk stuk")

    monkeypatch.setattr(nt, "_post", _ontploft)
    assert nt.send("backup_failed", "x", "y") is False        # geen exception


def test_mislukte_melding_komt_in_het_auditlog(nt, env, monkeypatch):
    from core import audit
    audit.init()
    env["core"].patch({"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t"})
    monkeypatch.setattr(nt, "_post", lambda *a, **kw: (False, "HTTP 500"))
    nt.send("backup_failed", "x", "y")
    regels = audit.query()
    assert any(r["src"] == "notify" and r["status"] == "fout" for r in regels), regels


# ── API ──────────────────────────────────────────────────────────────────────

def test_token_lekt_niet_via_de_api(client, env):
    csrf = login(client)
    client.put("/api/settings",
               json={"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t",
                     "NOTIFY_TOKEN": "tk_geheim",
                     "NOTIFY_EMAIL_PASSWORD": "smtp-geheim"},
               headers={"X-CSRF-Token": csrf})

    body = client.get("/api/notify").get_data(as_text=True)
    assert "tk_geheim" not in body
    assert "smtp-geheim" not in body
    assert '"has_token":true' in body.replace(" ", "")

    instellingen = client.get("/api/settings").get_data(as_text=True)
    assert "tk_geheim" not in instellingen
    assert "smtp-geheim" not in instellingen
    assert env["core"].load()["NOTIFY_TOKEN"] == "tk_geheim"


def test_leeg_token_wist_het_bestaande_niet(client, env):
    csrf = login(client)
    client.put("/api/settings", json={"NOTIFY_TOKEN": "blijf-staan"},
               headers={"X-CSRF-Token": csrf})
    client.put("/api/settings", json={"NOTIFY_TOKEN": ""},
               headers={"X-CSRF-Token": csrf})
    assert env["core"].load()["NOTIFY_TOKEN"] == "blijf-staan"


@pytest.mark.parametrize("waarden,melding", [
    ({"NOTIFY_CHANNEL": "duiven"}, "meldingskanaal"),
    ({"NOTIFY_EVENTS": ["bestaat_niet"]}, "gebeurtenis"),
    ({"NOTIFY_EVENTS": "geen lijst"}, "gebeurtenis"),
    ({"NOTIFY_EMAIL_SECURITY": "rot13"}, "SMTP"),
    ({"NOTIFY_EMAIL_PORT": 70000}, "NOTIFY_EMAIL_PORT"),
    ({"DISK_WARN_PCT": 150}, "DISK_WARN_PCT"),
    ({"DISK_MIN_FREE_GB": "veel"}, "DISK_MIN_FREE_GB"),
    ({"BACKUP_OFFSITE_DIR": "relatief/pad"}, "absoluut"),
])
def test_ongeldige_instellingen_worden_geweigerd(client, waarden, melding):
    csrf = login(client)
    r = client.put("/api/settings", json=waarden, headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400, r.get_json()
    assert melding in r.get_json()["msg"]


def test_geldige_instellingen_worden_bewaard(client, env):
    csrf = login(client)
    r = client.put("/api/settings",
                   json={"NOTIFY_CHANNEL": "webhook",
                         "NOTIFY_EVENTS": ["disk_low", "job_failed"],
                         "DISK_WARN_PCT": 15, "BACKUP_OFFSITE_DIR": "/mnt/backup"},
                   headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200, r.get_json()
    c = env["core"].load()
    assert c["NOTIFY_CHANNEL"] == "webhook"
    assert c["DISK_WARN_PCT"] == 15
    assert c["BACKUP_OFFSITE_DIR"] == "/mnt/backup"


def test_testknop_meldt_dat_er_geen_kanaal_is(client, env):
    csrf = login(client)
    r = client.post("/api/notify/test", headers={"X-CSRF-Token": csrf})
    assert r.status_code == 400
    assert "kanaal" in r.get_json()["msg"]


def test_testknop_verstuurt(client, env, monkeypatch):
    from core import notify
    monkeypatch.setattr(notify, "_post", lambda *a, **kw: (True, "HTTP 200"))
    csrf = login(client)
    client.put("/api/settings",
               json={"NOTIFY_CHANNEL": "ntfy", "NOTIFY_URL": "https://ntfy.sh/t"},
               headers={"X-CSRF-Token": csrf})
    r = client.post("/api/notify/test", headers={"X-CSRF-Token": csrf})
    assert r.status_code == 200, r.get_json()


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