mivanchenko
2f6e0c1459
Adds the plumbing that makes "can a customer actually get booked" true end to end at the API layer, on top of #15's schema/tenancy layer. - resource_hours table + min_notice_minutes/max_advance_days/buffer_minutes on resources -- config #15 didn't include but #16 depends on. - availability.py: pure slot-generation function, correct across a Europe/Berlin DST transition (tested both directions). - booking_api.py: JSON blueprint for slot listing, booking creation (auto_confirm -> confirmed/pending), and signed-JWT cancel/reschedule, registered into app.py. - booking_db.py gains resource-hours CRUD, a tenant-scoped busy-bookings query for buffer/slot validation, and a read-only client lookup. A true concurrent-threads test (not just sequential requests) surfaced a real gap: Postgres can raise DeadlockDetected instead of ExclusionViolation when two overlapping inserts race the exclusion constraint directly, which went uncaught and would have 500'd instead of giving the clean 4xx the ticket requires -- now caught alongside ExclusionViolation. Also fixed: reschedule used the request's raw UTC offset to pick the business day instead of the client's own timezone (could pick the wrong day's hours/bookings near local midnight); the cancel/reschedule JWT no longer falls back to reusing CRM_API_TOKEN as its signing secret. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
237 lines
9.3 KiB
Python
237 lines
9.3 KiB
Python
import threading
|
|
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_truly_concurrent_overlapping_inserts_only_one_wins():
|
|
"""Two threads, each on its own DB connection, racing to insert the same
|
|
overlapping slot -- not just a sequential second-request-fails check.
|
|
The Postgres EXCLUDE constraint (not app-level locking) is what has to
|
|
serialize this."""
|
|
r = bdb.create_resource(CLIENT_A, "Chair 1")
|
|
start_barrier = threading.Barrier(2)
|
|
results = {}
|
|
|
|
def attempt(name, customer):
|
|
start_barrier.wait()
|
|
try:
|
|
results[name] = bdb.create_booking(
|
|
CLIENT_A, r["resource_id"], customer, f"{customer}@x.com",
|
|
"Haircut", _dt(10), _dt(11))
|
|
except bdb.BookingConflict as e:
|
|
results[name] = e
|
|
|
|
t1 = threading.Thread(target=attempt, args=("t1", "Race1"))
|
|
t2 = threading.Thread(target=attempt, args=("t2", "Race2"))
|
|
t1.start()
|
|
t2.start()
|
|
t1.join()
|
|
t2.join()
|
|
|
|
outcomes = list(results.values())
|
|
conflicts = [o for o in outcomes if isinstance(o, bdb.BookingConflict)]
|
|
successes = [o for o in outcomes if not isinstance(o, bdb.BookingConflict)]
|
|
assert len(conflicts) == 1
|
|
assert len(successes) == 1
|
|
assert len(bdb.list_bookings(CLIENT_A)) == 1
|
|
|
|
|
|
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
|