mivanchenko
68430a5c0c
Test backoffice (smb-crm) / test (push) Successful in 1m40s
Rename + active-toggle already existed (this session, earlier commit) -- this adds a genuine, permanent delete alongside them, guarded by a confirm() prompt that points to the "Aktiv" checkbox as the reversible alternative for someone just temporarily off. Existing bookings against a deleted resource are left as-is (no FK in this schema, matching its existing convention) -- the owner agenda already falls back to the raw resource_id for a booking whose resource no longer resolves, same as it does today for a deactivated one. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
293 lines
12 KiB
Python
293 lines
12 KiB
Python
import threading
|
|
from datetime import datetime, time, 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)
|
|
|
|
|
|
def _loc(client_id):
|
|
return bdb.create_location(client_id, "Main")["location_id"]
|
|
|
|
|
|
# ---- resources / services ----
|
|
|
|
def test_create_and_get_resource():
|
|
r = bdb.create_resource(CLIENT_A, _loc(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, _loc(CLIENT_A), "Chair 1")
|
|
assert bdb.get_resource(CLIENT_B, r["resource_id"]) is None
|
|
|
|
|
|
def test_delete_resource_removes_it_and_its_hours():
|
|
r = bdb.create_resource(CLIENT_A, _loc(CLIENT_A), "Chair 1")
|
|
bdb.set_resource_hours(CLIENT_A, r["resource_id"], 0, time(9, 0), time(17, 0))
|
|
assert bdb.delete_resource(CLIENT_A, r["resource_id"]) == r["resource_id"]
|
|
assert bdb.get_resource(CLIENT_A, r["resource_id"]) is None
|
|
assert bdb.get_resource_hours(CLIENT_A, r["resource_id"]) == {}
|
|
|
|
|
|
def test_delete_resource_is_tenant_scoped():
|
|
r = bdb.create_resource(CLIENT_B, _loc(CLIENT_B), "Chair 1")
|
|
assert bdb.delete_resource(CLIENT_A, r["resource_id"]) is None
|
|
assert bdb.get_resource(CLIENT_B, r["resource_id"]) is not None
|
|
|
|
|
|
def test_delete_resource_leaves_existing_bookings_in_place():
|
|
r = bdb.create_resource(CLIENT_A, _loc(CLIENT_A), "Chair 1")
|
|
b = bdb.create_booking(CLIENT_A, r["resource_id"], "Alice", "a@x.com",
|
|
"Haircut", _dt(10), _dt(11))
|
|
bdb.delete_resource(CLIENT_A, r["resource_id"])
|
|
still_there = bdb.get_booking(CLIENT_A, b["booking_id"])
|
|
assert still_there is not None
|
|
assert still_there["resource_id"] == r["resource_id"]
|
|
|
|
|
|
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, _loc(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, _loc(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, _loc(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, _loc(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, _loc(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, _loc(CLIENT_A), "Chair 1")
|
|
other = bdb.create_resource(CLIENT_B, _loc(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, _loc(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():
|
|
loc = _loc(CLIENT_A)
|
|
r1 = bdb.create_resource(CLIENT_A, loc, "Chair 1")
|
|
r2 = bdb.create_resource(CLIENT_A, loc, "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, _loc(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, _loc(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, _loc(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, _loc(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, _loc(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, _loc(CLIENT_A), "Chair 1")
|
|
rb = bdb.create_resource(CLIENT_B, _loc(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
|
|
|
|
|
|
# ---- login lookup / operator listing (#19) ----
|
|
|
|
def test_find_user_by_email_is_not_tenant_scoped():
|
|
"""Login happens before client_id is known -- find_user_by_email looks up
|
|
by the globally-unique email alone, unlike get_user_by_email."""
|
|
u = bdb.create_user(CLIENT_A, "owner@example.com", "pw12345")
|
|
found = bdb.find_user_by_email("owner@example.com")
|
|
assert found["user_id"] == u["user_id"]
|
|
assert bdb.find_user_by_email("nobody@example.com") is None
|
|
|
|
|
|
def test_get_user_by_id():
|
|
u = bdb.create_user(CLIENT_A, "owner@example.com", "pw12345")
|
|
assert bdb.get_user(u["user_id"])["email"] == "owner@example.com"
|
|
assert bdb.get_user("U-does-not-exist") is None
|
|
|
|
|
|
def test_list_users_scoped_and_unscoped():
|
|
a = bdb.create_user(CLIENT_A, "a@example.com", "pw12345")
|
|
bdb.create_user(CLIENT_B, "b@example.com", "pw12345")
|
|
scoped = bdb.list_users(CLIENT_A)
|
|
assert [r["user_id"] for r in scoped] == [a["user_id"]]
|
|
everyone = bdb.list_users()
|
|
assert {r["client_id"] for r in everyone} == {CLIENT_A, CLIENT_B}
|
|
assert "password_hash" not in everyone[0]
|