564 lines
28 KiB
Python
564 lines
28 KiB
Python
 
|
"""Notifications on a real database: raise and resolve, seed, recipient resolution, delivery through the fake relay,
|
||
|
|
dry run, the digest, the watch checks, the version note, the feed hook and the two jobs.
|
||
|
|
|
||
|
|
One database for the module (the tunnel is slow); each test starts from empty notification tables.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import logging
|
||
|
|
from collections.abc import Iterator
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from typing import Any
|
||
|
|
from zoneinfo import ZoneInfo
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fakesmtp import Certs, FakeSmtp, can_make_ca, make_ca
|
||
|
|
from notifyhelpers import OUTSIDE, NotifyDir, add_users, clean, fresh_db, sent_since
|
||
|
|
from pshelpers import app_conn, needs_pg
|
||
|
|
|
||
|
|
from packetstation.feed.config import FeedSettings
|
||
|
|
from packetstation.notify import digest, raise_event, resolve_event, seed, send, watch
|
||
|
|
from packetstation.notify.routing import Resolver, load_rule
|
||
|
|
from packetstation.settings import Settings
|
||
|
|
from packetstation.worker.jobs import JobContext, Stack
|
||
|
|
|
||
|
|
pytestmark = [needs_pg, pytest.mark.skipif(not can_make_ca(), reason="no way to make a test CA")]
|
||
|
|
|
||
|
|
|
||
|
|
@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
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope="module")
|
||
|
|
def ndir() -> NotifyDir:
|
||
|
|
return NotifyDir()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope="module")
|
||
|
|
def ids(base: Settings, ndir: NotifyDir) -> dict[str, str]:
|
||
|
|
return add_users(base, ndir)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def s(base: Settings, ids: dict[str, str], relay: FakeSmtp, ndir: NotifyDir) -> Settings:
|
||
|
|
relay.messages.clear()
|
||
|
|
ndir.down = False
|
||
|
|
return base
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def marker(s: Settings) -> int:
|
||
|
|
return clean(s)
|
||
|
|
|
||
|
|
|
||
|
|
def one(settings: Settings, sql: str, params: Any = ()) -> Any:
|
||
|
|
with app_conn(settings) as conn:
|
||
|
|
return conn.execute(sql, params).fetchone()
|
||
|
|
|
||
|
|
|
||
|
|
def set_recipients(conn: Any, event_type: str, rows: list[dict[str, Any]]) -> None:
|
||
|
|
conn.execute("DELETE FROM notification_recipients WHERE event_type = %s", (event_type,))
|
||
|
|
for r in rows:
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO notification_recipients (event_type, kind, group_dn, group_name, user_id, address)"
|
||
|
|
" VALUES (%s, %s, %s, %s, %s, %s)",
|
||
|
|
(event_type, r["kind"], r.get("group_dn"), r.get("group_name"), r.get("user_id"), r.get("address")),
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def ctx_for(conn: Any, settings: Settings, directory: Any) -> JobContext:
|
||
|
|
stack = Stack()
|
||
|
|
stack.cache["directory"] = directory
|
||
|
|
return JobContext(
|
||
|
|
conn=conn,
|
||
|
|
settings=settings,
|
||
|
|
feed_settings=FeedSettings(),
|
||
|
|
run_id=0,
|
||
|
|
trigger="cli",
|
||
|
|
started_at=datetime.now(ZoneInfo(settings.site_tz)),
|
||
|
|
log=logging.getLogger("test.notify"),
|
||
|
|
stack=stack,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- raise and resolve
|
||
|
|
|
||
|
|
|
||
|
|
def test_condition_dedupes_and_resolve_adds_recovery(s: Settings, marker: int) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
first = raise_event(conn, "backup_stale", subject="prod", title="Backup is stale", condition=True)
|
||
|
|
assert first is not None
|
||
|
|
again = raise_event(conn, "backup_stale", subject="prod", title="Backup is stale", body="b2", condition=True)
|
||
|
|
assert again is None
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT raised_count, resolved_at, kind, event_type, body FROM notification_events WHERE id = %s", (first,)
|
||
|
|
).fetchone()
|
||
|
|
assert row[0] == 2 and row[1] is None and row[2] == row[3] == "backup_stale" and row[4] == "b2"
|
||
|
|
rec = resolve_event(conn, "backup_stale", subject="prod")
|
||
|
|
assert rec is not None
|
||
|
|
r = conn.execute(
|
||
|
|
"SELECT recovery_of, title, resolved_at, severity FROM notification_events WHERE id = %s", (rec,)
|
||
|
|
).fetchone()
|
||
|
|
assert r[0] == first and r[1] == "Cleared: Backup is stale" and r[2] is not None and r[3] == "info"
|
||
|
|
assert conn.execute("SELECT resolved_at FROM notification_events WHERE id = %s", (first,)).fetchone()[0]
|
||
|
|
assert resolve_event(conn, "backup_stale", subject="prod") is None
|
||
|
|
# Open again after it cleared: a new row, a new mail
|
||
|
|
assert raise_event(conn, "backup_stale", subject="prod", title="Backup is stale", condition=True) is not None
|
||
|
|
|
||
|
|
|
||
|
|
def test_one_shot_dedupe_and_unknown_type(s: Settings, marker: int) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
a = raise_event(conn, "security_alert", subject="ann", title="Cooldown", dedupe_minutes=15)
|
||
|
|
b = raise_event(conn, "security_alert", subject="ann", title="Cooldown", dedupe_minutes=15)
|
||
|
|
c = raise_event(conn, "security_alert", subject="bob", title="Cooldown", dedupe_minutes=15)
|
||
|
|
d = raise_event(conn, "security_alert", subject="ann", title="Cooldown")
|
||
|
|
assert a and c and d and b is None
|
||
|
|
assert conn.execute("SELECT resolved_at FROM notification_events WHERE id = %s", (a,)).fetchone()[0]
|
||
|
|
assert raise_event(conn, "no_such_type", subject="x", title="x") is None
|
||
|
|
assert resolve_event(conn, "no_such_type", subject="x") is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_seed_to_admins_is_idempotent(s: Settings, marker: int) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
added = seed(conn, s)
|
||
|
|
assert len(added) == 13
|
||
|
|
rows = conn.execute(
|
||
|
|
"SELECT r.event_type, r.enabled, r.delivery::text, r.digest_time::text, r.param, c.kind::text,"
|
||
|
|
" c.group_dn, c.group_name FROM notification_rules r JOIN notification_recipients c USING (event_type)"
|
||
|
|
).fetchall()
|
||
|
|
assert len(rows) == 13
|
||
|
|
for r in rows:
|
||
|
|
assert r[1] is True and r[2] == "immediate" and r[3] == "07:00:00" and r[5] == "group"
|
||
|
|
assert r[6] == f"CN=PS-Admins,{s.ldap_group_base_dn}" and r[7] == "PS-Admins"
|
||
|
|
params = {r[0]: r[4] for r in rows}
|
||
|
|
assert params["drawing_failing"] == 24 and params["co_waiting"] == 3 and params["qc_reject"] is None
|
||
|
|
conn.execute("UPDATE notification_rules SET enabled = false WHERE event_type = 'feed_failed'")
|
||
|
|
conn.execute("DELETE FROM notification_recipients WHERE event_type = 'feed_failed'")
|
||
|
|
assert seed(conn, s) == []
|
||
|
|
assert (
|
||
|
|
conn.execute("SELECT enabled FROM notification_rules WHERE event_type = 'feed_failed'").fetchone()[0]
|
||
|
|
is False
|
||
|
|
)
|
||
|
|
n = conn.execute("SELECT count(*) FROM notification_recipients WHERE event_type = 'feed_failed'").fetchone()[0]
|
||
|
|
assert n == 0
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- recipients
|
||
|
|
|
||
|
|
|
||
|
|
def test_resolve_recipients(s: Settings, marker: int, ndir: NotifyDir, ids: dict[str, str]) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
set_recipients(
|
||
|
|
conn,
|
||
|
|
"qc_reject",
|
||
|
|
[
|
||
|
|
{"kind": "group", "group_dn": f"CN=PS-Admins,{s.ldap_group_base_dn}", "group_name": "PS-Admins"},
|
||
|
|
{"kind": "group", "group_dn": ndir.dn(OUTSIDE), "group_name": OUTSIDE},
|
||
|
|
{"kind": "user", "user_id": ids["pm.paula"]},
|
||
|
|
{"kind": "user", "user_id": ids["disabled.dan"]},
|
||
|
|
{"kind": "mailbox", "address": "[email protected]"},
|
||
|
|
{"kind": "mailbox", "address": "[email protected]"},
|
||
|
|
{"kind": "dept_supervisor"},
|
||
|
|
],
|
||
|
|
)
|
||
|
|
rule = load_rule(conn, "qc_reject")
|
||
|
|
assert rule is not None
|
||
|
|
got = Resolver(conn, s, ndir).resolve(rule, "03")
|
||
|
|
by = {r.address: r for r in got}
|
||
|
|
assert by["[email protected]"].why == "group PS-Admins" # once, the group wins over the mailbox
|
||
|
|
assert by["[email protected]"].user_id == ids["admin.ann"]
|
||
|
|
assert by["[email protected]"].why == f"group {OUTSIDE}"
|
||
|
|
assert by["[email protected]"].why == "user"
|
||
|
|
assert "[email protected]" not in by # not active
|
||
|
|
assert by["[email protected]"].why == "mailbox" and by["[email protected]"].user_id is None
|
||
|
|
sup = {a for a, r in by.items() if r.why == "Supervisor group of 03"}
|
||
|
|
assert sup == {"[email protected]", "[email protected]"}
|
||
|
|
assert len(got) == len(by)
|
||
|
|
# No department: the Supervisor recipient resolves to nobody
|
||
|
|
none_dept = {r.address for r in Resolver(conn, s, ndir).resolve(rule, None)}
|
||
|
|
assert "[email protected]" not in none_dept
|
||
|
|
# Exclusions drop by id, and by the person's email (the mailbox row with Ann's address too)
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO notification_exclusions (event_type, user_id) VALUES ('qc_reject', %s), ('qc_reject', %s)",
|
||
|
|
(ids["admin.ann"], ids["sup.wes"]),
|
||
|
|
)
|
||
|
|
rule = load_rule(conn, "qc_reject")
|
||
|
|
assert rule is not None
|
||
|
|
after = {r.address for r in Resolver(conn, s, ndir).resolve(rule, "03")}
|
||
|
|
assert "[email protected]" not in after and "[email protected]" not in after
|
||
|
|
assert "[email protected]" in after and "[email protected]" in after
|
||
|
|
|
||
|
|
|
||
|
|
def test_group_members_disabled_or_without_mail_are_left_out(s: Settings, marker: int, ndir: NotifyDir) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
set_recipients(conn, "feed_failed", [{"kind": "group", "group_dn": ndir.dn("PS-03-WeldParts-Worker")}])
|
||
|
|
rule = load_rule(conn, "feed_failed")
|
||
|
|
assert rule is not None
|
||
|
|
got = {r.address for r in Resolver(conn, s, ndir).resolve(rule, None)}
|
||
|
|
assert "[email protected]" in got and "[email protected]" in got
|
||
|
|
assert "[email protected]" not in got
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- delivery
|
||
|
|
|
||
|
|
|
||
|
|
def test_immediate_send_through_the_relay(s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
eid = raise_event(
|
||
|
|
conn, "feed_failed", subject="visual-feed", title="visual-feed run failed", body="boom", condition=True
|
||
|
|
)
|
||
|
|
counts = send.dispatch_pending(conn, s, ndir)
|
||
|
|
assert counts == {"sent": 1}
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT dispatched_at, dispatch_result FROM notification_events WHERE id = %s", (eid,)
|
||
|
|
).fetchone()
|
||
|
|
assert row[0] is not None and row[1] == "sent"
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {}
|
||
|
|
(m,) = relay.messages
|
||
|
|
assert m.tls and m.header("Subject") == "[Packet Station TEST] visual-feed run failed"
|
||
|
|
assert m.header("To") == "[email protected]"
|
||
|
|
assert m.rcpts == ["[email protected]"]
|
||
|
|
assert "boom" in m.text and "Feed run failed" in m.text
|
||
|
|
(rec,) = sent_since(s, marker)
|
||
|
|
assert rec["result"] == "sent" and rec["event_ids"] == [eid] and not rec["digest"] and not rec["test"]
|
||
|
|
assert rec["recipients"] == [
|
||
|
|
{"address": "[email protected]", "user_id": str(ndir.user("admin.ann").guid), "why": "group PS-Admins"}
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_recovery_notice_goes_to_the_same_rule(s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
raise_event(conn, "backup_stale", subject="prod", title="Backup prod is stale", condition=True)
|
||
|
|
send.dispatch_pending(conn, s, ndir)
|
||
|
|
resolve_event(conn, "backup_stale", subject="prod", body="A fresh dump is there.")
|
||
|
|
raise_event(conn, "backup_stale", subject="prod", title="Backup prod is stale", condition=True)
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"sent": 2}
|
||
|
|
assert [m.header("Subject") for m in relay.messages] == [
|
||
|
|
"[Packet Station TEST] Backup prod is stale",
|
||
|
|
"[Packet Station TEST] Cleared: Backup prod is stale",
|
||
|
|
"[Packet Station TEST] Backup prod is stale",
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_dry_run_records_and_test_send_still_sends(
|
||
|
|
s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir, ids: dict[str, str]
|
||
|
|
) -> None:
|
||
|
|
dry = s.model_copy(update={"notify_dry_run": True})
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
eid = raise_event(conn, "version_deployed", subject="build:x", title="Version x deployed")
|
||
|
|
assert send.dispatch_pending(conn, dry, ndir) == {"dry_run": 1}
|
||
|
|
assert relay.messages == []
|
||
|
|
assert conn.execute("SELECT dispatch_result FROM notification_events WHERE id = %s", (eid,)).fetchone()[0] == (
|
||
|
|
"dry_run"
|
||
|
|
)
|
||
|
|
out = send.send_test(conn, dry, ndir, "version_deployed", requested_by=ids["admin.ann"])
|
||
|
|
assert out["result"] == "sent" and out["error"] is None
|
||
|
|
assert out["subject"] == "[Packet Station TEST] Test: Version deployed"
|
||
|
|
assert [r["address"] for r in out["recipients"]] == ["[email protected]"]
|
||
|
|
to = send.send_test(conn, dry, ndir, "version_deployed", to="[email protected]")
|
||
|
|
assert to["result"] == "sent" and to["recipients"] == [
|
||
|
|
{"address": "[email protected]", "user_id": None, "why": "test address"}
|
||
|
|
]
|
||
|
|
assert [m.rcpts for m in relay.messages] == [["[email protected]"], ["[email protected]"]]
|
||
|
|
rows = sent_since(s, marker)
|
||
|
|
assert [(r["result"], r["test"]) for r in rows] == [("dry_run", False), ("sent", True), ("sent", True)]
|
||
|
|
assert str(rows[1]["requested_by"]) == ids["admin.ann"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_mail_off_records_failed(s: Settings, marker: int, ndir: NotifyDir) -> None:
|
||
|
|
off = s.model_copy(update={"smtp_host": ""})
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
out = send.send_test(conn, off, ndir, "feed_failed")
|
||
|
|
assert out["result"] == "failed" and out["error"] == "Mail is off: SMTP_HOST is empty."
|
||
|
|
eid = raise_event(conn, "feed_failed", subject="drawing-lookup", title="t", condition=True)
|
||
|
|
assert send.dispatch_pending(conn, off, ndir) == {"failed": 1}
|
||
|
|
# Recorded and not retried
|
||
|
|
assert conn.execute("SELECT dispatched_at FROM notification_events WHERE id = %s", (eid,)).fetchone()[0]
|
||
|
|
assert [(r["result"], r["error"]) for r in sent_since(s, marker)] == [
|
||
|
|
("failed", "Mail is off: SMTP_HOST is empty."),
|
||
|
|
("failed", "Mail is off: SMTP_HOST is empty."),
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_rule_off_no_recipients_and_digest_queue(s: Settings, marker: int, ndir: NotifyDir) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
conn.execute("UPDATE notification_rules SET enabled = false WHERE event_type = 'group_missing'")
|
||
|
|
conn.execute("UPDATE notification_rules SET delivery = 'digest' WHERE event_type = 'ad_reconcile_report'")
|
||
|
|
conn.execute("DELETE FROM notification_recipients WHERE event_type = 'backup_stale'")
|
||
|
|
a = raise_event(conn, "group_missing", subject="PS-X", title="t", condition=True)
|
||
|
|
b = raise_event(conn, "ad_reconcile_report", subject="ad", title="t")
|
||
|
|
c = raise_event(conn, "backup_stale", subject="prod", title="t", condition=True)
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"rule_off": 1, "queued": 1, "no_recipients": 1}
|
||
|
|
res = dict(conn.execute("SELECT id, dispatch_result FROM notification_events").fetchall())
|
||
|
|
assert res == {a: "rule_off", b: "queued", c: "no_recipients"}
|
||
|
|
assert conn.execute("SELECT event_id FROM notification_outbox").fetchall() == [(b,)]
|
||
|
|
assert [r["result"] for r in sent_since(s, marker)] == ["no_recipients"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_directory_down_retries_then_fails(s: Settings, marker: int, ndir: NotifyDir) -> None:
|
||
|
|
ndir.down = True
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
eid = raise_event(conn, "integrity_failure", subject="integrity", title="t", condition=True)
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"retry": 1}
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"retry": 1}
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT dispatched_at, dispatch_result FROM notification_events WHERE id = %s", (eid,)
|
||
|
|
).fetchone()
|
||
|
|
assert row == (None, "directory_retry:2")
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"failed": 1}
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT dispatched_at, dispatch_result FROM notification_events WHERE id = %s", (eid,)
|
||
|
|
).fetchone()
|
||
|
|
assert row[0] is not None and row[1] == "failed"
|
||
|
|
rows = sent_since(s, marker)
|
||
|
|
assert [(r["result"], r["error"]) for r in rows] == [("failed", "directory unavailable")] * 3
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- digest
|
||
|
|
|
||
|
|
|
||
|
|
def test_digest_one_message_per_address(
|
||
|
|
s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir, ids: dict[str, str]
|
||
|
|
) -> None:
|
||
|
|
tz = ZoneInfo(s.site_tz)
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
for et in ("ad_reconcile_report", "group_missing"):
|
||
|
|
conn.execute(
|
||
|
|
"UPDATE notification_rules SET delivery = 'digest', digest_time = '07:00' WHERE event_type = %s", (et,)
|
||
|
|
)
|
||
|
|
set_recipients(conn, "group_missing", [{"kind": "user", "user_id": ids["pm.paula"]}])
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO notification_recipients (event_type, kind, user_id)"
|
||
|
|
" VALUES ('ad_reconcile_report', 'user', %s)",
|
||
|
|
(ids["pm.paula"],),
|
||
|
|
)
|
||
|
|
e1 = raise_event(conn, "ad_reconcile_report", subject="ad", title="Reconcile: 2 added")
|
||
|
|
e2 = raise_event(conn, "group_missing", subject="PS-A", title="Group PS-A missing", condition=True)
|
||
|
|
e3 = raise_event(conn, "group_missing", subject="PS-B", title="Group PS-B missing", condition=True)
|
||
|
|
assert send.dispatch_pending(conn, s, ndir) == {"queued": 3}
|
||
|
|
# Queued at "now"; a digest time later today has not passed yet for them
|
||
|
|
now = datetime.now(tz)
|
||
|
|
conn.execute("UPDATE notification_outbox SET queued_at = %s", (now - timedelta(days=1),))
|
||
|
|
before = datetime.combine(now.date(), datetime.min.time(), tzinfo=tz) + timedelta(hours=6, minutes=59)
|
||
|
|
assert digest.run_digest(conn, s, ndir, before) == {"due": 0}
|
||
|
|
at = before + timedelta(minutes=2)
|
||
|
|
counts = digest.run_digest(conn, s, ndir, at)
|
||
|
|
assert counts["messages"] == 2 and counts["sent"] == 2 and counts["left"] == 0
|
||
|
|
assert conn.execute("SELECT count(*) FROM notification_outbox WHERE sent_at IS NULL").fetchone()[0] == 0
|
||
|
|
assert digest.run_digest(conn, s, ndir, at) == {"due": 0}
|
||
|
|
by = {tuple(m.rcpts): m for m in relay.messages}
|
||
|
|
assert set(by) == {("[email protected]",), ("[email protected]",)}
|
||
|
|
# Paula is on both rules: one message with all three. Ann (PS-Admins, the seed) only on the reconcile rule.
|
||
|
|
paula = by[("[email protected]",)]
|
||
|
|
assert paula.header("Subject") == "[Packet Station TEST] Daily digest: 3 notifications"
|
||
|
|
for t in ("Reconcile: 2 added", "Group PS-A missing", "Group PS-B missing"):
|
||
|
|
assert t in paula.text
|
||
|
|
ann = by[("[email protected]",)]
|
||
|
|
assert ann.header("Subject") == "[Packet Station TEST] Daily digest: 1 notification"
|
||
|
|
assert "Reconcile: 2 added" in ann.text and "PS-A" not in ann.text
|
||
|
|
rows = sent_since(s, marker)
|
||
|
|
assert len(rows) == 2 and all(r["digest"] and r["event_type"] == "digest" for r in rows)
|
||
|
|
got = {r["recipients"][0]["address"]: sorted(r["event_ids"]) for r in rows}
|
||
|
|
assert got == {"[email protected]": [e1], "[email protected]": sorted([e1, e2, e3])}
|
||
|
|
|
||
|
|
|
||
|
|
def test_digest_failure_leaves_rows_and_skips_done_addresses(
|
||
|
|
s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir, ids: dict[str, str]
|
||
|
|
) -> None:
|
||
|
|
tz = ZoneInfo(s.site_tz)
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
conn.execute(
|
||
|
|
"UPDATE notification_rules SET delivery = 'digest', digest_time = '00:00'"
|
||
|
|
" WHERE event_type = 'ad_reconcile_report'"
|
||
|
|
)
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO notification_recipients (event_type, kind, address) VALUES"
|
||
|
|
" ('ad_reconcile_report', 'mailbox', '[email protected]')"
|
||
|
|
)
|
||
|
|
raise_event(conn, "ad_reconcile_report", subject="ad", title="Reconcile")
|
||
|
|
send.dispatch_pending(conn, s, ndir)
|
||
|
|
now = datetime.now(tz)
|
||
|
|
conn.execute("UPDATE notification_outbox SET queued_at = %s", (now - timedelta(days=1),))
|
||
|
|
at = now
|
||
|
|
relay.refuse.add("[email protected]")
|
||
|
|
try:
|
||
|
|
counts = digest.run_digest(conn, s, ndir, at)
|
||
|
|
assert counts["sent"] == 1 and counts["failed"] == 1 and counts["left"] == 1
|
||
|
|
assert conn.execute("SELECT count(*) FROM notification_outbox WHERE sent_at IS NULL").fetchone()[0] == 1
|
||
|
|
# Backoff after a failure
|
||
|
|
assert digest.run_digest(conn, s, ndir, at + timedelta(minutes=5))["backoff"] is True
|
||
|
|
finally:
|
||
|
|
relay.refuse.clear()
|
||
|
|
counts = digest.run_digest(conn, s, ndir, at + timedelta(minutes=31))
|
||
|
|
assert counts["messages"] == 1 and counts["sent"] == 1 # Ann already has it
|
||
|
|
assert conn.execute("SELECT count(*) FROM notification_outbox WHERE sent_at IS NULL").fetchone()[0] == 0
|
||
|
|
assert [m.rcpts for m in relay.messages] == [["[email protected]"], ["[email protected]"]]
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- watch checks, version, feed hook
|
||
|
|
|
||
|
|
|
||
|
|
def _packet_with_co(conn: Any, dept: str, days: int) -> int:
|
||
|
|
pid = conn.execute("INSERT INTO projects (number, name) VALUES ('L77001', 'P') RETURNING id").fetchone()[0]
|
||
|
|
uid = conn.execute("INSERT INTO units (project_id, lot) VALUES (%s, '001') RETURNING id", (pid,)).fetchone()[0]
|
||
|
|
pk = conn.execute(
|
||
|
|
"INSERT INTO packets (unit_id, dept_code, title) VALUES (%s, %s, 'Nozzle') RETURNING id", (uid, dept)
|
||
|
|
).fetchone()[0]
|
||
|
|
co = conn.execute(
|
||
|
|
"INSERT INTO change_orders (packet_id, co_no, origin, issued_at) VALUES (%s, 1, 'manual',"
|
||
|
|
" now() - make_interval(days => %s)) RETURNING id",
|
||
|
|
(pk, days),
|
||
|
|
).fetchone()[0]
|
||
|
|
return int(co)
|
||
|
|
|
||
|
|
|
||
|
|
def test_watch_drawings_and_change_orders(s: Settings, marker: int, ndir: NotifyDir) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
seed(conn, s)
|
||
|
|
conn.execute(
|
||
|
|
"INSERT INTO drawing_lookups (dwg, ok, fail_count, failing_since, note) VALUES"
|
||
|
|
" ('D-1', false, 9, now() - interval '30 hours', 'not found'),"
|
||
|
|
" ('D-2', false, 2, now() - interval '2 hours', 'timeout')"
|
||
|
|
)
|
||
|
|
assert watch.check_drawings(conn) == {"drawings_failing": 1}
|
||
|
|
title, body, dept = conn.execute(
|
||
|
|
"SELECT title, body, dept_code FROM notification_events WHERE kind = 'drawing_failing'"
|
||
|
|
).fetchone()
|
||
|
|
assert title == "1 drawing lookup failing for more than 24 hours" and "D-1" in body and "D-2" not in body
|
||
|
|
conn.execute("UPDATE notification_rules SET param = 1 WHERE event_type = 'drawing_failing'")
|
||
|
|
assert watch.check_drawings(conn) == {"drawings_failing": 2}
|
||
|
|
assert (
|
||
|
|
conn.execute("SELECT count(*) FROM notification_events WHERE kind = 'drawing_failing'").fetchone()[0] == 1
|
||
|
|
)
|
||
|
|
conn.execute("UPDATE drawing_lookups SET failing_since = NULL")
|
||
|
|
assert watch.check_drawings(conn) == {"drawings_failing": 0, "drawings_cleared": True}
|
||
|
|
|
||
|
|
co = _packet_with_co(conn, "04", 5)
|
||
|
|
out = watch.check_change_orders(conn)
|
||
|
|
assert out == {"co_waiting": 1, "co_raised": 1, "co_cleared": 0}
|
||
|
|
subj, dept, title = conn.execute(
|
||
|
|
"SELECT subject, dept_code, title FROM notification_events WHERE kind = 'co_waiting'"
|
||
|
|
).fetchone()
|
||
|
|
assert subj == f"co:{co}" and dept == "04" and title.startswith("Change order CO-1 waiting 5 days")
|
||
|
|
assert watch.check_change_orders(conn)["co_raised"] == 0
|
||
|
|
conn.execute("UPDATE change_orders SET review = 'approved' WHERE id = %s", (co,))
|
||
|
|
assert watch.check_change_orders(conn) == {"co_waiting": 0, "co_raised": 0, "co_cleared": 1}
|
||
|
|
rec = conn.execute(
|
||
|
|
"SELECT dept_code, title FROM notification_events WHERE kind = 'co_waiting' AND recovery_of IS NOT NULL"
|
||
|
|
).fetchone()
|
||
|
|
assert rec[0] == "04" and rec[1].startswith("Cleared: Change order CO-1")
|
||
|
|
conn.execute("DELETE FROM change_orders")
|
||
|
|
conn.execute("DELETE FROM packets")
|
||
|
|
|
||
|
|
|
||
|
|
def test_note_build(s: Settings, marker: int) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
first = watch.note_build(conn, "sha-aaa")
|
||
|
|
assert first is not None
|
||
|
|
assert watch.note_build(conn, "sha-aaa") is None
|
||
|
|
second = watch.note_build(conn, "sha-bbb")
|
||
|
|
assert second is not None
|
||
|
|
body = conn.execute("SELECT body FROM notification_events WHERE id = %s", (second,)).fetchone()[0]
|
||
|
|
assert "sha-bbb" in body and "sha-aaa" in body
|
||
|
|
value = conn.execute("SELECT value FROM settings WHERE key = 'notify.last_build'").fetchone()[0]
|
||
|
|
assert value == {"build": "sha-bbb"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_feed_run_hook(s: Settings, marker: int) -> None:
|
||
|
|
def open_kinds(conn: Any) -> set[tuple[str, str]]:
|
||
|
|
return {
|
||
|
|
(r[0], r[1])
|
||
|
|
for r in conn.execute("SELECT kind, subject FROM notification_events WHERE resolved_at IS NULL").fetchall()
|
||
|
|
}
|
||
|
|
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
watch.feed_run(conn, "visual-feed", "failed", ["VisualError: down"], 7)
|
||
|
|
watch.feed_run(conn, "drawing-lookup", "failed", [], 8)
|
||
|
|
assert open_kinds(conn) == {("feed_failed", "visual-feed"), ("feed_failed", "drawing-lookup")}
|
||
|
|
watch.feed_run(conn, "visual-feed", "degraded", ["one row bad"], 9)
|
||
|
|
assert open_kinds(conn) == {("feed_degraded", "visual-feed"), ("feed_failed", "drawing-lookup")}
|
||
|
|
watch.feed_run(conn, "drawing-lookup", "degraded", ["x"], 10)
|
||
|
|
watch.feed_run(conn, "visual-feed", "ok", [], 11)
|
||
|
|
assert open_kinds(conn) == set()
|
||
|
|
watch.feed_run(conn, "visual-feed", "skipped", [], 12)
|
||
|
|
watch.feed_run(conn, "housekeeping", "failed", [], 13)
|
||
|
|
assert open_kinds(conn) == set()
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------- the jobs and the quota
|
||
|
|
|
||
|
|
|
||
|
|
def test_notify_send_job(s: Settings, marker: int, relay: FakeSmtp, ndir: NotifyDir) -> None:
|
||
|
|
from packetstation.worker.jobs import notify_digest, notify_send
|
||
|
|
|
||
|
|
assert notify_send.SPEC.name == "notify-send" and not notify_send.SPEC.pausable
|
||
|
|
assert notify_digest.SPEC.name == "notify-digest" and not notify_digest.SPEC.pausable
|
||
|
|
build = s.model_copy(update={"ps_build": "sha-job"})
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
ctx = ctx_for(conn, build, ndir)
|
||
|
|
counts = notify_send.run(ctx)
|
||
|
|
assert counts["seeded"] == 13 and counts["version_deployed"] == "sha-job" and counts["sent"] == 1
|
||
|
|
assert counts["drawings_failing"] == 0 and counts["co_waiting"] == 0
|
||
|
|
again = notify_send.run(ctx)
|
||
|
|
assert "version_deployed" not in again and "drawings_failing" not in again # watch waits 15 minutes
|
||
|
|
assert notify_digest.run(ctx_for(conn, build, ndir)) == {"due": 0}
|
||
|
|
assert [m.header("Subject") for m in relay.messages] == ["[Packet Station TEST] Version sha-job deployed"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_check_quota_through_notify(s: Settings, marker: int, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
|
|
from packetstation.domain import assets
|
||
|
|
|
||
|
|
warn = {"total": 100, "used": 90, "free": 10, "pct": 90.0, "warn": True}
|
||
|
|
ok = {"total": 100, "used": 10, "free": 90, "pct": 10.0, "warn": False}
|
||
|
|
monkeypatch.setattr(assets, "usage_for", lambda *a, **k: dict(warn))
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
assert assets.check_quota(conn, s)["alert"] == "opened"
|
||
|
|
assert assets.check_quota(conn, s)["alert"] is None
|
||
|
|
row = conn.execute(
|
||
|
|
"SELECT event_type, title, detail, raised_count FROM notification_events WHERE kind = 'assets_quota'"
|
||
|
|
).fetchone()
|
||
|
|
assert row[0] == "assets_quota" and row[1] == "Assets storage is 90.0% full" and row[2]["pct"] == 90.0
|
||
|
|
assert row[3] == 1 # check_quota raises only when nothing is open
|
||
|
|
monkeypatch.setattr(assets, "usage_for", lambda *a, **k: dict(ok))
|
||
|
|
out = assets.check_quota(conn, s)
|
||
|
|
assert out["alert"] == "resolved" and out["pct"] == 10.0
|
||
|
|
assert (
|
||
|
|
conn.execute(
|
||
|
|
"SELECT count(*) FROM notification_events WHERE kind = 'assets_quota' AND recovery_of IS NOT NULL"
|
||
|
|
).fetchone()[0]
|
||
|
|
== 1
|
||
|
|
)
|
||
|
|
assert assets.check_quota(conn, s)["alert"] is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_digest_json_detail_is_kept(s: Settings, marker: int) -> None:
|
||
|
|
with app_conn(s) as conn:
|
||
|
|
eid = raise_event(conn, "ad_reconcile_report", subject="ad", title="t", detail={"when": datetime(2026, 1, 2)})
|
||
|
|
got = conn.execute("SELECT detail FROM notification_events WHERE id = %s", (eid,)).fetchone()[0]
|
||
|
|
assert got == {"when": "2026-01-02 00:00:00"}
|