21f7ef4d19
Move the flat backend/tests/*.py into domain subfolders so the suite mirrors the code's concerns: data/ climate, grading, scoring, grid, places, store web/ api, content, homepage, views notifications/ notify, digest, discord (+ dm/interactions/link) accounts/ api_accounts core/ metrics, singleton, dashboard conftest.py stays at the tests/ root, so its sys.path setup and shared fixtures still apply to every subfolder. The four tests that derive repo paths from __file__ get their depth bumped one level to match their new location. Pytest discovers the subfolders recursively; the CI command (python -m pytest backend/tests) is unchanged. Claude-Session: https://claude.ai/code/session_01XXxmNFy9cZ6Gh8Y9thZn62
121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
"""Tests for the digest signup and the mailer seam."""
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app as appmod
|
|
import db
|
|
import digest
|
|
import mailer
|
|
from db import async_session_maker
|
|
from 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):
|
|
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
|
|
|
|
|
|
def test_form_is_hidden_for_now(client):
|
|
"""The footer digest signup is temporarily hidden (commented out in the
|
|
base template). The /digest endpoint stays live; only the UI is pulled.
|
|
Restore the form and flip this back to asserting presence."""
|
|
for path in ["/", "/about", "/climate", "/privacy", "/glossary"]:
|
|
html = client.get(f"{B}{path}").text
|
|
assert "data-digest" not in html, path
|