aiolabs/chatelet
9
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: aiolabs:efd46a72c7b09921cf5f0eb6e69af2ae59d97ead
Branches Tags
aiolabs:main
aiolabs:v0.3.0
aiolabs:v0.2.0
aiolabs:v0.1.0
..
compare: aiolabs:27863db444019ae6c982fad01dd541012a578339
Branches Tags
aiolabs:main
aiolabs:v0.3.0
aiolabs:v0.2.0
aiolabs:v0.1.0

No commits in common. "efd46a72c7b09921cf5f0eb6e69af2ae59d97ead" and "27863db444019ae6c982fad01dd541012a578339" have entirely different histories.
efd46a72c7 ... 27863db444

7 changed files with 0 additions and 438 deletions
Download patch file Download diff file Expand all files Collapse all files

25
Makefile
View file

@ -1,25 +0,0 @@
all: format check
format: black ruff
check: mypy checkblack checkruff
mypy:
poetry run mypy .
black:
poetry run black .
ruff:
poetry run ruff check . --fix
checkruff:
poetry run ruff check .
checkblack:
poetry run black --check .
test:
PYTHONUNBUFFERED=1 \
DEBUG=true \
poetry run pytest

60
pyproject.toml
View file

@ -1,60 +0,0 @@
[tool.poetry]
name = "chatelet"
version = "0.1.0"
description = "Nostr-native room rentals (Airbnb-style) for LNbits"
authors = ["padreug <padreug@aiolabs.dev>"]
[tool.poetry.dependencies]
python = "^3.10 | ^3.9"
lnbits = {version = "*", allow-prereleases = true}
mypy = "^1.13.0"
[tool.poetry.group.dev.dependencies]
black = "^24.3.0"
pytest-asyncio = "^0.21.0"
pytest = "^7.3.2"
mypy = "^1.5.1"
pre-commit = "^3.2.2"
ruff = "^0.3.2"
pytest-md = "^0.2.0"
[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"
[tool.mypy]
[[tool.mypy.overrides]]
module = [
"lnbits.*",
"loguru.*",
"fastapi.*",
"pydantic.*",
]
ignore_missing_imports = "True"
[tool.pytest.ini_options]
log_cli = false
testpaths = [
"tests"
]
[tool.black]
line-length = 88
[tool.ruff]
line-length = 88
[tool.ruff.lint]
# F pyflakes / E,W pycodestyle / I isort / A builtins / C mccabe / N naming
# UP pyupgrade / RUF ruff / B bugbear
select = ["F", "E", "W", "I", "A", "C", "N", "UP", "RUF", "B"]
ignore = ["UP007"] # keep Optional[X] on py3.10
fixable = ["ALL"]
unfixable = []
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.lint.flake8-bugbear]
extend-immutable-calls = [
"fastapi.Depends",
"fastapi.Query",
]

0
tests/__init__.py
View file

88
tests/conftest.py
View file

@ -1,88 +0,0 @@
"""Pytest configuration + factories for the chatelet test suite.
Following the spirekeeper pattern: unit tests isolate the function under
test by monkeypatching the crud / lnbits calls it makes and driving async
functions with `asyncio.run(...)`, so no live LNbits DB or wallet is needed.
The `loguru_capture` fixture is here for the same reason spirekeeper needs
it — loguru binds sys.stderr at import, before capsys wraps it.
"""
from typing import Generator, List
import pytest
from loguru import logger
from ..models import (
BookingRequestData,
CreateRoomData,
Room,
RoomStatus,
)
@pytest.fixture(autouse=True)
def _reset_room_locks():
"""Each test drives its own `asyncio.run` loop, but `services._room_locks`
is module-level — a lock created in one test's loop would raise
'bound to a different event loop' when reused in the next. Clear the
registry around every test so locks are recreated in the active loop.
(Production is unaffected: LNbits runs one long-lived loop.)"""
from .. import services
services._room_locks.clear()
yield
services._room_locks.clear()
@pytest.fixture
def loguru_capture() -> Generator[List[str], None, None]:
captured: List[str] = []
handler_id = logger.add(
captured.append, level="WARNING", format="{level} {message}"
)
yield captured
logger.remove(handler_id)
# --- factories -------------------------------------------------------------
def make_room(
room_id: str = "room1",
*,
wallet: str = "wallet1",
currency: str = "sat", # "sat" so pricing skips the FX call in tests
price: float = 100.0,
min_nights: int = 1,
max_guests: int = 2,
status: RoomStatus = RoomStatus.active,
) -> Room:
return Room(
**CreateRoomData(
wallet=wallet,
title="Tower Room",
price_amount=price,
price_currency=currency,
min_nights=min_nights,
max_guests=max_guests,
).dict(),
id=room_id,
status=status,
)
def make_request(
room_id: str = "room1",
*,
guest: str = "npub_guest",
check_in: str = "2026-08-01",
check_out: str = "2026-08-04",
num_guests: int = 1,
) -> BookingRequestData:
return BookingRequestData(
room_id=room_id,
guest_pubkey=guest,
check_in=check_in,
check_out=check_out,
num_guests=num_guests,
)

96
tests/test_atomic_hold.py
View file

