PacketStation/tests/server/test_notify_api.py

"""The Notifications endpoints: the Admin's rules, group search, test send and sent log, and Account's view and
opt-out. One database for the module; the fake relay takes the test sends."""

from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import pytest
from fakesmtp import Certs, FakeSmtp, can_make_ca, make_ca
from fastapi.testclient import TestClient
from notifyhelpers import GROUP_BASE_DN, OUTSIDE, NotifyDir, add_users, clean, fresh_db
from pshelpers import ORIGIN_HEADERS, app_conn, needs_pg, sign_in

from packetstation.settings import Settings

pytestmark = [needs_pg, pytest.mark.skipif(not can_make_ca(), reason="no way to make a test CA")]

ADMIN = "/api/admin/notifications"
ACCOUNT = "/api/account/notifications"


@pytest.fixture(scope="module")
def certs(tmp_path_factory: pytest.TempPathFactory) -> Certs:
    return make_ca(tmp_path_factory.mktemp("ca"), "relay")


@pytest.fixture(scope="module")
def relay(certs: Certs) -> Iterator[FakeSmtp]:
    with FakeSmtp(certs) as server:
        yield server


@pytest.fixture(scope="module")
def base(tmp_path_factory: pytest.TempPathFactory, relay: FakeSmtp, certs: Certs) -> Iterator[Settings]:
    yield from fresh_db(
        tmp_path_factory.mktemp("db"),
        smtp_host=relay.host,
        smtp_port=relay.port,
        smtp_ca_bundle=certs.ca,
        notify_dry_run=True,
    )


@pytest.fixture(scope="module")
def ndir() -> NotifyDir:
    return NotifyDir()


def make_client(settings: Settings, directory: NotifyDir) -> TestClient:
    from packetstation import guard
    from packetstation.api import notifications
    from packetstation.main import create_app

    guard_load = guard.load_visual_settings
    from pshelpers import fake_visual

    guard.load_visual_settings = lambda: fake_visual(False)  # type: ignore[assignment]
    try:
        app = create_app(settings, directory=directory)
    finally:
        guard.load_visual_settings = guard_load  # type: ignore[assignment]
    if not any(getattr(r, "path", "") == ADMIN for r in app.router.routes):
        # main.py registers the router; until it does, put the routes ahead of the static catch-all
        app.router.routes[0:0] = notifications.router.routes
    return TestClient(app, headers=ORIGIN_HEADERS)


@pytest.fixture(scope="module")
def client(base: Settings, ndir: NotifyDir) -> Iterator[TestClient]:
    with make_client(base, ndir) as c:
        yield c


@pytest.fixture(scope="module")
def ids(base: Settings, ndir: NotifyDir, client: TestClient) -> dict[str, str]:
    return add_users(base, ndir)


@pytest.fixture
def c(client: TestClient, base: Settings, ids: dict[str, str], relay: FakeSmtp, ndir: NotifyDir) -> TestClient:
    clean(base)
    relay.messages.clear()
    ndir.down = False
    client.cookies.clear()
    return client


def as_user(client: TestClient, sam: str) -> None:
    r = sign_in(client, sam)
    assert r.status_code == 200, r.text


def rule(client: TestClient, event_type: str) -> dict[str, Any]:
    r = client.get(ADMIN)
    assert r.status_code == 200, r.text
    return next(x for x in r.json()["rules"] if x["event_type"] == event_type)


def body_of(r: dict[str, Any], **over: Any) -> dict[str, Any]:
    out = {
        "enabled": r["enabled"],
        "delivery": r["delivery"],
        "digest_time": r["digest_time"],
        "param": r["param"]["value"] if r["param"] else None,
        "recipients": [
            {k: v for k, v in x.items() if k in ("kind", "group_dn", "group_name", "user_id", "address") and v}
            for x in r["recipients"]
        ],
        "exclusions": [x["user_id"] for x in r["exclusions"] if not x["self"]],
        "version": r["version"],
    }
    out.update(over)
    return out


