116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""Tests for the digest signup and the mailer seam."""
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from fastapi.testclient import TestClient
|
|
|
|
from web import app as appmod
|
|
from accounts import db
|
|
from notifications import digest
|
|
from notifications import mailer
|
|
from accounts.db import async_session_maker
|
|
from accounts.models import PendingDigest
|
|
|
|
B = "/thermograph"
|
|
|
|
|
|
@pytest.fixture(scope="module", autouse=True)
|
|
def _tables():
|
|
# TestClient(app) (no context manager) skips the lifespan, so create the
|
|
# account tables directly against the temp DB.
|
|
db.Base.metadata.create_all(db.sync_engine)
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(appmod.app)
|
|
|
|
|
|
def _rows(client, email):
|
|
"""Count stored rows for an address, via the running event loop."""
|
|
import asyncio
|
|
|
|
async def go():
|
|
async with async_session_maker() as session:
|
|
return (await session.execute(
|
|
sa.select(sa.func.count()).select_from(PendingDigest)
|
|
.where(PendingDigest.email == email)
|
|
)).scalar_one()
|
|
return asyncio.run(go())
|
|
|
|
|
|
# --- mailer ------------------------------------------------------------------
|
|
def test_console_backend_sends_nothing(monkeypatch):
|
|
"""The default backend must never open a socket — this is what makes it safe
|
|
to run the full signup path in dev and in tests."""
|
|
monkeypatch.setattr(mailer, "BACKEND", "console")
|
|
|
|
def boom(*a, **k):
|
|
raise AssertionError("console backend must not connect to SMTP")
|
|
|
|
monkeypatch.setattr(mailer.smtplib, "SMTP", boom)
|
|
assert mailer.send("a@example.com", "s", "t") is True
|
|
assert mailer.enabled() is False
|
|
|
|
|
|
def test_send_swallows_connection_failure(monkeypatch):
|
|
"""A dead mail server must not raise into the request that triggered it."""
|
|
monkeypatch.setattr(mailer, "BACKEND", "smtp")
|
|
|
|
def boom(*a, **k):
|
|
raise OSError("connection refused")
|
|
|
|
monkeypatch.setattr(mailer.smtplib, "SMTP", boom)
|
|
assert mailer.send("a@example.com", "s", "t") is False
|
|
|
|
|
|
def test_disabled_backend_drops(monkeypatch):
|
|
monkeypatch.setattr(mailer, "BACKEND", "disabled")
|
|
assert mailer.send("a@example.com", "s", "t") is False
|
|
|
|
|
|
def test_invalid_recipient_rejected():
|
|
assert mailer.send("", "s", "t") is False
|
|
assert mailer.send("not-an-address", "s", "t") is False
|
|
|
|
|
|
# --- signup ------------------------------------------------------------------
|
|
def test_signup_stores_and_confirms(client):
|
|
r = client.post(f"{B}/digest", json={"email": "reader@example.com"})
|
|
assert r.status_code == 200
|
|
assert r.json()["message"] == digest.GENERIC_OK
|
|
assert _rows(client, "reader@example.com") == 1
|
|
|
|
|
|
def test_signup_is_idempotent_and_case_insensitive(client):
|
|
first = client.post(f"{B}/digest", json={"email": "Dup@Example.COM"})
|
|
second = client.post(f"{B}/digest", json={"email": "dup@example.com"})
|
|
# Same answer either way: the endpoint must not reveal whether an address is
|
|
# already on the list.
|
|
assert first.json()["message"] == second.json()["message"] == digest.GENERIC_OK
|
|
assert _rows(client, "dup@example.com") == 1
|
|
|
|
|
|
def test_signup_rejects_junk(client):
|
|
for bad in ["", "nope", "a@b", "@example.com", "x@", "a b@example.com"]:
|
|
r = client.post(f"{B}/digest", json={"email": bad})
|
|
assert r.status_code == 400, bad
|
|
|
|
|
|
def test_signup_accepts_plain_form_post(client):
|
|
"""The no-JS path: a real <form method=post> submission."""
|
|
r = client.post(f"{B}/digest", data={"email": "nojs@example.com"})
|
|
assert r.status_code == 200
|
|
assert _rows(client, "nojs@example.com") == 1
|
|
|
|
|
|
def test_signup_records_event(client):
|
|
from core import metrics
|
|
before = metrics.snapshot()["events"].get("home.digest_signup", {})
|
|
before_n = sum(sum(days.values()) for days in before.values())
|
|
client.post(f"{B}/digest", json={"email": "counted@example.com"})
|
|
after = metrics.snapshot()["events"].get("home.digest_signup", {})
|
|
assert sum(sum(days.values()) for days in after.values()) == before_n + 1
|
|
|
|
|
|
# test_form_is_hidden_for_now moved to frontend_ssr/tests/test_content.py
|
|
# (repo-split Stage 4) -- every page it checked is frontend-owned now.
|