saksham
1a4c80958f
FastAPI + Celery + Next.js + Postgres/Redis app with company monitoring, source collection, LLM-based change analysis, enrichment, and account security (Turnstile, escalating lockout, email verification).
122 lines
3.6 KiB
Python
122 lines
3.6 KiB
Python
"""POST /notification-destinations/{id}/test - dedicated from
|
|
test_notification_destinations.py since it exercises actual send-path
|
|
dispatch (console/SMTP) rather than just CRUD."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
|
|
def _register_and_login(client) -> dict[str, str]:
|
|
email = f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
tokens = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
|
|
|
|
|
def _create_company(client, headers) -> dict:
|
|
return client.post(
|
|
"/api/v1/companies",
|
|
json={"name": f"Co {uuid.uuid4().hex[:6]}", "frequency_type": "weekly"},
|
|
headers=headers,
|
|
).json()
|
|
|
|
|
|
def test_test_endpoint_sends_to_email_destination(client, monkeypatch):
|
|
sent = {}
|
|
|
|
class FakeSmtp:
|
|
def __init__(self, host, port, timeout=10):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
def starttls(self):
|
|
pass
|
|
|
|
def login(self, username, password):
|
|
pass
|
|
|
|
def sendmail(self, from_addr, to_addrs, message):
|
|
sent["to_addrs"] = to_addrs
|
|
|
|
monkeypatch.setattr("app.notifications.smtp_email.smtplib.SMTP", FakeSmtp)
|
|
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers)
|
|
destination = client.post(
|
|
"/api/v1/notification-destinations",
|
|
json={
|
|
"type": "email",
|
|
"destination_value": "[email protected]",
|
|
"company_ids": [company["id"]],
|
|
},
|
|
headers=headers,
|
|
).json()
|
|
|
|
resp = client.post(
|
|
f"/api/v1/notification-destinations/{destination['id']}/test", headers=headers
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["success"] is True
|
|
assert body["error"] is None
|
|
assert sent["to_addrs"] == ["[email protected]"]
|
|
|
|
|
|
def test_test_endpoint_reports_failure(client, monkeypatch):
|
|
class FailingSmtp:
|
|
def __init__(self, host, port, timeout=10):
|
|
raise ConnectionRefusedError("no mailpit running")
|
|
|
|
monkeypatch.setattr("app.notifications.smtp_email.smtplib.SMTP", FailingSmtp)
|
|
|
|
headers = _register_and_login(client)
|
|
company = _create_company(client, headers)
|
|
destination = client.post(
|
|
"/api/v1/notification-destinations",
|
|
json={
|
|
"type": "email",
|
|
"destination_value": "[email protected]",
|
|
"company_ids": [company["id"]],
|
|
},
|
|
headers=headers,
|
|
).json()
|
|
|
|
resp = client.post(
|
|
f"/api/v1/notification-destinations/{destination['id']}/test", headers=headers
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["success"] is False
|
|
assert "no mailpit running" in body["error"]
|
|
|
|
|
|
def test_test_endpoint_not_owned_returns_404(client):
|
|
owner_headers = _register_and_login(client)
|
|
other_headers = _register_and_login(client)
|
|
company = _create_company(client, owner_headers)
|
|
destination = client.post(
|
|
"/api/v1/notification-destinations",
|
|
json={
|
|
"type": "email",
|
|
"destination_value": "[email protected]",
|
|
"company_ids": [company["id"]],
|
|
},
|
|
headers=owner_headers,
|
|
).json()
|
|
|
|
resp = client.post(
|
|
f"/api/v1/notification-destinations/{destination['id']}/test", headers=other_headers
|
|
)
|
|
assert resp.status_code == 404
|