# ---------------------------------------------------------------- access


def test_admin_only(c: TestClient) -> None:
    calls = [
        ("get", ADMIN, None),
        ("put", f"{ADMIN}/feed_failed", {}),
        ("get", f"{ADMIN}/groups?q=ps", None),
        ("post", f"{ADMIN}/feed_failed/test", {}),
        ("get", f"{ADMIN}/sent", None),
    ]
    for method, path, body in calls:
        r = c.request(method, path, json=body)
        assert r.status_code == 401, (path, r.status_code)
    as_user(c, "clerk.clara")
    for method, path, body in calls:
        r = c.request(method, path, json=body)
        assert r.status_code == 403 and r.json()["code"] == "forbidden", (path, r.text)
    assert c.get(ACCOUNT).status_code == 200


# ---------------------------------------------------------------- rules


def test_get_seeds_every_rule(c: TestClient, base: Settings) -> None:
    as_user(c, "admin.ann")
    body = c.get(ADMIN).json()
    assert body["dry_run"] is True and body["subject_prefix"] == "[Packet Station TEST]"
    assert body["smtp"] == {"host": base.smtp_host, "port": base.smtp_port, "from": base.smtp_from, "starttls": True}
    assert len(body["rules"]) == 14
    r = next(x for x in body["rules"] if x["event_type"] == "drawing_failing")
    assert r["label"] == "Drawing lookup failing" and r["param"] == {"name": "hours", "value": 24, "default": 24}
    assert r["enabled"] is True and r["delivery"] == "immediate" and r["digest_time"] == "07:00"
    assert r["security"] is False and r["dept_scoped"] is False and r["version"] == 1 and r["last_sent"] is None
    (rc,) = r["recipients"]
    assert (
        rc["kind"] == "group" and rc["group_dn"] == f"CN=PS-Admins,{GROUP_BASE_DN}" and rc["group_name"] == "PS-Admins"
    )
    sec = next(x for x in body["rules"] if x["event_type"] == "security_alert")
    assert sec["security"] is True and sec["param"] is None


def test_update_rule(c: TestClient, base: Settings, ids: dict[str, str], ndir: NotifyDir) -> None:
    as_user(c, "admin.ann")
    r = rule(c, "qc_reject")
    new = body_of(
        r,
        delivery="digest",
        digest_time="06:30",
        recipients=[
            {"kind": "group", "group_dn": ndir.dn(OUTSIDE)},
            {"kind": "user", "user_id": ids["pm.paula"]},
            {"kind": "mailbox", "address": "[email protected]"},
            {"kind": "dept_supervisor"},
        ],
        exclusions=[ids["sup.wes"]],
    )
    put = c.put(f"{ADMIN}/qc_reject", json=new)
    assert put.status_code == 200, put.text
    out = put.json()
    assert out["delivery"] == "digest" and out["digest_time"] == "06:30" and out["version"] == 2
    kinds = [(x["kind"], x["group_name"], x["display_name"], x["address"]) for x in out["recipients"]]
    assert kinds == [
        ("group", OUTSIDE, None, None),
        ("user", None, "Paula PM", None),
        ("mailbox", None, None, "[email protected]"),
        ("dept_supervisor", None, None, None),
    ]
    assert out["exclusions"] == [{"user_id": ids["sup.wes"], "display_name": "Wes Super", "self": False}]
    # Stale version
    again = c.put(f"{ADMIN}/qc_reject", json=new)
    assert again.status_code == 409 and again.json()["current"]["version"] == 2
    with app_conn(base) as conn:
        a = conn.execute(
            "SELECT detail FROM audit_events WHERE action = 'admin.notification_rule_updated' ORDER BY id DESC"
        ).fetchone()[0]
    assert a["event_type"] == "qc_reject" and a["before"]["delivery"] == "immediate"
    assert a["after"]["delivery"] == "digest" and len(a["after"]["recipients"]) == 4
    # A param on drawing_failing; none on a type without one
    d = rule(c, "drawing_failing")
    ok = c.put(f"{ADMIN}/drawing_failing", json=body_of(d, param=6))
    assert ok.status_code == 200 and ok.json()["param"]["value"] == 6
    bad = c.put(f"{ADMIN}/feed_failed", json=body_of(rule(c, "feed_failed"), param=3))
    assert bad.status_code == 422


