234 lines
7.5 KiB
Python
234 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from datetime import UTC, datetime
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from backup_tool.db.models import (
|
|
NotificationDelivery,
|
|
NotificationEmailSettings,
|
|
NotificationEvent,
|
|
)
|
|
from backup_tool.ids import new_uuid7
|
|
from backup_tool.notifications.email import SMTPClient, deliver_email
|
|
from backup_tool.notifications.events import emit_event
|
|
from sqlalchemy import select
|
|
|
|
PASSWORD = "correct horse battery staple"
|
|
|
|
|
|
async def _setup(client) -> str:
|
|
response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
|
|
assert response.status_code == 201
|
|
return client.cookies["backup_tool_csrf"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filters_manual_test_retry_and_history(app_client) -> None:
|
|
client, _ = app_client
|
|
csrf = await _setup(client)
|
|
created = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["schedule.*", "notification.test_requested"],
|
|
"destination": {"recipients": ["operator@example.test"]},
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert created.status_code == 201
|
|
subscription_id = created.json()["id"]
|
|
app = client._transport.app
|
|
async with app.state.sessions() as db:
|
|
await emit_event(
|
|
db,
|
|
"execution.queued",
|
|
correlation_id=str(new_uuid7()),
|
|
resource={},
|
|
deduplication_key="filtered-out",
|
|
)
|
|
scheduled = await emit_event(
|
|
db,
|
|
"schedule.created",
|
|
correlation_id=str(new_uuid7()),
|
|
resource={},
|
|
deduplication_key="filtered-in",
|
|
)
|
|
await db.commit()
|
|
statement = select(NotificationDelivery).where(
|
|
NotificationDelivery.event_id == scheduled.id
|
|
)
|
|
deliveries = list((await db.scalars(statement)).all())
|
|
assert len(deliveries) == 1
|
|
delivery = deliveries[0]
|
|
delivery.state = "failed"
|
|
delivery.terminal_reason = "http_permanent"
|
|
await db.commit()
|
|
delivery_id = delivery.id
|
|
|
|
tested = await client.post(
|
|
f"/api/v2/notifications/subscriptions/{subscription_id}/test",
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "test-one"},
|
|
)
|
|
assert tested.status_code == 202
|
|
retried = await client.post(
|
|
f"/api/v2/notifications/deliveries/{delivery_id}/retry",
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "retry-one"},
|
|
)
|
|
assert retried.status_code == 202
|
|
replayed = await client.post(
|
|
f"/api/v2/notifications/deliveries/{delivery_id}/retry",
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "retry-one"},
|
|
)
|
|
assert replayed.status_code == 202
|
|
assert replayed.json() == retried.json()
|
|
history = await client.get("/api/v2/notifications/deliveries")
|
|
assert history.status_code == 200
|
|
row = next(item for item in history.json()["items"] if item["id"] == delivery_id)
|
|
assert row["state"] == "retry"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_manual_test_bypasses_filters_and_targets_only_selected_subscription(
|
|
app_client,
|
|
) -> None:
|
|
client, _ = app_client
|
|
csrf = await _setup(client)
|
|
selected = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["execution.failed"],
|
|
"destination": {"recipients": ["selected@example.test"]},
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
other = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["notification.test_requested"],
|
|
"destination": {"recipients": ["other@example.test"]},
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
assert selected.status_code == other.status_code == 201
|
|
response = await client.post(
|
|
f"/api/v2/notifications/subscriptions/{selected.json()['id']}/test",
|
|
headers={"X-CSRF-Token": csrf, "Idempotency-Key": "selected-test"},
|
|
)
|
|
assert response.status_code == 202
|
|
app = client._transport.app
|
|
async with app.state.sessions() as db:
|
|
rows = list(
|
|
(
|
|
await db.scalars(
|
|
select(NotificationDelivery.subscription_id).where(
|
|
NotificationDelivery.event_id == response.json()["event_id"]
|
|
)
|
|
)
|
|
).all()
|
|
)
|
|
assert rows == [selected.json()["id"]]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_subscription_does_not_receive_future_events(app_client) -> None:
|
|
client, _ = app_client
|
|
csrf = await _setup(client)
|
|
created = await client.post(
|
|
"/api/v2/notifications/subscriptions",
|
|
json={
|
|
"channel": "email",
|
|
"event_filters": ["execution.*"],
|
|
"destination": {"recipients": ["operator@example.test"]},
|
|
},
|
|
headers={"X-CSRF-Token": csrf},
|
|
)
|
|
subscription = created.json()
|
|
disabled = await client.patch(
|
|
f"/api/v2/notifications/subscriptions/{subscription['id']}",
|
|
json={"state": "disabled"},
|
|
headers={"X-CSRF-Token": csrf, "If-Match": created.headers["ETag"]},
|
|
)
|
|
assert disabled.status_code == 200
|
|
app = client._transport.app
|
|
async with app.state.sessions() as db:
|
|
event = await emit_event(
|
|
db,
|
|
"execution.queued",
|
|
correlation_id=str(new_uuid7()),
|
|
resource={},
|
|
occurred_at=datetime.now(UTC),
|
|
)
|
|
await db.commit()
|
|
assert (
|
|
await db.scalar(
|
|
select(NotificationDelivery.id).where(NotificationDelivery.event_id == event.id)
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_email_uses_ehlo_starttls_then_auth_with_hermetic_fake() -> None:
|
|
calls: list[str] = []
|
|
|
|
class FakeSMTP:
|
|
def __init__(self, *_args, **_kwargs) -> None:
|
|
calls.append("connect")
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args) -> None:
|
|
calls.append("close")
|
|
|
|
def ehlo(self) -> None:
|
|
calls.append("ehlo")
|
|
|
|
def starttls(self, *, context) -> None:
|
|
assert context.check_hostname
|
|
calls.append("starttls")
|
|
|
|
def login(self, username: str, password: str) -> None:
|
|
assert username == "operator"
|
|
assert password == "smtp-password"
|
|
calls.append("auth")
|
|
|
|
def send_message(self, message) -> None:
|
|
assert "smtp-password" not in message.as_string()
|
|
calls.append("send")
|
|
|
|
event = NotificationEvent(
|
|
id=str(new_uuid7()),
|
|
type="execution.queued",
|
|
schema_version=1,
|
|
occurred_at=datetime.now(UTC),
|
|
correlation_id=str(new_uuid7()),
|
|
severity="info",
|
|
resource_refs={},
|
|
payload={},
|
|
canonical_envelope="{}",
|
|
)
|
|
settings = NotificationEmailSettings(
|
|
id=1,
|
|
host="smtp.example.test",
|
|
port=587,
|
|
username="operator",
|
|
password_secret_id=str(new_uuid7()),
|
|
sender="sender@example.test",
|
|
max_attempts=5,
|
|
rate_limit_per_minute=60,
|
|
)
|
|
result = await deliver_email(
|
|
settings,
|
|
"smtp-password",
|
|
event,
|
|
["operator@example.test"],
|
|
smtp_factory=cast(Callable[..., SMTPClient], FakeSMTP),
|
|
)
|
|
assert result.response_class == "smtp_2xx"
|
|
assert calls == ["connect", "ehlo", "starttls", "ehlo", "auth", "send", "close"]
|