feat(slice-2): accounts domain — request_code/verify/get_account (INV-2/INV-3/INV-6)
Email-canonical get-or-create, peppered-hash one-time codes with 10-min TTL, single-use, 5-attempt cap, and 60s resend cooldown; uniform new-or-known (no enumeration, §6.6). All identity rules in the domain layer once (INV-6); scenario + invariant tests green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,66 @@
|
||||
import re
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
|
||||
from app.domains import accounts
|
||||
from app.platform import db, mailer
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def conn(fresh_db_url):
|
||||
with psycopg.connect(fresh_db_url) as c:
|
||||
db.migrate(c)
|
||||
yield c
|
||||
|
||||
|
||||
def _code_in(outbox) -> str:
|
||||
# The 6-digit code appears in the latest message body (LogMailer dev channel).
|
||||
return re.search(r"\b(\d{6})\b", outbox[-1].body).group(1)
|
||||
|
||||
|
||||
def test_14_01_0002_request_code_sends_a_six_digit_code(conn):
|
||||
m = mailer.LogMailer()
|
||||
accounts.request_code(conn, m, "Merchant@Example.com")
|
||||
assert len(m.outbox) == 1
|
||||
assert m.outbox[-1].to == "merchant@example.com" # normalized recipient (INV-2)
|
||||
assert re.search(r"\b\d{6}\b", m.outbox[-1].body)
|
||||
|
||||
|
||||
def test_14_01_0003_request_code_stores_only_a_hash(conn):
|
||||
m = mailer.LogMailer()
|
||||
accounts.request_code(conn, m, "merchant@example.com")
|
||||
code = _code_in(m.outbox)
|
||||
rows = conn.execute("SELECT code_hash FROM auth_code WHERE email = %s", ("merchant@example.com",)).fetchall()
|
||||
assert len(rows) == 1
|
||||
# The plaintext code is never stored (INV-3).
|
||||
assert code not in rows[0][0]
|
||||
assert len(rows[0][0]) >= 16 # a hash, not the 6-digit code
|
||||
|
||||
|
||||
def test_request_code_uniform_for_new_and_known(conn):
|
||||
# No account enumeration (§6.6): the call behaves identically whether or not the email
|
||||
# already has an account — it never signals existence and never raises.
|
||||
m = mailer.LogMailer()
|
||||
accounts.request_code(conn, m, "newcomer@example.com") # no account exists
|
||||
# create an account, then request again for a known email
|
||||
conn.execute("INSERT INTO account (email) VALUES (%s)", ("known@example.com",))
|
||||
conn.commit()
|
||||
accounts.request_code(conn, m, "known@example.com")
|
||||
assert len(m.outbox) == 2 # both sent a code; caller cannot tell new from known
|
||||
|
||||
|
||||
def test_puc_02c_resend_is_rate_limited(conn):
|
||||
m = mailer.LogMailer()
|
||||
accounts.request_code(conn, m, "merchant@example.com")
|
||||
with pytest.raises(accounts.ResendCooldown) as exc:
|
||||
accounts.request_code(conn, m, "merchant@example.com")
|
||||
assert 0 < exc.value.retry_after_s <= 60
|
||||
assert len(m.outbox) == 1 # the second code was NOT sent
|
||||
|
||||
|
||||
def test_request_code_rejects_invalid_email(conn):
|
||||
m = mailer.LogMailer()
|
||||
with pytest.raises(accounts.InvalidEmail):
|
||||
accounts.request_code(conn, m, "not-an-email")
|
||||
assert m.outbox == []
|
||||
@@ -0,0 +1,100 @@
|
||||
import re
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
|
||||
from app.domains import accounts
|
||||
from app.platform import db, mailer
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def conn(fresh_db_url):
|
||||
with psycopg.connect(fresh_db_url) as c:
|
||||
db.migrate(c)
|
||||
yield c
|
||||
|
||||
|
||||
def _issue(conn, email) -> str:
|
||||
m = mailer.LogMailer()
|
||||
accounts.request_code(conn, m, email)
|
||||
return re.search(r"\b(\d{6})\b", m.outbox[-1].body).group(1)
|
||||
|
||||
|
||||
def test_14_01_0004_verify_creates_account_and_reports_created(conn):
|
||||
code = _issue(conn, "merchant@example.com")
|
||||
account, created = accounts.verify(conn, "merchant@example.com", code)
|
||||
assert created is True
|
||||
assert account.email == "merchant@example.com"
|
||||
assert account.id > 0
|
||||
# the account row exists exactly once (INV-2)
|
||||
n = conn.execute("SELECT count(*) FROM account WHERE email = %s", ("merchant@example.com",)).fetchone()[0]
|
||||
assert n == 1
|
||||
|
||||
|
||||
def test_14_01_0009_returning_account_reports_not_created(conn):
|
||||
# First sign-up creates the account.
|
||||
code1 = _issue(conn, "merchant@example.com")
|
||||
a1, created1 = accounts.verify(conn, "merchant@example.com", code1)
|
||||
assert created1 is True
|
||||
# A later login resolves to the SAME account (INV-2 — email is canonical), created=False.
|
||||
code2 = _issue(conn, "Merchant@Example.com") # different casing, same identity
|
||||
a2, created2 = accounts.verify(conn, "Merchant@Example.com", code2)
|
||||
assert created2 is False
|
||||
assert a2.id == a1.id
|
||||
n = conn.execute("SELECT count(*) FROM account").fetchone()[0]
|
||||
assert n == 1 # no duplicate account (14.01.0010)
|
||||
|
||||
|
||||
def test_verify_consumes_the_code_single_use(conn):
|
||||
code = _issue(conn, "merchant@example.com")
|
||||
accounts.verify(conn, "merchant@example.com", code)
|
||||
# Reusing a consumed code is refused (single-use, INV-3): no live code -> CodeExpired.
|
||||
with pytest.raises(accounts.CodeExpired):
|
||||
accounts.verify(conn, "merchant@example.com", code)
|
||||
|
||||
|
||||
def test_puc_02a_wrong_code_decrements_attempts(conn):
|
||||
_issue(conn, "merchant@example.com")
|
||||
with pytest.raises(accounts.CodeMismatch) as exc:
|
||||
accounts.verify(conn, "merchant@example.com", "000000")
|
||||
# 5 attempts allowed; one wrong -> 4 remaining (INV-3).
|
||||
assert exc.value.attempts_remaining == 4
|
||||
|
||||
|
||||
def test_puc_02b_expired_code_is_refused(conn):
|
||||
code = _issue(conn, "merchant@example.com")
|
||||
# Force the live code to be expired.
|
||||
conn.execute(
|
||||
"UPDATE auth_code SET expires_at = %s WHERE email = %s",
|
||||
(datetime.now(timezone.utc) - timedelta(minutes=1), "merchant@example.com"),
|
||||
)
|
||||
conn.commit()
|
||||
with pytest.raises(accounts.CodeExpired):
|
||||
accounts.verify(conn, "merchant@example.com", code)
|
||||
|
||||
|
||||
def test_inv3_attempts_exhausted_invalidates_code(conn):
|
||||
code = _issue(conn, "merchant@example.com")
|
||||
# Five wrong attempts: the 5th raises CodeExhausted and invalidates the code.
|
||||
for _ in range(4):
|
||||
with pytest.raises(accounts.CodeMismatch):
|
||||
accounts.verify(conn, "merchant@example.com", "000000")
|
||||
with pytest.raises(accounts.CodeExhausted):
|
||||
accounts.verify(conn, "merchant@example.com", "000000")
|
||||
# Even the correct code no longer works — a fresh request is required.
|
||||
with pytest.raises((accounts.CodeExpired, accounts.CodeExhausted)):
|
||||
accounts.verify(conn, "merchant@example.com", code)
|
||||
|
||||
|
||||
def test_verify_with_no_request_is_code_expired(conn):
|
||||
with pytest.raises(accounts.CodeExpired):
|
||||
accounts.verify(conn, "stranger@example.com", "123456")
|
||||
|
||||
|
||||
def test_get_account_loads_by_id(conn):
|
||||
code = _issue(conn, "merchant@example.com")
|
||||
account, _ = accounts.verify(conn, "merchant@example.com", code)
|
||||
loaded = accounts.get_account(conn, account.id)
|
||||
assert loaded is not None and loaded.email == "merchant@example.com"
|
||||
assert accounts.get_account(conn, 999999) is None
|
||||
Reference in New Issue
Block a user