giorgio/read.markets
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
read.markets/tests/test_signup_acknowledgement.py
Giorgio Gilestro f3ac65f8f7 auth: affirmative versioned sign-up acknowledgement (EN + IT) 
Replaces the passive "by signing in you agree…" paragraph on /login with
an active un-pre-ticked checkbox over three short statements that the
user actively accepts. Each acceptance is recorded against a code-level
version pointer, so a future wording revision bumps the version and
prompts every user again.

- New app/legal.py::ACKNOWLEDGEMENT_VERSION (currently 1).
- New UserAcknowledgement model + migration 0027 (user_id FK CASCADE,
  version, lang, accepted_at, composite index on user_id+version).
- auth_service: has_acknowledged_current() and record_acknowledgement()
  helpers; POST /login validates the checkbox, falls through to a 400
  with the localised error otherwise, and writes a row iff the user has
  no current-version row (so existing-already-accepted users don't
  produce duplicates).
- GET /login: language detection mirrors the landing's
  detect_public_lang(); ?lang=en|it overrides; stamps the rtm.lang
  cookie; passes the locale dict + version into the template.
- login.html: EN/IT pill, localised lede/banner/legal footer, required
  checkbox in an acknowledgement block, hidden lang+ack_version fields.
  Submit disabled until the box is ticked (UX polish; the server check
  is what carries weight).
- locales/{en,it}.yaml: new auth.ack.* section with TODO(legal) marker.
  Wording matches the brief's substance pending solicitor sign-off.
- tests/test_signup_acknowledgement.py: 10 tests (EN + IT rejection,
  one row per acceptance, displayed-lang recorded, idempotent on
  current version, version-bump writes new row, helper unit tests).

The acknowledgement strengthens the user-civil-claim vector — combined
with the liability cap and the Ltd, it makes "I was misled into thinking
this was advice" much harder to argue. It does NOT move the regulatory
perimeter, which is governed by the content discipline shipped in
47dce1a. Belt-and-braces, not a substitute.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-29 20:50:38 +02:00

388 lines
13 KiB
Python
Raw Blame History

