92 lines
2.9 KiB
Python
92 lines
2.9 KiB
Python
"""A guest's own bookings, on both doors. HTTP identity is the LNbits account
|
|
pubkey; RPC identity is the signed sender_pubkey. Neither leaks another
|
|
guest's rows, and the Lightning/Nostr plumbing stays internal."""
|
|
|
|
import asyncio
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from .. import crud, transport_rpcs, views_api
|
|
from ..models import Booking, BookingStatus, guest_booking_dict
|
|
|
|
PK = "ab" * 32
|
|
|
|
|
|
def _booking(i: int, **over: Any) -> Booking:
|
|
base: dict[str, Any] = {
|
|
"id": f"bk{i}",
|
|
"room_id": "a",
|
|
"guest_pubkey": PK,
|
|
"guest_contact": "me@example.com",
|
|
"check_in": f"2026-10-{10 + i:02d}",
|
|
"check_out": f"2026-10-{12 + i:02d}",
|
|
"nights": 2,
|
|
"num_guests": 1,
|
|
"currency": "EUR",
|
|
"price_fiat": 200.0,
|
|
"amount_sat": 300000,
|
|
"deposit_sat": 300000,
|
|
"status": BookingStatus.confirmed,
|
|
"payment_hash": f"ph{i}",
|
|
"request_event_id": "req",
|
|
"reservation_event_id": "res",
|
|
}
|
|
base.update(over)
|
|
return Booking(**base)
|
|
|
|
|
|
def _patch_store(monkeypatch, rows):
|
|
seen = {}
|
|
|
|
async def for_guest(pubkey, limit=200):
|
|
seen["pubkey"] = pubkey
|
|
return [b for b in rows if b.guest_pubkey == pubkey]
|
|
|
|
monkeypatch.setattr(crud, "get_bookings_for_guest", for_guest)
|
|
return seen
|
|
|
|
|
|
def test_guest_dict_keeps_own_contact_but_hides_plumbing():
|
|
d = guest_booking_dict(_booking(1))
|
|
assert d["guest_contact"] == "me@example.com"
|
|
assert d["guest_pubkey"] == PK
|
|
for hidden in ("payment_hash", "request_event_id", "reservation_event_id"):
|
|
assert hidden not in d
|
|
|
|
|
|
def test_http_lists_only_the_callers_rows(monkeypatch):
|
|
rows = [_booking(1), _booking(2, guest_pubkey="cd" * 32)]
|
|
seen = _patch_store(monkeypatch, rows)
|
|
user = SimpleNamespace(id="u1", pubkey=PK)
|
|
out = asyncio.run(views_api.api_my_bookings(user=user))
|
|
assert seen["pubkey"] == PK
|
|
assert [b["id"] for b in out] == ["bk1"]
|
|
assert "payment_hash" not in out[0]
|
|
|
|
|
|
def test_http_rejects_account_without_pubkey(monkeypatch):
|
|
_patch_store(monkeypatch, [])
|
|
with pytest.raises(HTTPException) as e:
|
|
asyncio.run(
|
|
views_api.api_my_bookings(user=SimpleNamespace(id="u1", pubkey=None))
|
|
)
|
|
assert e.value.status_code == 409
|
|
|
|
|
|
def test_rpc_scopes_by_sender_and_requires_it(monkeypatch):
|
|
rows = [_booking(1)]
|
|
_patch_store(monkeypatch, rows)
|
|
req = transport_rpcs.NostrRpcRequest(
|
|
rpc_name="chatelet_booking_list_mine", request_id="r", body={}, sender_pubkey=PK
|
|
)
|
|
out = asyncio.run(transport_rpcs.handle_booking_list_mine(None, req))
|
|
assert [b["id"] for b in out] == ["bk1"]
|
|
|
|
anon = transport_rpcs.NostrRpcRequest(
|
|
rpc_name="chatelet_booking_list_mine", request_id="r", body={}
|
|
)
|
|
with pytest.raises(PermissionError):
|
|
asyncio.run(transport_rpcs.handle_booking_list_mine(None, anon))
|