Booking schema + tenancy-safe data-access module (#15)
Adds the plumbing every other booking ticket builds on: resources, services, users, and password_reset_tokens tables, plus the clients columns (slug, timezone, auto_confirm, ics_token) and the bookings resource_id/EXCLUDE-constraint double-booking protection described in #14. booking_db.py is the only place raw SQL runs against these tables -- every function takes client_id and injects the tenant filter itself, and create/update_booking additionally verify the resource_id belongs to that client before writing, closing a guessed-ID cross-tenant hole. Tests spin up a real throwaway Postgres 16 container (matching prod) and exercise the EXCLUDE constraint, tenancy isolation, and password-reset single-use semantics end to end, per #14's "real Postgres, no mocking" testing decision. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,217 @@
|
||||
"""Tenancy-safe data-access layer for the booking module (#15).
|
||||
|
||||
This is the only place that runs raw SQL against resources, services,
|
||||
bookings, users, and password_reset_tokens. Every function takes a
|
||||
client_id (or, for password-reset consumption, an unguessable token that
|
||||
resolves straight to a user) and injects the tenant filter itself -- callers
|
||||
never write "WHERE client_id = ..." by hand.
|
||||
"""
|
||||
import secrets
|
||||
import time
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import psycopg
|
||||
from werkzeug.security import check_password_hash, generate_password_hash
|
||||
|
||||
import db
|
||||
|
||||
|
||||
class BookingConflict(Exception):
|
||||
"""Raised when a booking would overlap another on the same resource."""
|
||||
|
||||
|
||||
class UnknownResource(Exception):
|
||||
"""Raised when a resource_id doesn't belong to the given client_id --
|
||||
guards against a booking write smuggling in another tenant's resource."""
|
||||
|
||||
|
||||
def _new_id(prefix):
|
||||
return f"{prefix}-{int(time.time() * 1000)}-{secrets.token_hex(3)}"
|
||||
|
||||
|
||||
# ---- resources ----
|
||||
|
||||
def create_resource(client_id, name, active=True):
|
||||
resource_id = _new_id("RS")
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO resources (resource_id, client_id, name, active) "
|
||||
"VALUES (%s, %s, %s, %s) RETURNING *",
|
||||
(resource_id, client_id, name, active))
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
return row
|
||||
|
||||
|
||||
def get_resource(client_id, resource_id):
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT * FROM resources WHERE client_id = %s AND resource_id = %s",
|
||||
(client_id, resource_id))
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
# ---- services ----
|
||||
|
||||
def create_service(client_id, name, duration_minutes, price=None, active=True):
|
||||
service_id = _new_id("SV")
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO services (service_id, client_id, name, duration_minutes, "
|
||||
"price, active) VALUES (%s, %s, %s, %s, %s, %s) RETURNING *",
|
||||
(service_id, client_id, name, duration_minutes, price, active))
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
return row
|
||||
|
||||
|
||||
def get_service(client_id, service_id):
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT * FROM services WHERE client_id = %s AND service_id = %s",
|
||||
(client_id, service_id))
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
# ---- bookings ----
|
||||
|
||||
_BOOKING_UPDATABLE = {"resource_id", "customer_name", "customer_contact",
|
||||
"service", "start_time", "end_time", "status"}
|
||||
|
||||
|
||||
def create_booking(client_id, resource_id, customer_name, customer_contact,
|
||||
service, start_time, end_time, source=None, status="confirmed"):
|
||||
"""Insert a booking. Raises UnknownResource if resource_id isn't one of
|
||||
client_id's own resources (the EXCLUDE constraint only scopes overlap by
|
||||
resource_id, so this is the one place that has to stop a guessed
|
||||
resource_id from another tenant). Raises BookingConflict if it overlaps
|
||||
an existing booking on the same resource -- the Postgres EXCLUDE
|
||||
constraint is the single source of truth for that, this just turns it
|
||||
into a clean error."""
|
||||
if get_resource(client_id, resource_id) is None:
|
||||
raise UnknownResource(f"no resource {resource_id} for client {client_id}")
|
||||
booking_id = _new_id("BK")
|
||||
try:
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO bookings (booking_id, created_at, client_id, resource_id, "
|
||||
"customer_name, customer_contact, service, start_time, end_time, "
|
||||
"source, status) VALUES (%s, now(), %s, %s, %s, %s, %s, %s, %s, %s, %s) "
|
||||
"RETURNING *",
|
||||
(booking_id, client_id, resource_id, customer_name, customer_contact,
|
||||
service, start_time, end_time, source, status))
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
except psycopg.errors.ExclusionViolation as e:
|
||||
raise BookingConflict(
|
||||
f"resource {resource_id} is already booked for that time") from e
|
||||
return row
|
||||
|
||||
|
||||
def get_booking(client_id, booking_id):
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT * FROM bookings WHERE client_id = %s AND booking_id = %s",
|
||||
(client_id, booking_id))
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
def list_bookings(client_id):
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT * FROM bookings WHERE client_id = %s ORDER BY start_time",
|
||||
(client_id,))
|
||||
return cur.fetchall()
|
||||
|
||||
|
||||
def update_booking(client_id, booking_id, **fields):
|
||||
"""Update a booking scoped to client_id (e.g. reschedule/cancel).
|
||||
Returns the updated row, or None if no such booking exists for this
|
||||
client. Raises UnknownResource if fields moves the booking onto another
|
||||
tenant's resource_id. Raises BookingConflict if the update would overlap
|
||||
another booking on the same resource."""
|
||||
bad = set(fields) - _BOOKING_UPDATABLE
|
||||
if bad:
|
||||
raise ValueError(f"not updatable: {', '.join(sorted(bad))}")
|
||||
if not fields:
|
||||
return get_booking(client_id, booking_id)
|
||||
if "resource_id" in fields and get_resource(client_id, fields["resource_id"]) is None:
|
||||
raise UnknownResource(
|
||||
f"no resource {fields['resource_id']} for client {client_id}")
|
||||
setsql = ", ".join(f"{c} = %s" for c in fields)
|
||||
try:
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
f"UPDATE bookings SET {setsql} WHERE client_id = %s AND booking_id = %s "
|
||||
"RETURNING *",
|
||||
[*fields.values(), client_id, booking_id])
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
except psycopg.errors.ExclusionViolation as e:
|
||||
raise BookingConflict("resource is already booked for that time") from e
|
||||
return row
|
||||
|
||||
|
||||
# ---- users (owner login) ----
|
||||
|
||||
def create_user(client_id, email, password):
|
||||
user_id = _new_id("U")
|
||||
password_hash = generate_password_hash(password)
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO users (user_id, client_id, email, password_hash) "
|
||||
"VALUES (%s, %s, %s, %s) RETURNING *",
|
||||
(user_id, client_id, email, password_hash))
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
return row
|
||||
|
||||
|
||||
def get_user_by_email(client_id, email):
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT * FROM users WHERE client_id = %s AND email = %s",
|
||||
(client_id, email))
|
||||
return cur.fetchone()
|
||||
|
||||
|
||||
def verify_password(user, password):
|
||||
return check_password_hash(user["password_hash"], password)
|
||||
|
||||
|
||||
# ---- password reset ----
|
||||
|
||||
def create_password_reset_token(user_id, ttl_minutes=30):
|
||||
token = secrets.token_urlsafe(32)
|
||||
expires_at = datetime.now(timezone.utc) + timedelta(minutes=ttl_minutes)
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"INSERT INTO password_reset_tokens (token, user_id, expires_at) "
|
||||
"VALUES (%s, %s, %s) RETURNING *",
|
||||
(token, user_id, expires_at))
|
||||
row = cur.fetchone()
|
||||
conn.commit()
|
||||
return row
|
||||
|
||||
|
||||
def consume_password_reset_token(token, new_password):
|
||||
"""Validate + single-use consume a reset token, then set the new
|
||||
password. Returns the user_id on success, None if the token is
|
||||
missing/expired/already used -- the UPDATE ... WHERE used_at IS NULL
|
||||
makes the consume-and-invalidate step atomic."""
|
||||
with db.connect() as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"UPDATE password_reset_tokens SET used_at = now() "
|
||||
"WHERE token = %s AND used_at IS NULL AND expires_at > now() "
|
||||
"RETURNING user_id",
|
||||
(token,))
|
||||
row = cur.fetchone()
|
||||
if row is None:
|
||||
conn.commit()
|
||||
return None
|
||||
user_id = row["user_id"]
|
||||
cur.execute(
|
||||
"UPDATE users SET password_hash = %s WHERE user_id = %s",
|
||||
(generate_password_hash(new_password), user_id))
|
||||
conn.commit()
|
||||
return user_id
|
||||
@@ -0,0 +1,2 @@
|
||||
-r requirements.txt
|
||||
pytest==9.1.1
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Spins up a throwaway Postgres 16 container (matching production) and
|
||||
applies backoffice/db/init.sql against it, per the module's testing decision
|
||||
(#14): real Postgres, no mocking, so the EXCLUDE constraint and tenancy
|
||||
filters are exercised for real, not asserted by inspection.
|
||||
"""
|
||||
import atexit
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
_CONTAINER = f"smb-booking-test-db-{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
def _free_port():
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("127.0.0.1", 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
def _wait_ready(url, timeout=30):
|
||||
deadline = time.time() + timeout
|
||||
last_err = None
|
||||
while time.time() < deadline:
|
||||
try:
|
||||
with psycopg.connect(url, connect_timeout=2):
|
||||
return
|
||||
except psycopg.OperationalError as e:
|
||||
last_err = e
|
||||
time.sleep(0.5)
|
||||
raise RuntimeError(f"test db never became ready: {last_err}")
|
||||
|
||||
|
||||
def _start_db():
|
||||
port = _free_port()
|
||||
subprocess.run(
|
||||
["docker", "run", "-d", "--rm", "--name", _CONTAINER,
|
||||
"-e", "POSTGRES_DB=smbcrm_test",
|
||||
"-e", "POSTGRES_USER=smbcrm",
|
||||
"-e", "POSTGRES_PASSWORD=test",
|
||||
"-p", f"127.0.0.1:{port}:5432",
|
||||
"postgres:16-alpine"],
|
||||
check=True, capture_output=True)
|
||||
atexit.register(
|
||||
lambda: subprocess.run(["docker", "stop", _CONTAINER], capture_output=True))
|
||||
url = f"postgresql://smbcrm:test@127.0.0.1:{port}/smbcrm_test"
|
||||
_wait_ready(url)
|
||||
schema_path = os.path.join(os.path.dirname(__file__), "..", "..", "db", "init.sql")
|
||||
with open(schema_path) as f:
|
||||
schema = f.read()
|
||||
with psycopg.connect(url) as conn, conn.cursor() as cur:
|
||||
cur.execute(schema)
|
||||
conn.commit()
|
||||
return url
|
||||
|
||||
|
||||
DATABASE_URL = _start_db()
|
||||
os.environ["DATABASE_URL"] = DATABASE_URL
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clean_tables():
|
||||
with psycopg.connect(DATABASE_URL) as conn, conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"TRUNCATE resources, services, bookings, users, "
|
||||
"password_reset_tokens RESTART IDENTITY CASCADE")
|
||||
conn.commit()
|
||||
yield
|
||||
@@ -0,0 +1,202 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
import booking_db as bdb
|
||||
|
||||
CLIENT_A = "C-TEST-A"
|
||||
CLIENT_B = "C-TEST-B"
|
||||
|
||||
|
||||
def _dt(hour, minute=0):
|
||||
return datetime(2026, 8, 3, hour, minute, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
# ---- resources / services ----
|
||||
|
||||
def test_create_and_get_resource():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
assert r["client_id"] == CLIENT_A
|
||||
assert r["name"] == "Chair 1"
|
||||
assert r["active"] is True
|
||||
assert bdb.get_resource(CLIENT_A, r["resource_id"])["resource_id"] == r["resource_id"]
|
||||
|
||||
|
||||
def test_get_resource_is_tenant_scoped():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
assert bdb.get_resource(CLIENT_B, r["resource_id"]) is None
|
||||
|
||||
|
||||
def test_create_and_get_service():
|
||||
s = bdb.create_service(CLIENT_A, "Haircut", 30, price=25)
|
||||
assert s["duration_minutes"] == 30
|
||||
assert bdb.get_service(CLIENT_A, s["service_id"])["name"] == "Haircut"
|
||||
|
||||
|
||||
def test_get_service_is_tenant_scoped():
|
||||
s = bdb.create_service(CLIENT_A, "Haircut", 30, price=25)
|
||||
assert bdb.get_service(CLIENT_B, s["service_id"]) is None
|
||||
|
||||
|
||||
# ---- bookings: double-booking protection ----
|
||||
|
||||
def test_create_booking_succeeds():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "alice@example.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
assert b["status"] == "confirmed"
|
||||
assert b["resource_id"] == r["resource_id"]
|
||||
|
||||
|
||||
def test_overlapping_booking_same_resource_raises_conflict():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
with pytest.raises(bdb.BookingConflict):
|
||||
bdb.create_booking(CLIENT_A, r["resource_id"], "Bob", "b@x.com",
|
||||
"Haircut", _dt(10, 30), _dt(11, 30))
|
||||
|
||||
|
||||
def test_adjacent_non_overlapping_bookings_both_succeed():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
b2 = bdb.create_booking(CLIENT_A, r["resource_id"], "Bob", "b@x.com",
|
||||
"Haircut", _dt(11), _dt(12))
|
||||
assert b2["start_time"] == _dt(11)
|
||||
|
||||
|
||||
def test_create_booking_rejects_resource_from_another_client():
|
||||
other = bdb.create_resource(CLIENT_B, "Chair 1")
|
||||
with pytest.raises(bdb.UnknownResource):
|
||||
bdb.create_booking(CLIENT_A, other["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
|
||||
|
||||
def test_update_booking_rejects_moving_to_another_clients_resource():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
other = bdb.create_resource(CLIENT_B, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
with pytest.raises(bdb.UnknownResource):
|
||||
bdb.update_booking(CLIENT_A, b["booking_id"], resource_id=other["resource_id"])
|
||||
|
||||
|
||||
def test_create_pending_booking():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11), status="pending")
|
||||
assert bdb.get_booking(CLIENT_A, b["booking_id"])["status"] == "pending"
|
||||
|
||||
|
||||
def test_overlap_on_different_resource_succeeds():
|
||||
r1 = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
r2 = bdb.create_resource(CLIENT_A, "Chair 2")
|
||||
bdb.create_booking(CLIENT_A, r1["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
b2 = bdb.create_booking(CLIENT_A, r2["resource_id"], "Bob", "b@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
assert b2["resource_id"] == r2["resource_id"]
|
||||
|
||||
|
||||
def test_reschedule_into_conflict_raises_and_leaves_original_untouched():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
b2 = bdb.create_booking(CLIENT_A, r["resource_id"], "Bob", "b@x.com",
|
||||
"Haircut", _dt(12), _dt(13))
|
||||
with pytest.raises(bdb.BookingConflict):
|
||||
bdb.update_booking(CLIENT_A, b2["booking_id"], start_time=_dt(10, 30),
|
||||
end_time=_dt(11, 30))
|
||||
unchanged = bdb.get_booking(CLIENT_A, b2["booking_id"])
|
||||
assert unchanged["start_time"] == _dt(12)
|
||||
|
||||
|
||||
def test_reschedule_to_free_slot_succeeds():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
updated = bdb.update_booking(CLIENT_A, b["booking_id"], start_time=_dt(14),
|
||||
end_time=_dt(15))
|
||||
assert updated["start_time"] == _dt(14)
|
||||
|
||||
|
||||
def test_update_booking_rejects_non_updatable_field():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
with pytest.raises(ValueError):
|
||||
bdb.update_booking(CLIENT_A, b["booking_id"], created_at=_dt(9))
|
||||
|
||||
|
||||
# ---- bookings: tenancy isolation ----
|
||||
|
||||
def test_get_booking_is_tenant_scoped_even_with_correct_id():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
assert bdb.get_booking(CLIENT_B, b["booking_id"]) is None
|
||||
|
||||
|
||||
def test_update_booking_cannot_touch_other_clients_booking():
|
||||
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
result = bdb.update_booking(CLIENT_B, b["booking_id"], status="cancelled")
|
||||
assert result is None
|
||||
assert bdb.get_booking(CLIENT_A, b["booking_id"])["status"] == "confirmed"
|
||||
|
||||
|
||||
def test_list_bookings_only_returns_own_client():
|
||||
ra = bdb.create_resource(CLIENT_A, "Chair 1")
|
||||
rb = bdb.create_resource(CLIENT_B, "Chair 1")
|
||||
bdb.create_booking(CLIENT_A, ra["resource_id"], "Alice", "a@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
bdb.create_booking(CLIENT_B, rb["resource_id"], "Zoe", "z@x.com",
|
||||
"Haircut", _dt(10), _dt(11))
|
||||
rows = bdb.list_bookings(CLIENT_A)
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["customer_name"] == "Alice"
|
||||
|
||||
|
||||
# ---- users / owner login ----
|
||||
|
||||
def test_create_user_and_verify_password():
|
||||
u = bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
|
||||
assert bdb.verify_password(u, "correct horse")
|
||||
assert not bdb.verify_password(u, "wrong password")
|
||||
|
||||
|
||||
def test_get_user_by_email_is_tenant_scoped():
|
||||
bdb.create_user(CLIENT_A, "owner@example.com", "pw12345")
|
||||
assert bdb.get_user_by_email(CLIENT_B, "owner@example.com") is None
|
||||
assert bdb.get_user_by_email(CLIENT_A, "owner@example.com") is not None
|
||||
|
||||
|
||||
# ---- password reset: single-use semantics ----
|
||||
|
||||
def test_consume_password_reset_token_sets_new_password():
|
||||
u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
|
||||
tok = bdb.create_password_reset_token(u["user_id"])
|
||||
user_id = bdb.consume_password_reset_token(tok["token"], "new-password")
|
||||
assert user_id == u["user_id"]
|
||||
refreshed = bdb.get_user_by_email(CLIENT_A, "owner@example.com")
|
||||
assert bdb.verify_password(refreshed, "new-password")
|
||||
assert not bdb.verify_password(refreshed, "old-password")
|
||||
|
||||
|
||||
def test_consume_password_reset_token_is_single_use():
|
||||
u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
|
||||
tok = bdb.create_password_reset_token(u["user_id"])
|
||||
assert bdb.consume_password_reset_token(tok["token"], "new-password") == u["user_id"]
|
||||
assert bdb.consume_password_reset_token(tok["token"], "another-password") is None
|
||||
|
||||
|
||||
def test_consume_expired_password_reset_token_fails():
|
||||
u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
|
||||
tok = bdb.create_password_reset_token(u["user_id"], ttl_minutes=-1)
|
||||
assert bdb.consume_password_reset_token(tok["token"], "new-password") is None
|
||||
|
||||
|
||||
def test_consume_unknown_token_fails():
|
||||
assert bdb.consume_password_reset_token("not-a-real-token", "new-password") is None
|
||||
Reference in New Issue
Block a user