Giorgio Gilestro
e338650dfa
- verify_submit now applies the subscribe checkbox only at first sign-up. Returning users keep whatever they set via Settings or the one-click unsubscribe link — previously, every login silently re-enrolled them. - JOB_NAMES gains email_digest_job so the ops footer reflects its health. Adds tests/test_verify_subscribe.py::test_returning_user_login_preserves_unsubscribe.
130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
"""Verify-POST persists the subscribe_to_digests form field."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
|
|
def _build(tmp_path):
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|
|
|
from app import db as db_mod
|
|
from app.db import Base
|
|
from app.models import User
|
|
from app.routers import auth as auth_router
|
|
from app.auth import _pending_serializer
|
|
|
|
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/v.db")
|
|
factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
db_mod._engine = engine
|
|
db_mod._session_factory = factory
|
|
|
|
async def _seed():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
async with factory() as s:
|
|
s.add(User(id=10, email="newbie@x", tier="free",
|
|
email_digest_opt_in=True))
|
|
await s.commit()
|
|
|
|
asyncio.run(_seed())
|
|
app = FastAPI()
|
|
app.include_router(auth_router.router)
|
|
pending = _pending_serializer().dumps({"email": "newbie@x", "uid": 10, "ref": None})
|
|
return TestClient(app), pending
|
|
|
|
|
|
def test_verify_with_unchecked_subscribe_disables_opt_in(tmp_path, monkeypatch):
|
|
from app.services import otp_service
|
|
|
|
async def _ok(*args, **kwargs):
|
|
return None
|
|
monkeypatch.setattr(otp_service, "verify", _ok)
|
|
|
|
client, pending = _build(tmp_path)
|
|
r = client.post(
|
|
"/verify",
|
|
data={"code": "000000"}, # subscribe_to_digests omitted = unchecked
|
|
cookies={"cassandra_pending": pending},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303, r.text
|
|
|
|
async def _check():
|
|
from app import db as db_mod
|
|
from app.models import User
|
|
async with db_mod._session_factory() as s:
|
|
u = await s.get(User, 10)
|
|
assert u.email_digest_opt_in is False
|
|
asyncio.run(_check())
|
|
|
|
|
|
def test_verify_with_checked_subscribe_keeps_opt_in(tmp_path, monkeypatch):
|
|
from app.services import otp_service
|
|
|
|
async def _ok(*args, **kwargs):
|
|
return None
|
|
monkeypatch.setattr(otp_service, "verify", _ok)
|
|
|
|
client, pending = _build(tmp_path)
|
|
r = client.post(
|
|
"/verify",
|
|
data={"code": "000000", "subscribe_to_digests": "on"},
|
|
cookies={"cassandra_pending": pending},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
|
|
async def _check():
|
|
from app import db as db_mod
|
|
from app.models import User
|
|
async with db_mod._session_factory() as s:
|
|
u = await s.get(User, 10)
|
|
assert u.email_digest_opt_in is True
|
|
asyncio.run(_check())
|
|
|
|
|
|
def test_returning_user_login_preserves_unsubscribe(tmp_path, monkeypatch):
|
|
"""A user who unsubscribed (via Settings or the one-click link) must
|
|
not be silently re-enrolled when they log in again, even if the verify
|
|
page's default-checked checkbox gets submitted."""
|
|
from datetime import datetime, timezone
|
|
|
|
from app.services import otp_service
|
|
|
|
async def _ok(*args, **kwargs):
|
|
return None
|
|
monkeypatch.setattr(otp_service, "verify", _ok)
|
|
|
|
client, pending = _build(tmp_path)
|
|
|
|
# Simulate a returning user: previously logged in, then opted out.
|
|
async def _make_returning():
|
|
from app import db as db_mod
|
|
from app.models import User
|
|
async with db_mod._session_factory() as s:
|
|
u = await s.get(User, 10)
|
|
u.last_login_at = datetime(2026, 5, 20, 12, 0, tzinfo=timezone.utc)
|
|
u.email_digest_opt_in = False
|
|
await s.commit()
|
|
asyncio.run(_make_returning())
|
|
|
|
# They log in again. The verify checkbox is default-checked in the
|
|
# template, so the form will submit "on" — but the handler must NOT
|
|
# apply that to a returning user.
|
|
r = client.post(
|
|
"/verify",
|
|
data={"code": "000000", "subscribe_to_digests": "on"},
|
|
cookies={"cassandra_pending": pending},
|
|
follow_redirects=False,
|
|
)
|
|
assert r.status_code == 303
|
|
|
|
async def _check():
|
|
from app import db as db_mod
|
|
from app.models import User
|
|
async with db_mod._session_factory() as s:
|
|
u = await s.get(User, 10)
|
|
assert u.email_digest_opt_in is False, "returning user re-enrolled"
|
|
asyncio.run(_check())
|