mivanchenko
456ca3872f
Test backoffice (smb-crm) / test (push) Successful in 1m46s
Enables multiple barbers/staff bookable at the same location and time -- previously "resource" conflated "location" and "the thing that can't double-book itself" into one row, so a Filiale could only ever have exactly one bookable slot at once. - New `locations` table; `resources.location_id` with a generic, idempotent backfill migration (any resource without a location gets one auto-created matching its name -- not a one-off for any single client, protects any future resource stuck in the old flat shape too) - `resources`/`resource_hours`/services keep everything they already had (hours, min-notice, max-advance, buffer, the no-overlap constraint) scoped to resource_id, not location_id -- two barbers at one location must stay independently bookable at the same time - booking_db.py: new locations CRUD mirroring the existing resources/services pattern; create_resource now requires a location_id, guarded the same way every other tenant check here is (get_location existence check, no real FK -- matches this schema's existing no-FK convention throughout) - app.py: new POST /api/locations provisioning route; POST /api/resources now requires location_id - owner_settings.py + settings.html: new self-service "add a Filiale" / "add a barber" UI -- there was previously no way to create a resource at all outside the CRM/n8n provisioning API - public_booking.py + book.html: new Filiale picker (reuses the existing wireOptionGroup button-group pattern), filtering the Mitarbeiter picker to the selected location -- a single-location client sees no extra click, same as before Filialen existed - owner_booking.py + agenda.html: the Filiale show/hide toggle and hide-cancelled toggle (shipped earlier this session) now key off location_id instead of resource_id, so hiding a Filiale hides every barber's bookings at it; manual-booking dropdown grouped by Filiale - n8n/onboarding.json: default provisioning now creates a "Hauptfiliale" location before its resource (inert until re-imported into the live n8n instance) Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
383 lines
18 KiB
Python
383 lines
18 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_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
|