email: tighten unsubscribe — test isolation, accurate comments, tighter assertion
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
a292289dc6
commit
0a476bed22
2 changed files with 13 additions and 13 deletions
|
|
@ -15,6 +15,7 @@ from fastapi.responses import HTMLResponse
|
||||||
from itsdangerous import BadSignature, URLSafeSerializer
|
from itsdangerous import BadSignature, URLSafeSerializer
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
|
from app import branding
|
||||||
from app.config import get_settings
|
from app.config import get_settings
|
||||||
from app.db import get_session
|
from app.db import get_session
|
||||||
from app.logging import get_logger
|
from app.logging import get_logger
|
||||||
|
|
@ -30,8 +31,9 @@ _SALT = "digest-unsubscribe-v1"
|
||||||
def _serializer() -> URLSafeSerializer:
|
def _serializer() -> URLSafeSerializer:
|
||||||
s = get_settings()
|
s = get_settings()
|
||||||
if not s.CASSANDRA_SESSION_SECRET:
|
if not s.CASSANDRA_SESSION_SECRET:
|
||||||
# In tests with no secret configured, fall back to a constant —
|
# In tests with no secret configured, fall back to a constant.
|
||||||
# NEVER reach production; settings validation should catch this.
|
# An empty CASSANDRA_SESSION_SECRET in prod would also break login,
|
||||||
|
# so this branch is "best-effort dev fallback", not a real prod path.
|
||||||
return URLSafeSerializer("dev-only-empty-secret", salt=_SALT)
|
return URLSafeSerializer("dev-only-empty-secret", salt=_SALT)
|
||||||
return URLSafeSerializer(s.CASSANDRA_SESSION_SECRET, salt=_SALT)
|
return URLSafeSerializer(s.CASSANDRA_SESSION_SECRET, salt=_SALT)
|
||||||
|
|
||||||
|
|
@ -84,7 +86,6 @@ async def unsubscribe(
|
||||||
token: str = Query(...),
|
token: str = Query(...),
|
||||||
session: AsyncSession = Depends(get_session),
|
session: AsyncSession = Depends(get_session),
|
||||||
):
|
):
|
||||||
from app import branding
|
|
||||||
uid = verify_unsubscribe_token(token)
|
uid = verify_unsubscribe_token(token)
|
||||||
if uid is not None:
|
if uid is not None:
|
||||||
user = await session.get(User, uid)
|
user = await session.get(User, uid)
|
||||||
|
|
|
||||||
|
|
@ -4,9 +4,8 @@ from __future__ import annotations
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
||||||
|
|
||||||
def _build_app(tmp_path, secret="rt-secret-32-bytes-or-so-padding-here"):
|
def _build_app(tmp_path, monkeypatch, secret="rt-secret-32-bytes-or-so-padding-here"):
|
||||||
import os
|
monkeypatch.setenv("CASSANDRA_SESSION_SECRET", secret)
|
||||||
os.environ["CASSANDRA_SESSION_SECRET"] = secret
|
|
||||||
|
|
||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
@ -48,8 +47,8 @@ def test_sign_and_verify_token_roundtrip(monkeypatch):
|
||||||
assert verify_unsubscribe_token("garbage") is None
|
assert verify_unsubscribe_token("garbage") is None
|
||||||
|
|
||||||
|
|
||||||
def test_get_unsubscribe_flips_flag(tmp_path):
|
def test_get_unsubscribe_flips_flag(tmp_path, monkeypatch):
|
||||||
client = _build_app(tmp_path)
|
client = _build_app(tmp_path, monkeypatch)
|
||||||
from app.routers.email import sign_unsubscribe_token
|
from app.routers.email import sign_unsubscribe_token
|
||||||
tok = sign_unsubscribe_token(42)
|
tok = sign_unsubscribe_token(42)
|
||||||
r = client.get(f"/email/unsubscribe?token={tok}")
|
r = client.get(f"/email/unsubscribe?token={tok}")
|
||||||
|
|
@ -65,16 +64,16 @@ def test_get_unsubscribe_flips_flag(tmp_path):
|
||||||
asyncio.run(_check())
|
asyncio.run(_check())
|
||||||
|
|
||||||
|
|
||||||
def test_get_unsubscribe_invalid_token_returns_generic_page(tmp_path):
|
def test_get_unsubscribe_invalid_token_returns_generic_page(tmp_path, monkeypatch):
|
||||||
client = _build_app(tmp_path)
|
client = _build_app(tmp_path, monkeypatch)
|
||||||
r = client.get("/email/unsubscribe?token=garbage")
|
r = client.get("/email/unsubscribe?token=garbage")
|
||||||
# We don't 4xx — that would leak token validity. Show the generic page.
|
# We don't 4xx — that would leak token validity. Show the generic page.
|
||||||
assert r.status_code == 200
|
assert r.status_code == 200
|
||||||
assert "unsubscribed" in r.text.lower() or "preferences" in r.text.lower()
|
assert "you're unsubscribed" in r.text.lower()
|
||||||
|
|
||||||
|
|
||||||
def test_replay_is_idempotent(tmp_path):
|
def test_replay_is_idempotent(tmp_path, monkeypatch):
|
||||||
client = _build_app(tmp_path)
|
client = _build_app(tmp_path, monkeypatch)
|
||||||
from app.routers.email import sign_unsubscribe_token
|
from app.routers.email import sign_unsubscribe_token
|
||||||
tok = sign_unsubscribe_token(42)
|
tok = sign_unsubscribe_token(42)
|
||||||
r1 = client.get(f"/email/unsubscribe?token={tok}")
|
r1 = client.get(f"/email/unsubscribe?token={tok}")
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue