mivanchenko
c3e520aabb
Test backoffice (smb-crm) / test (push) Successful in 1m43s
Public booking API now rejects a 6th active booking from the same customer_contact within 24h (429), stopping one contact from filling every slot on every resource, while owner-entered manual bookings stay unaffected. Add POST /api/contact: client sites can reach their own owner's inbox directly (via their existing login email) for general inquiries, separate from the agency's leads/Telegram pipeline (n8n/lead-intake.json), which stays reserved for actual prospects contacting the agency itself. Paris Barber Shop's contact form and Rückruf widget now point here; the Rückruf floating widget itself has been removed from the site. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
415 lines
19 KiB
Python
415 lines
19 KiB
Python
"""Flask test client / real-DB integration tests for the booking API (#16),
|
|
per #14's testing decision: assert on HTTP response + resulting DB state,
|
|
not on which internal function got called.
|
|
"""
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
import booking_db as bdb
|
|
from app import app as flask_app
|
|
|
|
CLIENT_A = "C-TEST-A"
|
|
CLIENT_B = "C-TEST-B"
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
flask_app.config["TESTING"] = True
|
|
return flask_app.test_client()
|
|
|
|
|
|
def _next_monday(after):
|
|
d = after + timedelta(days=1)
|
|
while d.weekday() != 0:
|
|
d += timedelta(days=1)
|
|
return d
|
|
|
|
|
|
def _setup_resource_and_service(client_id=CLIENT_A, auto_confirm=True, **resource_kwargs):
|
|
with bdb.db.connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO clients (client_id, timezone, auto_confirm) VALUES (%s, %s, %s) "
|
|
"ON CONFLICT (client_id) DO UPDATE SET timezone = EXCLUDED.timezone, "
|
|
"auto_confirm = EXCLUDED.auto_confirm",
|
|
(client_id, "Europe/Berlin", auto_confirm))
|
|
conn.commit()
|
|
location = bdb.create_location(client_id, "Main")
|
|
resource = bdb.create_resource(client_id, location["location_id"], "Chair 1", **resource_kwargs)
|
|
bdb.set_resource_hours(client_id, resource["resource_id"], 0, time(9, 0), time(17, 0))
|
|
service = bdb.create_service(client_id, "Haircut", 60, price=25)
|
|
return resource, service
|
|
|
|
|
|
def test_slots_endpoint_lists_bookable_starts(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
resp = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()})
|
|
assert resp.status_code == 200
|
|
body = resp.get_json()
|
|
assert len(body["slots"]) == 8 # 09:00-17:00, 60min slots
|
|
|
|
|
|
def test_create_booking_auto_confirm_true_yields_confirmed(client):
|
|
resource, service = _setup_resource_and_service(
|
|
auto_confirm=True, min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[0],
|
|
"customer_name": "Alice", "customer_contact": "alice@example.com"})
|
|
assert resp.status_code == 201
|
|
body = resp.get_json()
|
|
assert body["status"] == "confirmed"
|
|
assert "token" in body
|
|
|
|
|
|
def test_create_booking_via_public_endpoint_sends_confirmation_email(client, monkeypatch):
|
|
"""#18's acceptance criterion: completing a booking via ticket 3's public
|
|
page (this same POST /api/booking endpoint) triggers the confirmation
|
|
email, with the manage-booking token embedded in it."""
|
|
sent = []
|
|
monkeypatch.setattr(
|
|
"booking_mail.mailer.send_email",
|
|
lambda to, subject, html, from_addr=None: sent.append(
|
|
{"to": to, "subject": subject, "html": html, "from_addr": from_addr}))
|
|
resource, service = _setup_resource_and_service(
|
|
auto_confirm=True, min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[0],
|
|
"customer_name": "Kim", "customer_contact": "kim@example.com"})
|
|
assert resp.status_code == 201
|
|
token = resp.get_json()["token"]
|
|
|
|
assert len(sent) == 1
|
|
assert sent[0]["to"] == "kim@example.com"
|
|
assert f"/manage/{token}" in sent[0]["html"]
|
|
assert "Haircut" in sent[0]["html"]
|
|
|
|
|
|
def test_create_booking_notifies_owner(client, monkeypatch):
|
|
"""#23's acceptance criterion: creating a booking via the public API
|
|
triggers the owner notification webhook, carrying the client's
|
|
notify_channel and business_name alongside the booking."""
|
|
notified = []
|
|
monkeypatch.setattr(
|
|
"owner_notify.notify",
|
|
lambda client, booking, event: notified.append((client, booking, event)))
|
|
resource, service = _setup_resource_and_service(
|
|
auto_confirm=True, min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[0],
|
|
"customer_name": "Nia", "customer_contact": "nia@example.com"})
|
|
assert resp.status_code == 201
|
|
booking_id = resp.get_json()["booking_id"]
|
|
|
|
assert len(notified) == 1
|
|
notified_client, notified_booking, notified_event = notified[0]
|
|
assert notified_client["client_id"] == CLIENT_A
|
|
assert notified_booking["booking_id"] == booking_id
|
|
assert notified_booking["customer_name"] == "Nia"
|
|
assert notified_event == "created"
|
|
|
|
|
|
def test_create_booking_auto_confirm_false_yields_pending(client):
|
|
resource, service = _setup_resource_and_service(
|
|
auto_confirm=False, min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[0],
|
|
"customer_name": "Bob", "customer_contact": "bob@example.com"})
|
|
assert resp.status_code == 201
|
|
assert resp.get_json()["status"] == "pending"
|
|
|
|
|
|
def test_create_booking_rejects_slot_outside_business_rules(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
# 20:00 is outside the 09:00-17:00 hours configured above.
|
|
outside = datetime.combine(day, time(20, 0), tzinfo=timezone.utc).isoformat()
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": outside,
|
|
"customer_name": "Carl", "customer_contact": "c@example.com"})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_concurrent_booking_requests_only_one_succeeds(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"][0]
|
|
|
|
payload = {"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot,
|
|
"customer_name": "Race1", "customer_contact": "r1@example.com"}
|
|
first = client.post("/api/booking", json=payload)
|
|
payload2 = dict(payload, customer_name="Race2", customer_contact="r2@example.com")
|
|
second = client.post("/api/booking", json=payload2)
|
|
|
|
statuses = sorted([first.status_code, second.status_code])
|
|
assert statuses == [201, 409]
|
|
assert len(bdb.list_bookings(CLIENT_A)) == 1
|
|
|
|
|
|
def test_contact_rate_limit_blocks_after_five_recent_bookings(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
assert len(slots) >= 6 # 09:00-17:00, 60min slots -- 8 available
|
|
|
|
for slot in slots[:5]:
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot,
|
|
"customer_name": "Serial Booker", "customer_contact": "serial@example.com"})
|
|
assert resp.status_code == 201
|
|
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[5],
|
|
"customer_name": "Serial Booker", "customer_contact": "serial@example.com"})
|
|
assert resp.status_code == 429
|
|
assert len(bdb.list_bookings(CLIENT_A)) == 5
|
|
|
|
# A different contact is unaffected by the first contact's count.
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[5],
|
|
"customer_name": "Someone Else", "customer_contact": "else@example.com"})
|
|
assert resp.status_code == 201
|
|
|
|
|
|
def test_cancel_with_valid_token_cancels_booking(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"][0]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot,
|
|
"customer_name": "Dana", "customer_contact": "d@example.com"}).get_json()
|
|
|
|
resp = client.post("/api/booking/cancel", json={"token": created["token"]})
|
|
assert resp.status_code == 200
|
|
assert bdb.get_booking(CLIENT_A, created["booking_id"])["status"] == "cancelled"
|
|
|
|
# already-used: cancelling an already-cancelled booking is rejected, not
|
|
# silently repeated.
|
|
resp2 = client.post("/api/booking/cancel", json={"token": created["token"]})
|
|
assert resp2.status_code == 409
|
|
|
|
# garbage token is rejected cleanly, not a 500.
|
|
resp3 = client.post("/api/booking/cancel", json={"token": "not-a-real-token"})
|
|
assert resp3.status_code == 400
|
|
|
|
|
|
def test_cancel_notifies_owner(client, monkeypatch):
|
|
notified = []
|
|
monkeypatch.setattr(
|
|
"owner_notify.notify",
|
|
lambda client, booking, event: notified.append((client, booking, event)))
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"][0]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot,
|
|
"customer_name": "Omar", "customer_contact": "o@example.com"}).get_json()
|
|
notified.clear() # drop the create-time notification, isolate the cancel one
|
|
|
|
resp = client.post("/api/booking/cancel", json={"token": created["token"]})
|
|
assert resp.status_code == 200
|
|
|
|
assert len(notified) == 1
|
|
notified_client, notified_booking, notified_event = notified[0]
|
|
assert notified_client["client_id"] == CLIENT_A
|
|
assert notified_booking["booking_id"] == created["booking_id"]
|
|
assert notified_booking["status"] == "cancelled"
|
|
assert notified_event == "cancelled"
|
|
|
|
|
|
def test_reschedule_notifies_owner(client, monkeypatch):
|
|
notified = []
|
|
monkeypatch.setattr(
|
|
"owner_notify.notify",
|
|
lambda client, booking, event: notified.append((client, booking, event)))
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot_list = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[0],
|
|
"customer_name": "Priya", "customer_contact": "p@example.com"}).get_json()
|
|
notified.clear() # drop the create-time notification, isolate the reschedule one
|
|
|
|
resp = client.post("/api/booking/reschedule", json={
|
|
"token": created["token"], "start_time": slot_list[2]})
|
|
assert resp.status_code == 200
|
|
|
|
assert len(notified) == 1
|
|
notified_client, notified_booking, notified_event = notified[0]
|
|
assert notified_client["client_id"] == CLIENT_A
|
|
assert notified_booking["start_time"].isoformat() == slot_list[2]
|
|
assert notified_event == "rescheduled"
|
|
|
|
|
|
def test_cancel_with_expired_token_is_rejected(client):
|
|
import jwt as pyjwt
|
|
from booking_api import TOKEN_SECRET
|
|
resource, service = _setup_resource_and_service()
|
|
booking = bdb.create_booking(
|
|
CLIENT_A, resource["resource_id"], "Zara", "z@example.com", "Haircut",
|
|
datetime.now(timezone.utc) + timedelta(days=1),
|
|
datetime.now(timezone.utc) + timedelta(days=1, hours=1))
|
|
expired = pyjwt.encode(
|
|
{"client_id": CLIENT_A, "booking_id": booking["booking_id"],
|
|
"exp": datetime.now(timezone.utc) - timedelta(minutes=1)},
|
|
TOKEN_SECRET, algorithm="HS256")
|
|
resp = client.post("/api/booking/cancel", json={"token": expired})
|
|
assert resp.status_code == 400
|
|
assert bdb.get_booking(CLIENT_A, booking["booking_id"])["status"] != "cancelled"
|
|
|
|
|
|
def test_reschedule_of_already_cancelled_booking_is_rejected(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot_list = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[0],
|
|
"customer_name": "Jan", "customer_contact": "j@example.com"}).get_json()
|
|
client.post("/api/booking/cancel", json={"token": created["token"]})
|
|
|
|
resp = client.post("/api/booking/reschedule", json={
|
|
"token": created["token"], "start_time": slot_list[1]})
|
|
assert resp.status_code == 409
|
|
|
|
|
|
def test_reschedule_with_valid_token_updates_time(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot_list = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[0],
|
|
"customer_name": "Eve", "customer_contact": "e@example.com"}).get_json()
|
|
|
|
resp = client.post("/api/booking/reschedule", json={
|
|
"token": created["token"], "start_time": slot_list[2]})
|
|
assert resp.status_code == 200
|
|
booking = bdb.get_booking(CLIENT_A, created["booking_id"])
|
|
assert booking["start_time"].isoformat() == slot_list[2]
|
|
|
|
|
|
def test_reschedule_into_occupied_slot_fails_cleanly(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot_list = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
|
|
first = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[0],
|
|
"customer_name": "Fay", "customer_contact": "f@example.com"}).get_json()
|
|
second = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[1],
|
|
"customer_name": "Gus", "customer_contact": "g@example.com"}).get_json()
|
|
|
|
resp = client.post("/api/booking/reschedule", json={
|
|
"token": second["token"], "start_time": slot_list[0]})
|
|
assert resp.status_code == 409
|
|
assert bdb.get_booking(CLIENT_A, second["booking_id"])["start_time"].isoformat() \
|
|
== slot_list[1]
|
|
|
|
|
|
def test_reschedule_to_same_slot_is_a_noop_success(client):
|
|
resource, service = _setup_resource_and_service(
|
|
min_notice_minutes=0, max_advance_days=365)
|
|
day = _next_monday(date.today())
|
|
slot_list = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
created = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slot_list[0],
|
|
"customer_name": "Hana", "customer_contact": "h@example.com"}).get_json()
|
|
|
|
resp = client.post("/api/booking/reschedule", json={
|
|
"token": created["token"], "start_time": slot_list[0]})
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_manage_token_is_scoped_to_its_own_client():
|
|
resource, service = _setup_resource_and_service(client_id=CLIENT_A)
|
|
from booking_api import _mint_manage_token, verify_manage_token
|
|
booking = bdb.create_booking(
|
|
CLIENT_A, resource["resource_id"], "Ivy", "i@example.com", "Haircut",
|
|
datetime.now(timezone.utc) + timedelta(days=1),
|
|
datetime.now(timezone.utc) + timedelta(days=1, hours=1))
|
|
token = _mint_manage_token(CLIENT_A, booking["booking_id"])
|
|
assert verify_manage_token(token) == (CLIENT_A, booking["booking_id"])
|
|
tampered = token[:-1] + ("A" if token[-1] != "A" else "B")
|
|
assert verify_manage_token(tampered) is None
|