@pytest.mark.parametrize(
    ("event_type", "over", "text"),
    [
        ("feed_failed", {"recipients": [{"kind": "mailbox", "address": "not an address"}]}, "email"),
        ("feed_failed", {"recipients": [{"kind": "group", "group_dn": "CN=X,DC=evil,DC=com"}]}, "domain"),
        ("feed_failed", {"recipients": [{"kind": "dept_supervisor"}]}, "department"),
        ("security_alert", {"exclusions": ["00000000-0000-0000-0000-000000000001"]}, "security"),
        ("feed_failed", {"digest_time": "25:00"}, "HH:MM"),
        ("feed_failed", {"exclusions": ["00000000-0000-0000-0000-000000000001"]}, "known user"),
    ],
)
def test_update_rule_refused(c: TestClient, event_type: str, over: dict[str, Any], text: str) -> None:
    as_user(c, "admin.ann")
    r = c.put(f"{ADMIN}/{event_type}", json=body_of(rule(c, event_type), **over))
    assert r.status_code == 422, r.text
    assert text in r.json()["detail"]


def test_unknown_event_type(c: TestClient) -> None:
    as_user(c, "admin.ann")
    assert c.post(f"{ADMIN}/nope/test", json={}).status_code == 404


def test_group_search(c: TestClient, ndir: NotifyDir) -> None:
    as_user(c, "admin.ann")
    found = c.get(f"{ADMIN}/groups", params={"q": "ops"}).json()
    assert found == [{"dn": ndir.dn(OUTSIDE), "name": OUTSIDE, "description": f"group {OUTSIDE}"}]
    assert c.get(f"{ADMIN}/groups", params={"q": "o"}).status_code == 422
    ndir.down = True
    r = c.get(f"{ADMIN}/groups", params={"q": "ops"})
    assert r.status_code == 503 and r.json()["code"] == "ldap_unavailable"


# ---------------------------------------------------------------- test send and the log


def test_test_send_goes_even_in_dry_run(c: TestClient, base: Settings, relay: FakeSmtp, ids: dict[str, str]) -> None:
    as_user(c, "admin.ann")
    r = c.post(f"{ADMIN}/backup_stale/test", json={})
    assert r.status_code == 200, r.text
    out = r.json()
    assert out["result"] == "sent" and out["error"] is None and out["sent_id"]
    assert out["subject"] == "[Packet Station TEST] Test: Backup failed or stale"
    assert [x["address"] for x in out["recipients"]] == ["[email protected]"]
    to = c.post(f"{ADMIN}/backup_stale/test", json={"to": "[email protected]"}).json()
    assert to["result"] == "sent" and [x["address"] for x in to["recipients"]] == ["[email protected]"]
    assert [m.rcpts for m in relay.messages] == [["[email protected]"], ["[email protected]"]]
    assert c.post(f"{ADMIN}/backup_stale/test", json={"to": "nope"}).status_code == 422
    with app_conn(base) as conn:
        n = conn.execute("SELECT count(*) FROM audit_events WHERE action = 'admin.notification_test'").fetchone()[0]
    assert n >= 2
    assert rule(c, "backup_stale")["last_sent"]["result"] == "sent"

    log = c.get(f"{ADMIN}/sent", params={"event_type": "backup_stale", "limit": 1}).json()
    assert len(log["rows"]) == 1 and log["rows"][0]["test"] is True and log["next_before"] == log["rows"][0]["id"]
    nxt = c.get(f"{ADMIN}/sent", params={"event_type": "backup_stale", "limit": 1, "before": log["next_before"]})
    assert nxt.json()["rows"][0]["id"] < log["rows"][0]["id"]
    row = log["rows"][0]
    assert set(row) == {"id", "at", "event_type", "digest", "test", "subject", "recipients", "result", "error"}


