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 # ---- 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]