@ -1,96 +0,0 @@
"""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

87
tests/test_availability.py
View file

@ -1,87 +0,0 @@
"""Availability arbiter: half-open overlap semantics + is_available()."""
import asyncio
from .. import crud
from ..models import Block, Booking, BookingStatus, RoomStatus
from .conftest import make_room
# --- pure interval logic ---------------------------------------------------
class TestOverlap:
def test_disjoint_ranges_do_not_overlap(self):
assert not crud._overlaps("2026-08-01", "2026-08-04", "2026-08-04", "2026-08-06")
def test_back_to_back_same_day_does_not_overlap(self):
# check_out is exclusive: one guest leaves 08-04, next arrives 08-04.
assert not crud._overlaps("2026-08-01", "2026-08-04", "2026-08-04", "2026-08-07")
def test_partial_overlap(self):
assert crud._overlaps("2026-08-01", "2026-08-05", "2026-08-04", "2026-08-08")
def test_full_containment(self):
assert crud._overlaps("2026-08-01", "2026-08-10", "2026-08-03", "2026-08-05")
def test_nights_between(self):
assert crud.nights_between("2026-08-01", "2026-08-04") == 3
# --- is_available (crud getters monkeypatched) -----------------------------
def _patch(monkeypatch, *, room, bookings=None, blocks=None):
async def _get_room(_):
return room
async def _get_bookings(_):
return bookings or []
async def _get_blocks(_):
return blocks or []
monkeypatch.setattr(crud, "get_room", _get_room)
monkeypatch.setattr(crud, "get_bookings_for_room", _get_bookings)
monkeypatch.setattr(crud, "get_blocks_for_room", _get_blocks)
def _booking(status: BookingStatus, ci="2026-08-02", co="2026-08-05") -> Booking:
return Booking(
id="b1", room_id="room1", guest_pubkey="g", check_in=ci, check_out=co,
nights=3, num_guests=1, currency="sat", price_fiat=300.0,
amount_sat=300, deposit_sat=300, status=status,
)
class TestIsAvailable:
def test_open_range_is_available(self, monkeypatch):
_patch(monkeypatch, room=make_room())
assert asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))
def test_inactive_room_is_never_available(self, monkeypatch):
_patch(monkeypatch, room=make_room(status=RoomStatus.inactive))
assert not asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))
def test_overlapping_confirmed_booking_blocks(self, monkeypatch):
_patch(monkeypatch, room=make_room(),
bookings=[_booking(BookingStatus.confirmed)])
assert not asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))
def test_overlapping_held_booking_blocks(self, monkeypatch):
# A mere hold occupies the calendar — this is what makes the atomic
# check-then-hold work.
_patch(monkeypatch, room=make_room(),
bookings=[_booking(BookingStatus.held)])
assert not asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))
def test_cancelled_booking_does_not_block(self, monkeypatch):
_patch(monkeypatch, room=make_room(),
bookings=[_booking(BookingStatus.cancelled)])
assert asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))
def test_manual_block_blocks(self, monkeypatch):
blk = Block(id="k1", room_id="room1", start_date="2026-08-03",
end_date="2026-08-06", reason="maintenance")
_patch(monkeypatch, room=make_room(), blocks=[blk])
assert not asyncio.run(crud.is_available("room1", "2026-08-01", "2026-08-04"))

82
tests/test_booking_flow.py
View file

@ -1,82 +0,0 @@
"""Booking flow through services.request_booking: canonical amount, invoice
tagging, and hold-release on invoice failure."""
import asyncio
from types import SimpleNamespace
import pytest
from lnbits.exceptions import InvoiceError
from .. import crud, services
from ..models import BookingQuote, BookingStatus
from .conftest import make_request, make_room
def _setup(monkeypatch, room, *, invoice_raises=False):
created: list = [] # bookings passed to create_booking
updated: list = [] # bookings passed to update_booking (captures final state)
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(*_):
return True
async def fake_create_booking(booking):
# Snapshot the status at creation time — `booking` is mutated in place
# later (held -> declined), so the reference alone can't prove it was
# ever held.
created.append(SimpleNamespace(id=booking.id, status=booking.status))
return booking
async def fake_update_booking(booking):
updated.append(booking)
return booking
async def fake_create_invoice(**kwargs):
if invoice_raises:
raise InvoiceError("no funding source")
return SimpleNamespace(payment_hash="ph_1", 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 created, updated
def test_happy_path_holds_then_awaits_payment(monkeypatch):
# sat currency + price 100, 3 nights -> canonical amount_sat = 300.
_setup(monkeypatch, make_room(price=100.0, currency="sat"))
quote = asyncio.run(services.request_booking(make_request()))
assert isinstance(quote, BookingQuote)
assert quote.booking.amount_sat == 300 # 100 * 3 nights, canonical
assert quote.booking.deposit_sat == 300 # deposit_percent 100
assert quote.booking.status == BookingStatus.awaiting_payment
assert quote.booking.payment_hash == "ph_1"
assert quote.payment_request == "lnbc_fake"
def test_min_nights_enforced(monkeypatch):
_setup(monkeypatch, make_room(min_nights=5))
with pytest.raises(ValueError, match="Minimum stay"):
asyncio.run(services.request_booking(make_request())) # 3 nights < 5
def test_invoice_failure_releases_hold(monkeypatch):
created, updated = _setup(
monkeypatch, make_room(), invoice_raises=True
)
with pytest.raises(services.BookingError):
asyncio.run(services.request_booking(make_request()))
# The hold was written, then flipped to declined so the dates free up —
# a dead hold must not block the calendar.
assert created and created[0].status == BookingStatus.held
assert updated and updated[-1].status == BookingStatus.declined