tests/e2e/test_notifications_e2e.py
Rama dev · réplica de lectura del repositorio original.
"""E2E: the webhook, end to end — the real app, a real run against the fake Plex, and a receiver on loopback.
Unit tests prove each rule; this proves the wiring that only a running app has: the settings PUT and its
validation, the encrypted auth header reaching the wire, the run and the job queue queuing their events,
the queue draining them, and every message that leaves naming nobody.
"""
from __future__ import annotations
import json
import re
import threading
import time
from collections.abc import Iterator
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import pytest
from playwright.sync_api import Page, expect
from tests.e2e.conftest import ShortlistApp
pytestmark = pytest.mark.e2e
AUTH_NAME, AUTH_VALUE = "X-Test-Key", "k3y-for-the-receiver"
@pytest.fixture
def receiver() -> Iterator[tuple[str, list[dict]]]:
"""A webhook receiver on loopback that records each POST's headers and JSON body."""
received: list[dict] = []
class Receiver(BaseHTTPRequestHandler):
def do_POST(self) -> None:
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
received.append({"headers": {k.lower(): v for k, v in self.headers.items()}, "body": body})
self.send_response(204)
self.end_headers()
def log_message(self, format: str, *args: object) -> None:
pass
server = ThreadingHTTPServer(("127.0.0.1", 0), Receiver)
threading.Thread(target=server.serve_forever, daemon=True).start()
try:
yield f"http://127.0.0.1:{server.server_address[1]}/hook", received
finally:
server.shutdown()
def _events(received: list[dict]) -> list[str]:
return [r["body"]["event"] for r in received]
def _wait_for(received: list[dict], wanted: set[str], timeout_s: float = 90) -> None:
"""The queue drains at the end of each run and job, and on the worker's 60-second tick."""
deadline = time.monotonic() + timeout_s
while time.monotonic() < deadline:
if wanted <= set(_events(received)):
return
time.sleep(0.2)
raise AssertionError(f"never received {sorted(wanted - set(_events(received)))}; got {_events(received)}")
def _configure(app: ShortlistApp, url: str, events: list[str]) -> None:
resp = app.api(
"PUT",
"/api/settings",
json={
"values": {
"notify.webhook.enabled": True,
"notify.webhook.url": url,
"notify.webhook.events": events,
"notify.webhook.auth_header_name": AUTH_NAME,
"notify.webhook.auth_header_value": AUTH_VALUE,
}
},
)
assert resp.status_code == 200, resp.text
class TestTheWebhookEndToEnd:
def test_a_run_and_a_job_reach_the_receiver_with_the_auth_header_and_no_names(self, app: ShortlistApp, receiver):
url, received = receiver
_configure(app, url, ["run.started", "run.finished", "run.partial", "job.started", "job.finished"])
assert app.api("GET", "/api/settings").json()["notify.webhook.auth_header_value"] == "•••••"
refused = app.api("PUT", "/api/settings", json={"values": {"notify.webhook.auth_header_value": "key\t"}})
assert refused.status_code == 422
test = app.api("POST", "/api/settings/test/notify").json()
assert test["ok"], test
_wait_for(received, {"test"})
job = app.api("POST", "/api/system/jobs", json={"kind": "backup.take", "payload": {}}).json()
assert job["status"] == "done", job
created = app.api("POST", "/api/runs", json={"dry_run": False}).json()
run = app.wait_for_run(created["run_id"])
assert run["status"] == "ok", run
_wait_for(received, {"job.started", "job.finished", "run.started"})
_wait_for(received, {"run.finished"} if not run["stats"].get("users_error") else {"run.partial"})
usernames = {u["username"].lower() for u in app.api("GET", "/api/users").json()}
assert usernames, "the fake server has people, so the no-names check has teeth"
for message in received:
assert message["headers"][AUTH_NAME.lower()] == AUTH_VALUE
body = message["body"]
assert body["content"] == body["text"] == f"{body['title']}\n{body['message']}"
said = f"{body['title']} {body['message']}".lower()
assert not [name for name in usernames if name in said], body
def test_a_dry_run_sends_nothing(self, app: ShortlistApp, receiver):
url, received = receiver
events = ["run.started", "run.finished", "run.partial", "run.failed"]
_configure(app, url, events)
# Positive control: this config does reach the receiver, so an empty receiver below is not a dead setup.
assert app.api("POST", "/api/settings/test/notify").json()["ok"]
_wait_for(received, {"test"})
received.clear()
created = app.api("POST", "/api/runs", json={"dry_run": True}).json()
app.wait_for_run(created["run_id"])
# Nothing is even queued: a dry run never reaches `notify.send`, so the queue is the evidence
# rather than a pause to see whether something arrives. `run.started` is queued before the run
# begins and `run.finished` after it persists; neither exists for a dry run.
queued = app.api("GET", "/api/system/jobs", params={"kind": "notify.send", "limit": 200}).json()
assert queued == []
assert received == []
# Same config, same app: a real run does queue `notify.send`, so the empty queue above has teeth.
real = app.api("POST", "/api/runs", json={"dry_run": False}).json()
app.wait_for_run(real["run_id"])
queued = app.api("GET", "/api/system/jobs", params={"kind": "notify.send", "limit": 200}).json()
assert queued, "a non-dry run with this config queues notify.send"
_wait_for(received, {"run.started"})
paths = {message["body"].get("path") for message in received}
assert f"/runs/{real['run_id']}" in paths
assert f"/runs/{created['run_id']}" not in paths
def test_the_settings_card_lists_the_events_and_ticks_the_chosen_ones(page: Page, app: ShortlistApp):
resp = app.api(
"PUT",
"/api/settings",
json={"values": {"notify.webhook.enabled": True, "notify.webhook.events": ["run.failed", "job.failed"]}},
)
assert resp.status_code == 200, resp.text
page.goto("/settings#notifications")
expect(page.get_by_text("What to send")).to_be_visible(timeout=20_000)
expect(page.get_by_role("checkbox", name="A run failed")).to_be_checked()
expect(page.get_by_role("checkbox", name="A job failed")).to_be_checked()
expect(page.get_by_role("checkbox", name="A run started")).not_to_be_checked()
page.get_by_role("checkbox", name="A run started").click()
expect(page.get_by_role("checkbox", name="A run started")).to_be_checked()
app.wait_for_setting("notify.webhook.events", ["run.started", "run.failed", "job.failed"])
def test_the_webhook_is_set_up_and_removed_from_its_connection_card(page: Page, app: ShortlistApp):
"""The address and the auth header live on one card in Connections, with Test and Remove like every
other service — removing it is the way to take the header off along with the address."""
page.goto("/settings")
card = page.get_by_test_id("connection-notify")
expect(card).to_be_visible(timeout=20_000)
card.get_by_role("button", name="Set up").click()
card.get_by_label("Address").fill("https://hooks.example.com/shortlist/abc")
card.get_by_label("Header name").fill("X-Api-Key")
card.get_by_label("Header value").fill("k3y-value")
card.get_by_role("button", name="Save", exact=True).click()
expect(card.get_by_text("Address and auth header saved").filter(visible=True)).to_be_visible()
# Saving sends nothing yet: the card says so and links to where it is switched on.
expect(card.get_by_text("Not sending yet", exact=False)).to_be_visible()
card.get_by_role("link", name="Notifications").click()
expect(page).to_have_url(re.compile(r"#notifications$"))
expect(page.get_by_role("switch", name="Send alerts to a webhook")).to_be_in_viewport()
saved = app.api("GET", "/api/settings").json()
assert saved["notify.webhook.url"] == "•••••"
assert saved["notify.webhook.auth_header_name"] == "X-Api-Key"
assert saved["notify.webhook.auth_header_value"] == "•••••"
page.goto("/settings#connections")
card.get_by_role("button", name="Remove Webhook connection").click()
card.get_by_role("button", name="Remove", exact=True).click()
expect(card.get_by_role("button", name="Set up")).to_be_visible()
cleared = app.api("GET", "/api/settings").json()
assert (cleared["notify.webhook.url"], cleared["notify.webhook.auth_header_value"]) == ("", "")
assert cleared["notify.webhook.auth_header_name"] == ""