def test_test_send_with_mail_off(base: Settings, ndir: NotifyDir, ids: dict[str, str]) -> None:
    clean(base)
    off = base.model_copy(update={"smtp_host": ""})
    with make_client(off, ndir) as c:
        as_user(c, "admin.ann")
        out = c.post(f"{ADMIN}/feed_failed/test", json={"to": "[email protected]"}).json()
    assert out["result"] == "failed" and out["error"] == "Mail is off: SMTP_HOST is empty."


# ---------------------------------------------------------------- account


def mine(c: TestClient) -> dict[str, dict[str, Any]]:
    r = c.get(ACCOUNT)
    assert r.status_code == 200, r.text
    return {x["event_type"]: x for x in r.json()}


def test_account_view_and_opt_out(c: TestClient, base: Settings, ids: dict[str, str], ndir: NotifyDir) -> None:
    as_user(c, "admin.ann")
    q = rule(c, "qc_reject")
    recips = [*body_of(q)["recipients"], {"kind": "dept_supervisor"}, {"kind": "group", "group_dn": ndir.dn(OUTSIDE)}]
    assert c.put(f"{ADMIN}/qc_reject", json=body_of(q, recipients=recips)).status_code == 200
    f = rule(c, "feed_failed")
    assert c.put(f"{ADMIN}/feed_failed", json=body_of(f, enabled=False)).status_code == 200

    ann = mine(c)
    assert "feed_failed" not in ann  # off
    assert ann["security_alert"] == {
        "event_type": "security_alert",
        "label": "Security alert",
        "security": True,
        "delivery": "immediate",
        "reaches": True,
        "why": ["group PS-Admins"],
        "opted_out": False,
        "can_opt_out": False,
    }
    r = c.put(f"{ACCOUNT}/security_alert", json={"opt_out": True})
    assert r.status_code == 403
    r = c.put(f"{ACCOUNT}/qc_reject", json={"opt_out": True})
    assert r.status_code == 200, r.text
    assert r.json()["opted_out"] is True and r.json()["reaches"] is False
    assert c.put(f"{ACCOUNT}/qc_reject", json={"opt_out": False}).json()["opted_out"] is False
    with app_conn(base) as conn:
        acts = conn.execute(
            "SELECT detail FROM audit_events WHERE action = 'account.notification_opt_out' ORDER BY id"
        ).fetchall()
    assert [a[0]["opt_out"] for a in acts][-2:] == [True, False]

    as_user(c, "sup.wes")
    wes = mine(c)
    assert wes["qc_reject"]["reaches"] is True and wes["qc_reject"]["why"] == ["Supervisor group of 03"]
    assert wes["security_alert"]["reaches"] is False and wes["security_alert"]["why"] == []

    as_user(c, "qc.quinn")
    quinn = mine(c)
    assert quinn["qc_reject"]["why"] == [f"group {OUTSIDE}"]
    ndir.down = True
    down = mine(c)
    assert down["qc_reject"]["reaches"] is False and down["qc_reject"]["why"] == ["directory unavailable"]


def test_admin_exclusion_shows_on_account(c: TestClient, ids: dict[str, str]) -> None:
    as_user(c, "admin.ann")
    b = rule(c, "backup_stale")
    assert c.put(f"{ADMIN}/backup_stale", json=body_of(b, exclusions=[ids["admin.ann"]])).status_code == 200
    item = mine(c)["backup_stale"]
    assert item["reaches"] is False and "left out by an Admin" in item["why"] and item["opted_out"] is False