test: booking-flow, availability, and #4 concurrency regression

16 tests, run under the LNbits pytest env (spirekeeper pattern: monkeypatch
crud/invoice, drive async via asyncio.run — no live DB/wallet):

- test_availability: half-open overlap semantics (back-to-back stays OK),
  and is_available blocking on held/confirmed bookings + manual blocks.
- test_booking_flow: canonical amount_sat, awaiting_payment transition,
  min-nights guard, and hold-release (declined) on InvoiceError.
- test_atomic_hold: the #4 regression — two concurrent same-date requests
  yield exactly one hold + one conflict; non-overlapping both succeed. The
  fakes yield mid-check to open the race window, so the test fails without
  the per-room lock (verified by neutering it) and passes with it.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019VUQCfdqiLSsFS2jcGnaFD
This commit is contained in:
Padreug 2026-07-19 17:48:11 +02:00
commit 7d704a8549
5 changed files with 353 additions and 0 deletions

96
tests/test_atomic_hold.py Normal file
View file

@ -0,0 +1,96 @@
"""Regression for #4 — atomic check-then-hold.
The fakes here yield control (`await asyncio.sleep`) between reading
availability and committing the hold, opening the exact race window the
per-room lock closes. Without the lock in `services.request_booking`, the
first test double-books (two quotes, zero conflicts) and fails; with it, one
request wins and the other gets `Unavailable`.
"""
import asyncio
from types import SimpleNamespace
from .. import crud, services
from ..models import BookingQuote
from .conftest import make_request, make_room
def _setup(monkeypatch, room):
"""In-memory occupancy store + async crud/invoice stubs. Returns the
(held_ranges, invoices) lists so tests can assert on them."""
held: list[tuple[str, str]] = []
invoices: list[dict] = []
async def fake_get_room(_):
return room
async def fake_settings():
return SimpleNamespace(deposit_percent=100, default_hold_minutes=30)
async def fake_is_available(room_id, ci, co):
await asyncio.sleep(0.02) # race window: let a concurrent caller interleave
return not any(crud._overlaps(ci, co, s, e) for s, e in held)
async def fake_create_booking(booking):
await asyncio.sleep(0.01)
held.append((booking.check_in, booking.check_out))
return booking
async def fake_update_booking(booking):
return booking
async def fake_create_invoice(**kwargs):
invoices.append(kwargs)
return SimpleNamespace(
payment_hash="ph_" + kwargs["extra"]["booking_id"], bolt11="lnbc_fake"
)
monkeypatch.setattr(crud, "get_room", fake_get_room)
monkeypatch.setattr(crud, "get_or_create_settings", fake_settings)
monkeypatch.setattr(crud, "is_available", fake_is_available)
monkeypatch.setattr(crud, "create_booking", fake_create_booking)
monkeypatch.setattr(crud, "update_booking", fake_update_booking)
monkeypatch.setattr(services, "create_invoice", fake_create_invoice)
return held, invoices
def test_concurrent_same_dates_yield_one_hold_one_conflict(monkeypatch):
held, invoices = _setup(monkeypatch, make_room())
async def run_two():
return await asyncio.gather(
services.request_booking(make_request(guest="g1")),
services.request_booking(make_request(guest="g2")),
return_exceptions=True,
)
results = asyncio.run(run_two())
quotes = [r for r in results if isinstance(r, BookingQuote)]
conflicts = [r for r in results if isinstance(r, services.Unavailable)]
assert len(quotes) == 1, f"expected exactly one hold, got {results}"
assert len(conflicts) == 1, f"expected exactly one conflict, got {results}"
assert len(held) == 1, "only one booking should occupy the calendar"
assert len(invoices) == 1, "only the winner should be invoiced"
def test_concurrent_non_overlapping_both_succeed(monkeypatch):
"""The lock must not over-serialize: two distinct date ranges on the same
room (same lock) should both go through."""
held, invoices = _setup(monkeypatch, make_room())
async def run_two():
return await asyncio.gather(
services.request_booking(
make_request(guest="g1", check_in="2026-08-01", check_out="2026-08-04")
),
services.request_booking(
make_request(guest="g2", check_in="2026-08-10", check_out="2026-08-12")
),
return_exceptions=True,
)
results = asyncio.run(run_two())
assert all(isinstance(r, BookingQuote) for r in results), results
assert len(held) == 2
assert len(invoices) == 2