"""Sign-up acknowledgement: the affirmative checkbox at /login.
Covers:
- POST /login without the box ticked → 400, form re-rendered with the error.
- New email with box ticked → User row + UserAcknowledgement row at the
current version, in the language the user actually saw.
- Existing user with a current-version ack row → POST succeeds, no duplicate.
- Existing user with only an older-version ack row → new row at current.
- has_acknowledged_current() unit tests.
"""
from __future__ import annotations
import asyncio
def _build(tmp_path):
"""Spin up a fresh app + sqlite DB + tables. Returns (TestClient, factory).
Patches otp_service and email send into no-ops so POST /login can complete
without hitting SMTP. The acknowledgement is captured during POST /login
(before OTP), so /verify never needs to be exercised here. Static files
are mounted because the rejection path re-renders login.html which
references ``url_for('static', ...)``."""
from pathlib import Path
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
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
import app.models # noqa: F401 — registers tables
from app.routers import auth as auth_router
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/ack.db")
factory = async_sessionmaker(engine, expire_on_commit=False)
db_mod._engine = engine
db_mod._session_factory = factory
async def _create_all():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
asyncio.run(_create_all())
app = FastAPI()
app.include_router(auth_router.router)
static_dir = Path(__file__).resolve().parent.parent / "app" / "static"
app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")
return TestClient(app), factory
def _patch_otp_email(monkeypatch):
"""Stub OTP issuance and email send so POST /login can run end-to-end
without a Redis-backed OTP service or real SMTP."""
from app.services import otp_service
from app.routers import auth as auth_router
async def _allowed(*_a, **_kw):
return (True, 0)
async def _issue(*_a, **_kw):
return "123456"
async def _send_ok(*_a, **_kw):
return True
monkeypatch.setattr(otp_service, "can_request_new", _allowed)
monkeypatch.setattr(otp_service, "issue", _issue)
# _issue_and_send_otp lives on the router and wraps the email send;
# easier to stub the whole helper than to thread through email_service.
monkeypatch.setattr(auth_router, "_issue_and_send_otp", _send_ok)
async def _count_acks(factory, user_id: int, version: int | None = None) -> int:
from sqlalchemy import select, func
from app.models import UserAcknowledgement
async with factory() as s:
q = select(func.count()).select_from(UserAcknowledgement).where(
UserAcknowledgement.user_id == user_id,
)
if version is not None:
q = q.where(UserAcknowledgement.version == version)
return (await s.execute(q)).scalar() or 0
# ---------------------------------------------------------------------------
# POST /login validation: missing-checkbox path
# ---------------------------------------------------------------------------
def test_post_login_rejects_when_acknowledged_unchecked(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, _ = _build(tmp_path)
r = client.post(
"/login",
data={"email": "alice@example.com", "next": "/", "lang": "en"},
follow_redirects=False,
)
assert r.status_code == 400
# The localised error message goes back in the rendered template.
assert "tick the box" in r.text.lower() or "confirm" in r.text.lower()
def test_post_login_rejection_preserves_email(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, _ = _build(tmp_path)
r = client.post(
"/login",
data={"email": "alice@example.com", "next": "/", "lang": "en"},
follow_redirects=False,
)
assert r.status_code == 400
assert "alice@example.com" in r.text
def test_post_login_localises_error_in_italian(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, _ = _build(tmp_path)
r = client.post(
"/login",
data={"email": "anna@example.com", "next": "/", "lang": "it"},
follow_redirects=False,
)
assert r.status_code == 400
# IT error: "Spunta la casella per confermare prima di continuare."
assert "spunta la casella" in r.text.lower() or "confermare" in r.text.lower()
# ---------------------------------------------------------------------------
# Successful POST /login: writes User + UserAcknowledgement
# ---------------------------------------------------------------------------
def test_new_signup_writes_acknowledgement_row(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, factory = _build(tmp_path)
r = client.post(
"/login",
data={
"email": "alice@example.com",
"next": "/",
"lang": "en",
"acknowledged": "on",
"ack_version": "1",
},
follow_redirects=False,
)
# 303 → /verify
assert r.status_code == 303
assert r.headers["location"].startswith("/verify")
# Find the new user and assert exactly one acknowledgement row at v1.
from app.models import User, UserAcknowledgement
from sqlalchemy import select
async def _check():
async with factory() as s:
user = (await s.execute(
select(User).where(User.email == "alice@example.com")
)).scalar_one()
rows = (await s.execute(
select(UserAcknowledgement).where(
UserAcknowledgement.user_id == user.id,
)
)).scalars().all()
return user, rows
user, rows = asyncio.run(_check())
assert len(rows) == 1
ack = rows[0]
assert ack.version == 1
assert ack.lang == "en"
assert ack.accepted_at is not None
def test_acknowledgement_records_displayed_language(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, factory = _build(tmp_path)
r = client.post(
"/login",
data={
"email": "anna@example.it",
"next": "/",
"lang": "it",
"acknowledged": "on",
"ack_version": "1",
},
follow_redirects=False,
)
assert r.status_code == 303
from app.models import User, UserAcknowledgement
from sqlalchemy import select
async def _check():
async with factory() as s:
user = (await s.execute(
select(User).where(User.email == "anna@example.it")
)).scalar_one()
ack = (await s.execute(
select(UserAcknowledgement).where(
UserAcknowledgement.user_id == user.id,
)
)).scalar_one()
return ack.lang
assert asyncio.run(_check()) == "it"
# ---------------------------------------------------------------------------
# Idempotency: existing user already at current version → no dup row
# ---------------------------------------------------------------------------
def test_existing_user_current_version_no_duplicate(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, factory = _build(tmp_path)
# Pre-seed: User + one acknowledgement at the current version.
async def _seed():
from app.models import User, UserAcknowledgement
from app.legal import ACKNOWLEDGEMENT_VERSION
from app.db import utcnow
async with factory() as s:
u = User(email="repeat@example.com", tier="free",
settings_json={}, created_at=utcnow())
s.add(u)
await s.commit()
await s.refresh(u)
s.add(UserAcknowledgement(
user_id=u.id,
version=ACKNOWLEDGEMENT_VERSION,
lang="en",
accepted_at=utcnow(),
))
await s.commit()
return u.id
user_id = asyncio.run(_seed())
before = asyncio.run(_count_acks(factory, user_id))
assert before == 1
r = client.post(
"/login",
data={
"email": "repeat@example.com",
"next": "/",
"lang": "en",
"acknowledged": "on",
"ack_version": "1",
},
follow_redirects=False,
)
assert r.status_code == 303
after = asyncio.run(_count_acks(factory, user_id))
assert after == 1, "must not write a duplicate row when user already at current version"
# ---------------------------------------------------------------------------
# Version bump: existing user only at older version → new current-version row
# ---------------------------------------------------------------------------
def test_existing_user_older_version_writes_new_current_row(tmp_path, monkeypatch):
_patch_otp_email(monkeypatch)
client, factory = _build(tmp_path)
# Pre-seed a user with an OLD-version acknowledgement (version=0).
# The current version constant is 1 → this user is "stale" and should
# be prompted again.
async def _seed():
from app.models import User, UserAcknowledgement
from app.db import utcnow
async with factory() as s:
u = User(email="bump@example.com", tier="free",
settings_json={}, created_at=utcnow())
s.add(u)
await s.commit()
await s.refresh(u)
s.add(UserAcknowledgement(
user_id=u.id, version=0, lang="en", accepted_at=utcnow(),
))
await s.commit()
return u.id
user_id = asyncio.run(_seed())
r = client.post(
"/login",
data={
"email": "bump@example.com",
"next": "/",
"lang": "en",
"acknowledged": "on",
"ack_version": "1",
},
follow_redirects=False,
)
assert r.status_code == 303
# Total rows: 1 old + 1 new = 2. Current-version rows: exactly 1.
total = asyncio.run(_count_acks(factory, user_id))
current = asyncio.run(_count_acks(factory, user_id, version=1))
assert total == 2
assert current == 1
# ---------------------------------------------------------------------------
# has_acknowledged_current() — unit-ish, no HTTP
# ---------------------------------------------------------------------------
def test_has_acknowledged_current_no_row(tmp_path):
_, factory = _build(tmp_path)
async def _go():
from app.models import User
from app.services.auth_service import has_acknowledged_current
from app.db import utcnow
async with factory() as s:
u = User(email="empty@example.com", tier="free",
settings_json={}, created_at=utcnow())
s.add(u)
await s.commit()
await s.refresh(u)
return await has_acknowledged_current(s, u)
assert asyncio.run(_go()) is False
def test_has_acknowledged_current_only_old(tmp_path):
_, factory = _build(tmp_path)
async def _go():
from app.models import User, UserAcknowledgement
from app.services.auth_service import has_acknowledged_current
from app.db import utcnow
async with factory() as s:
u = User(email="oldonly@example.com", tier="free",
settings_json={}, created_at=utcnow())
s.add(u)
await s.commit()
await s.refresh(u)
s.add(UserAcknowledgement(
user_id=u.id, version=0, lang="en", accepted_at=utcnow(),
))
await s.commit()
return await has_acknowledged_current(s, u)
assert asyncio.run(_go()) is False
def test_has_acknowledged_current_at_current(tmp_path):
_, factory = _build(tmp_path)
async def _go():
from app.models import User, UserAcknowledgement
from app.services.auth_service import has_acknowledged_current
from app.legal import ACKNOWLEDGEMENT_VERSION
from app.db import utcnow
async with factory() as s:
u = User(email="atcurrent@example.com", tier="free",
settings_json={}, created_at=utcnow())
s.add(u)
await s.commit()
await s.refresh(u)
s.add(UserAcknowledgement(
user_id=u.id,
version=ACKNOWLEDGEMENT_VERSION,
lang="en",
accepted_at=utcnow(),
))
await s.commit()
return await has_acknowledged_current(s, u)
assert asyncio.run(_go()) is True
Reference in a new issue View git blame Copy permalink