-
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
@@ -48,6 +78,8 @@
+
+
![Ticket QR]()
@@ -76,6 +108,7 @@
.ticket-print-sheet {
display: flex !important;
+ flex-direction: column;
align-items: center;
justify-content: center;
width: 100vw;
@@ -87,6 +120,18 @@
print-color-adjust: exact;
}
+ .ticket-print-event {
+ text-align: center;
+ margin-bottom: 8px;
+ }
+
+ .ticket-print-type {
+ text-align: center;
+ margin-top: 0;
+ margin-bottom: 20px;
+ color: #666 !important;
+ }
+
.ticket-print-qr {
display: block;
width: 320px;
diff --git a/static/routes.json b/static/routes.json
index f94b832..f2b90ec 100644
--- a/static/routes.json
+++ b/static/routes.json
@@ -6,6 +6,13 @@
"component": "/events/static/js/index.js",
"i18n": "/events/static/js/i18n"
},
+ {
+ "path": "/events/basket/:id",
+ "name": "PageEventsBasket",
+ "template": "/events/static/js/basket.vue",
+ "component": "/events/static/js/basket.js",
+ "i18n": "/events/static/js/i18n"
+ },
{
"path": "/events/:id",
"name": "PageEventsDisplay",
diff --git a/tasks.py b/tasks.py
index 1d30dce..a26c426 100644
--- a/tasks.py
+++ b/tasks.py
@@ -1,12 +1,6 @@
import asyncio
-from lnbits.core.models import Payment
-from lnbits.tasks import register_invoice_listener
-from loguru import logger
-
-from .crud import get_ticket
from .models import Ticket
-from .services import send_ticket_notification_in_background, set_ticket_paid
payment_listeners: dict[str, list[asyncio.Queue[Ticket]]] = {}
@@ -22,28 +16,3 @@ def deregister_payment_listener(payment_hash, queue: asyncio.Queue[Ticket]) -> N
payment_listeners[payment_hash].remove(queue)
if not payment_listeners[payment_hash]:
del payment_listeners[payment_hash]
-
-
-async def wait_for_paid_invoices():
- invoice_queue = asyncio.Queue()
- register_invoice_listener(invoice_queue, "ext_events")
-
- while True:
- payment = await invoice_queue.get()
- await on_invoice_paid(payment)
-
-
-async def on_invoice_paid(payment: Payment) -> None:
- if not payment.extra or "events" != payment.extra.get("tag"):
- return
-
- ticket = await get_ticket(payment.payment_hash)
- if not ticket:
- logger.warning(f"Ticket for payment {payment.payment_hash} not found.")
- return
-
- ticket = await set_ticket_paid(ticket)
- send_ticket_notification_in_background(ticket)
- if payment_listeners.get(payment.payment_hash):
- for paid_ticket_queue in payment_listeners[payment.payment_hash]:
- paid_ticket_queue.put_nowait(ticket)
diff --git a/templates/event_public.html b/templates/event_public.html
new file mode 100644
index 0000000..12fd8f1
--- /dev/null
+++ b/templates/event_public.html
@@ -0,0 +1,15 @@
+{% extends "base.html" %}
+{% block head_scripts %}
+{{ super() }}
+{% if event %}
+
+
+{% if event.banner %}
+
+
+{% else %}
+
+{% endif %}
+
+{% endif %}
+{% endblock %}
\ No newline at end of file
diff --git a/tests/_inmem_crud.py b/tests/_inmem_crud.py
new file mode 100644
index 0000000..8e305f0
--- /dev/null
+++ b/tests/_inmem_crud.py
@@ -0,0 +1,147 @@
+# In-memory CRUD stub backing the real events/services.py during tests.
+from datetime import datetime, timezone
+
+from events.models import Ticket, TicketExtra, sold_counts
+
+_STORE = {"tickets": {}, "ticket_types": {}, "events": {}, "baskets": {}}
+
+
+def reset():
+ """Empty the in-memory store (used between tests)."""
+ for collection in _STORE.values():
+ collection.clear()
+
+
+def _now():
+ return datetime.now(timezone.utc)
+
+
+# --- tickets ---
+async def create_ticket(payment_hash, wallet, event, name, email, extra,
+ ticket_type_id=None, basket_id=None):
+ t = Ticket(id=payment_hash, wallet=wallet, event=event, name=name, email=email,
+ registered=False, paid=False, reg_timestamp=_now(), time=_now(),
+ ticket_type_id=ticket_type_id, basket_id=basket_id,
+ extra=TicketExtra(**extra) if extra else TicketExtra())
+ _STORE["tickets"][t.id] = t
+ return t
+
+
+async def update_ticket(ticket):
+ _STORE["tickets"][ticket.id] = ticket
+ return ticket
+
+
+async def get_ticket(payment_hash):
+ return _STORE["tickets"].get(payment_hash)
+
+
+async def get_tickets(wallet_ids):
+ ids = wallet_ids if isinstance(wallet_ids, list) else [wallet_ids]
+ return [t for t in _STORE["tickets"].values() if t.wallet in ids]
+
+
+async def get_event_tickets(event_id):
+ return [t for t in _STORE["tickets"].values() if t.event == event_id]
+
+
+async def get_basket_tickets(basket_id):
+ return [t for t in _STORE["tickets"].values() if t.basket_id == basket_id]
+
+
+async def delete_ticket(ticket_id):
+ _STORE["tickets"].pop(ticket_id, None)
+
+
+async def delete_event_tickets(event_id):
+ for k in [k for k in _STORE["tickets"] if _STORE["tickets"][k].event == event_id]:
+ _STORE["tickets"].pop(k, None)
+
+
+async def purge_unpaid_tickets(event_id):
+ for k in [k for k in _STORE["tickets"]
+ if _STORE["tickets"][k].event == event_id and not _STORE["tickets"][k].paid]:
+ _STORE["tickets"].pop(k, None)
+
+
+# --- ticket types ---
+async def get_ticket_type(tt_id):
+ return _STORE["ticket_types"].get(tt_id)
+
+
+async def get_all_ticket_types():
+ return list(_STORE["ticket_types"].values())
+
+
+async def get_ticket_types(event_id):
+ types = sorted(
+ (tt for tt in _STORE["ticket_types"].values() if tt.event_id == event_id),
+ key=lambda tt: tt.sort_order,
+ )
+ counts = sold_counts(list(_STORE["tickets"].values()))
+ out = []
+ for tt in types:
+ tt = tt.copy() # the real reader returns fresh rows
+ tt.sold = counts.get(tt.id, 0)
+ out.append(tt)
+ return out
+
+
+async def count_tickets_by_type(ticket_type_id):
+ tt = _STORE["ticket_types"].get(ticket_type_id)
+ if not tt:
+ return 0
+ tickets = [t for t in _STORE["tickets"].values() if t.event == tt.event_id]
+ return sold_counts(tickets).get(ticket_type_id, 0)
+
+
+async def create_ticket_type(ticket_type):
+ _STORE["ticket_types"][ticket_type.id] = ticket_type
+ return ticket_type
+
+
+async def update_ticket_type(ticket_type):
+ _STORE["ticket_types"][ticket_type.id] = ticket_type
+ return ticket_type
+
+
+async def delete_ticket_type(tt_id):
+ _STORE["ticket_types"].pop(tt_id, None)
+
+
+# --- events ---
+async def create_event(event):
+ _STORE["events"][event.id] = event
+ return event
+
+
+async def update_event(event):
+ _STORE["events"][event.id] = event
+ return event
+
+
+async def get_event(event_id):
+ return _STORE["events"].get(event_id)
+
+
+async def get_events(*a, **k):
+ return list(_STORE["events"].values())
+
+
+# --- baskets ---
+async def create_basket(basket):
+ _STORE["baskets"][basket.id] = basket
+ return basket
+
+
+async def update_basket(basket):
+ _STORE["baskets"][basket.id] = basket
+ return basket
+
+
+async def get_basket(basket_id):
+ return _STORE["baskets"].get(basket_id)
+
+
+async def get_event_baskets(event_id):
+ return [b for b in _STORE["baskets"].values() if b.event_id == event_id]
diff --git a/tests/_lnbits_stub.py b/tests/_lnbits_stub.py
new file mode 100644
index 0000000..2609eaa
--- /dev/null
+++ b/tests/_lnbits_stub.py
@@ -0,0 +1,71 @@
+"""Minimal stubs for lnbits modules so the extension imports standalone.
+
+LNbits is not a dependency of these tests; registering these fake modules lets
+``events.models`` and ``events.services`` import without a real LNbits install.
+"""
+import base64
+import secrets
+import sys
+import types
+
+lnbits = types.ModuleType("lnbits")
+lnbits_core = types.ModuleType("lnbits.core")
+lnbits_core_models = types.ModuleType("lnbits.core.models")
+lnbits_core_users = types.ModuleType("lnbits.core.models.users")
+lnbits_core_svc = types.ModuleType("lnbits.core.services")
+lnbits_core_notif = types.ModuleType("lnbits.core.services.notifications")
+lnbits_helpers = types.ModuleType("lnbits.helpers")
+lnbits_settings = types.ModuleType("lnbits.settings")
+lnbits_db = types.ModuleType("lnbits.db")
+
+sys.modules["lnbits"] = lnbits
+sys.modules["lnbits.core"] = lnbits_core
+sys.modules["lnbits.core.models"] = lnbits_core_models
+sys.modules["lnbits.core.models.users"] = lnbits_core_users
+sys.modules["lnbits.core.services"] = lnbits_core_svc
+sys.modules["lnbits.core.services.notifications"] = lnbits_core_notif
+sys.modules["lnbits.helpers"] = lnbits_helpers
+sys.modules["lnbits.settings"] = lnbits_settings
+sys.modules["lnbits.db"] = lnbits_db
+
+
+class UserNotifications:
+ def __init__(self, *a, **k):
+ pass
+
+
+lnbits_core_users.UserNotifications = UserNotifications
+
+
+async def send_user_notification(*a, **k):
+ return None
+
+
+lnbits_core_notif.send_user_notification = send_user_notification
+
+
+def is_valid_email_address(v):
+ return isinstance(v, str) and "@" in v
+
+
+def urlsafe_short_hash():
+ return base64.urlsafe_b64encode(secrets.token_bytes(6)).decode().rstrip("=")
+
+
+lnbits_helpers.is_valid_email_address = is_valid_email_address
+lnbits_helpers.urlsafe_short_hash = urlsafe_short_hash
+
+
+class _Settings:
+ host = "127.0.0.1"
+ port = 5000
+
+
+lnbits_settings.settings = _Settings()
+
+
+class FilterModel:
+ pass
+
+
+lnbits_db.FilterModel = FilterModel
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..89c4c73
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,71 @@
+"""Bootstrap so the extension's own modules can be imported and exercised
+without a running LNbits instance or database.
+
+``_lnbits_stub`` registers fake ``lnbits.*`` modules (a real LNbits install is
+not required), and an in-memory CRUD (``_inmem_crud``) is registered as
+``events.crud``.
+
+pytest imports this conftest as part of the ``events.tests`` package, which
+means the extension package (and its DB-backed ``events.crud``) may already be
+in ``sys.modules``. The data layer is therefore dropped and rebuilt after the
+stub is in place, so that ``events.services`` -- whose ``from .crud import ...``
+bindings resolve at import time -- talks to the in-memory store.
+"""
+import importlib.util
+import sys
+import types
+from pathlib import Path
+
+import pytest
+
+TESTS_DIR = Path(__file__).resolve().parent
+REPO_DIR = TESTS_DIR.parent
+
+for _p in (str(TESTS_DIR), str(REPO_DIR)):
+ if _p not in sys.path:
+ sys.path.insert(0, _p)
+
+
+def _load(name, filename):
+ """Load a helper module from this directory under an explicit name."""
+ spec = importlib.util.spec_from_file_location(name, TESTS_DIR / filename)
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+# If the ``events`` package is not importable as a package yet, register a
+# lightweight one so the heavy ``events/__init__`` (routers, DB) is not executed.
+if "events" not in sys.modules:
+ _pkg = types.ModuleType("events")
+ _pkg.__path__ = [str(REPO_DIR)]
+ _pkg.__package__ = "events"
+ sys.modules["events"] = _pkg
+
+# 1. fake lnbits.* modules, registered before any extension module is rebuilt.
+_lnbits_stub = _load("_lnbits_stub", "_lnbits_stub.py") # noqa: F841
+
+# 2. rebuild the data layer against the stub + the in-memory store. Only the
+# modules that bind crud are rebuilt: `events.models` is pure pydantic and
+# is kept as-is (re-executing it would trip pydantic's duplicate-validator
+# guard for the same qualified names).
+for _mod in ("events.services", "events.crud"):
+ sys.modules.pop(_mod, None)
+
+import events.models # noqa: E402,F401
+
+_inmem_crud = _load("events.crud", "_inmem_crud.py")
+# `from events import crud` resolves the package attribute, not sys.modules, so
+# rebind it explicitly (the original DB-backed crud is still the attribute).
+sys.modules["events"].crud = _inmem_crud
+
+import events.services # noqa: E402,F401
+
+
+@pytest.fixture(autouse=True)
+def _clean_store():
+ """Give every test an empty in-memory store."""
+ _inmem_crud.reset()
+ yield
+ _inmem_crud.reset()
diff --git a/tests/test_addon.py b/tests/test_addon.py
new file mode 100644
index 0000000..8e273d0
--- /dev/null
+++ b/tests/test_addon.py
@@ -0,0 +1,365 @@
+"""Add-on ticket entitlements: gated offer codes that GRANT rather than
+discount (the Bitflix screening-seat case).
+
+A grant rule authorises ADDING tickets of another type, only while a
+qualifying source ticket is present (in the basket, or presented by id), never
+beyond ``per_source x admits`` per source ticket, and never past the add-on
+type's counted cap.
+"""
+import asyncio
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from events import crud
+from events import models as M
+from events import services as S
+
+TODAY = datetime.now(timezone.utc).date().isoformat()
+
+WEEKEND = "WEEKEND"
+FAMILY = "FAMILY"
+ADDON = "ADDON"
+
+
+async def make_event(addon_cap=100):
+ rule = M.PromoCode(
+ code="",
+ source=M.PromoSource(require_ticket=True, ticket_type_ids=[WEEKEND, FAMILY]),
+ grants=M.PromoGrant(ticket_type_id=ADDON, per_source=1),
+ )
+ ev = M.Event(
+ id="EV1", wallet="w1", name="Bitfest", info="", canceled=False,
+ event_start_date="2026-01-01", event_end_date="2026-12-31",
+ currency="GBP", admin_email=None, allow_fiat=False, fiat_currency="GBP",
+ amount_tickets=100, price_per_ticket=149, time=datetime.now(timezone.utc),
+ sold=0, banner=None, extra=M.EventExtra(promo_codes=[rule]),
+ )
+ await crud.create_event(ev)
+ for tid, name, price, admits, cap in [
+ (WEEKEND, "Single - Weekend", 149, 1, 500),
+ (FAMILY, "Family - Weekend", 399, 4, 50),
+ (ADDON, "Bitflix - The Six Billion Dollar Man", 0, 1, addon_cap),
+ ]:
+ await crud.create_ticket_type(M.TicketType(
+ id=tid, event_id="EV1", name=name, price=price, max_tickets=cap,
+ available_from=TODAY, available_to="2026-12-31",
+ sort_order=0, extra={"admits": admits},
+ ))
+ return ev
+
+
+async def make_ticket(ticket_id, ticket_type_id, paid=True, event="EV1", extra=None):
+ t = M.Ticket(
+ id=ticket_id, wallet="w1", event=event, name="Holder", email="a@b.c",
+ registered=False, paid=paid, time=datetime.now(timezone.utc),
+ reg_timestamp=datetime.now(timezone.utc), ticket_type_id=ticket_type_id,
+ basket_id=None, extra=extra or M.TicketExtra(),
+ )
+ await crud.update_ticket(t)
+ return t
+
+
+def items(*pairs):
+ return [SimpleNamespace(ticket_type_id=tid, quantity=q) for tid, q in pairs]
+
+
+async def gate(ev, basket_items, presented=()):
+ """Run the real resolution + the add-on gate, as the checkout does."""
+ ticket_types = await S.get_ticket_types(ev.id)
+ reservation = await S._resolve_ticket_rules(
+ ev, list(presented), sum(i.quantity for i in basket_items)
+ )
+ await S._validate_addon_items(ev, basket_items, ticket_types, reservation)
+
+
+# --- model invariants -------------------------------------------------------
+
+def test_offer_title_round_trips():
+ """A code-less offer has no identifier; `title` names it in the admin."""
+ assert M.PromoCode(code="X").title == ""
+ assert M.PromoCode(code="X", title="Bitflix free seat").title == "Bitflix free seat"
+ # the case it exists for: a code-less rule (blank code needs a ticket gate)
+ codeless = M.PromoCode(
+ code="", source=M.PromoSource(require_ticket=True), title="Screening seat")
+ assert codeless.code == "" and codeless.title == "Screening seat"
+ # survives the legacy-axis migration path too
+ legacy = M.PromoCode(code="TICKETID", type="ticket", title="Old rule")
+ assert legacy.grants is None and legacy.title == "Old rule"
+
+
+def test_grant_implies_the_source_gate():
+ """A stored grant with no explicit source condition still loads, and is
+ read as requiring a ticket — otherwise the codeless rule is invalid and
+ the whole event fails to load."""
+ promo = M.PromoCode(code="", grants=M.PromoGrant(ticket_type_id=ADDON))
+ assert promo.grants is not None
+ assert promo.source.require_ticket is True
+
+
+def test_grant_keeps_and_ignores_a_discount():
+ """A grant must not DESTROY a discount the admin typed — only ignore it, so
+ flipping the effect back and forth in the UI is lossless."""
+ promo = M.PromoCode(
+ code="", discount_percent=50,
+ source=M.PromoSource(require_ticket=True),
+ grants=M.PromoGrant(ticket_type_id=ADDON, per_source=2),
+ )
+ assert promo.grants.per_source == 2
+ assert promo.discount_percent == 50, "the typed discount is retained"
+ assert promo.source.require_ticket is True, "a grant implies the source gate"
+
+
+def test_per_source_must_be_positive():
+ with pytest.raises(Exception):
+ M.PromoGrant(ticket_type_id=ADDON, per_source=0)
+
+
+def test_admits_defaults_and_survives_junk():
+ assert S.ticket_admits(None) == 1
+ assert S.ticket_admits(M.TicketType(
+ id="x", event_id="EV1", name="n", available_from=TODAY,
+ available_to="2026-12-31", extra={})) == 1
+ assert S.ticket_admits(M.TicketType(
+ id="x", event_id="EV1", name="n", available_from=TODAY,
+ available_to="2026-12-31", extra={"admits": "4"})) == 4
+ assert S.ticket_admits(M.TicketType(
+ id="x", event_id="EV1", name="n", available_from=TODAY,
+ available_to="2026-12-31", extra={"admits": "junk"})) == 1
+
+
+# --- the count itself ---------------------------------------------------------
+
+def test_sold_counts_is_committed_seats_only():
+ def mk(i, ttype, paid=True, refunded=False, deactivated=False):
+ return M.Ticket(
+ id=i, wallet="w", event="EV1", name="n", email="a@b.c", registered=False,
+ paid=paid, time=datetime.now(timezone.utc), ticket_type_id=ttype,
+ extra=M.TicketExtra(refunded=refunded, deactivated=deactivated))
+ tickets = [
+ mk("1", "A"), mk("2", "A"),
+ mk("3", "A", paid=False), # unpaid basket
+ mk("4", "A", refunded=True), # refunded frees the seat
+ mk("5", "A", deactivated=True), # an upgrade voids its source
+ mk("6", "B"),
+ mk("7", None), # no type -> not attributable
+ ]
+ assert M.sold_counts(tickets) == {"A": 2, "B": 1}
+
+
+def test_sold_is_derived_not_the_stored_column():
+ """The stored `sold` drifts (imported tickets skip the increment). Readers
+ must derive it from the rows, so a wrong column cannot leak a cap."""
+ async def run():
+ await make_event()
+ tt = await crud.get_ticket_type(ADDON)
+ tt.sold = 999
+ await crud.update_ticket_type(tt)
+ await make_ticket("S1", ADDON)
+ await make_ticket("S2", ADDON, paid=False)
+ types = {t.id: t for t in await crud.get_ticket_types("EV1")}
+ assert types[ADDON].sold == 1, "derived from rows, ignoring the stored 999"
+ asyncio.run(run())
+
+
+def test_addon_cap_counts_only_committed_seats():
+ async def run():
+ ev = await make_event(addon_cap=2)
+ await make_ticket("U1", ADDON, paid=False) # abandoned
+ await make_ticket("R1", ADDON, extra=M.TicketExtra(refunded=True))
+ await gate(ev, items((WEEKEND, 2), (ADDON, 2))) # 2 committed -> fits
+ await make_ticket("P1", ADDON) # now 1 committed
+ with pytest.raises(ValueError):
+ await gate(ev, items((WEEKEND, 2), (ADDON, 2))) # 1+2 > 2
+ await gate(ev, items((WEEKEND, 2), (ADDON, 1))) # only 1 left
+ asyncio.run(run())
+
+
+# --- the gate ---------------------------------------------------------------
+
+def test_addon_alone_is_refused():
+ async def run():
+ ev = await make_event()
+ with pytest.raises(ValueError) as exc:
+ await gate(ev, items((ADDON, 1)))
+ assert "qualifying ticket" in str(exc.value)
+ assert "Promo Code box" in str(exc.value), (
+ "the message must say WHERE to put the ticket ID"
+ )
+ asyncio.run(run())
+
+
+def test_weekend_in_the_basket_authorises_one():
+ async def run():
+ ev = await make_event()
+ await gate(ev, items((WEEKEND, 1), (ADDON, 1))) # allowed
+ with pytest.raises(ValueError):
+ await gate(ev, items((WEEKEND, 1), (ADDON, 2))) # one per person
+ await gate(ev, items((WEEKEND, 2), (ADDON, 2))) # two tickets, two seats
+ asyncio.run(run())
+
+
+def test_family_weekend_authorises_four():
+ async def run():
+ ev = await make_event()
+ await gate(ev, items((FAMILY, 1), (ADDON, 4))) # allowed
+ with pytest.raises(ValueError):
+ await gate(ev, items((FAMILY, 1), (ADDON, 5))) # one too many
+ asyncio.run(run())
+
+
+def test_presented_ticket_authorises_one():
+ async def run():
+ ev = await make_event()
+ await make_ticket("SRC1", WEEKEND)
+ await gate(ev, items((ADDON, 1)), presented=["SRC1"]) # allowed
+ with pytest.raises(ValueError):
+ await gate(ev, items((ADDON, 2)), presented=["SRC1"])
+ asyncio.run(run())
+
+
+def test_presented_family_ticket_authorises_four():
+ async def run():
+ ev = await make_event()
+ await make_ticket("SRC2", FAMILY)
+ await gate(ev, items((ADDON, 4)), presented=["SRC2"]) # allowed
+ with pytest.raises(ValueError):
+ await gate(ev, items((ADDON, 5)), presented=["SRC2"])
+ asyncio.run(run())
+
+
+def test_cross_event_source_keeps_its_admits():
+ """A source ticket belonging to ANOTHER event must still be worth its
+ admits rather than defaulting to one: the allowance looks its type up."""
+ async def run():
+ ev = await make_event()
+ other = M.Event(
+ id="EV2", wallet="w1", name="Freedom Fest", info="", canceled=False,
+ event_start_date="2026-01-01", event_end_date="2026-12-31",
+ currency="GBP", admin_email=None, allow_fiat=False, fiat_currency="GBP",
+ amount_tickets=10, price_per_ticket=119, time=datetime.now(timezone.utc),
+ sold=0, banner=None, extra=M.EventExtra(promo_codes=[]),
+ )
+ await crud.create_event(other)
+ await crud.create_ticket_type(M.TicketType(
+ id="FF_FAMILY", event_id="EV2", name="Family (Full Access)", price=119,
+ max_tickets=21, available_from=TODAY, available_to="2026-12-31",
+ sort_order=0, extra={"admits": 4},
+ ))
+ rule = ev.extra.promo_codes[0]
+ rule.source.event_ids = ["EV1", "EV2"]
+ rule.source.ticket_type_ids = [WEEKEND, FAMILY, "FF_FAMILY"]
+ await crud.update_event(ev)
+ await crud.update_ticket(M.Ticket(
+ id="FF1", wallet="w1", event="EV2", name="Holder", email="a@b.c",
+ registered=False, paid=True, time=datetime.now(timezone.utc),
+ reg_timestamp=datetime.now(timezone.utc), ticket_type_id="FF_FAMILY",
+ basket_id=None, extra=M.TicketExtra()))
+ await gate(ev, items((ADDON, 4)), presented=["FF1"]) # four allowed
+ with pytest.raises(ValueError):
+ await gate(ev, items((ADDON, 5)), presented=["FF1"]) # not five
+ asyncio.run(run())
+
+
+def test_unpaid_source_is_refused():
+ async def run():
+ ev = await make_event()
+ await make_ticket("SRC3", WEEKEND, paid=False)
+ with pytest.raises(ValueError) as exc:
+ await gate(ev, items((ADDON, 1)), presented=["SRC3"])
+ assert "qualifying ticket" in str(exc.value)
+ asyncio.run(run())
+
+
+def test_addon_is_never_its_own_source():
+ async def run():
+ ev = await make_event()
+ await make_ticket("A1", ADDON)
+ with pytest.raises(ValueError):
+ await gate(ev, items((ADDON, 2)), presented=["A1"])
+ asyncio.run(run())
+
+
+def test_counted_cap_is_enforced():
+ async def run():
+ ev = await make_event(addon_cap=2)
+ await make_ticket("B1", ADDON) # 1 already issued
+ await gate(ev, items((WEEKEND, 2), (ADDON, 1))) # allowance 2, 1 left
+ with pytest.raises(ValueError) as exc:
+ await gate(ev, items((WEEKEND, 2), (ADDON, 2))) # 1 left of a cap of 2
+ assert "left" in str(exc.value)
+ asyncio.run(run())
+
+
+def test_shared_cap_spans_events():
+ """Types sharing extra.cap_group consume ONE pool, so issuing seats on
+ another event's checkout eats this one's headroom."""
+ async def run():
+ ev = await make_event(addon_cap=3)
+ await crud.create_ticket_type(M.TicketType(
+ id="ADDON_FF", event_id="EV2", name="Bitflix (Freedom Fest)", price=0,
+ max_tickets=3, available_from=TODAY, available_to="2026-12-31",
+ sort_order=0, extra={"admits": 1, "cap_group": "bitflix"}))
+ tt = await crud.get_ticket_type(ADDON)
+ tt.extra = {**tt.extra, "cap_group": "bitflix"}
+ await crud.update_ticket_type(tt)
+ await make_ticket("F1", "ADDON_FF", event="EV2")
+ await make_ticket("F2", "ADDON_FF", event="EV2")
+ # two weekend tickets authorise two seats, so the CAP is what refuses
+ await gate(ev, items((WEEKEND, 2), (ADDON, 1))) # 1 seat left of 3
+ with pytest.raises(ValueError) as exc:
+ await gate(ev, items((WEEKEND, 2), (ADDON, 2))) # only 1 left
+ assert "left" in str(exc.value)
+ asyncio.run(run())
+
+
+def test_grant_rule_discounts_nothing():
+ """A grant rule must not leak into the discount engine — not even when it
+ carries a retained discount value (the lossless-UI case)."""
+ async def run():
+ ev = await make_event()
+ totals = await S.calculate_basket_total(
+ ev, [{"ticket_type_id": WEEKEND, "quantity": 1}], []
+ )
+ assert totals.total == 149 and totals.discount == 0, (
+ "no discount applied for a grant rule with no source presented"
+ )
+ # now make it carry a discount AND present an eligible source ticket
+ rule = ev.extra.promo_codes[0]
+ rule.discount_percent = 50
+ await crud.update_event(ev)
+ await make_ticket("SRC-D", WEEKEND)
+ totals2 = await S.calculate_basket_total(
+ ev, [{"ticket_type_id": WEEKEND, "quantity": 1}], ["SRC-D"]
+ )
+ assert totals2.discount == 0 and totals2.total == 149, (
+ "a grant rule's retained discount must stay inert"
+ )
+ asyncio.run(run())
+
+
+# --- end to end: the zero-total claim ---------------------------------------
+
+def test_zero_total_addon_basket_issues_and_activates():
+ async def run():
+ ev = await make_event()
+ await make_ticket("SRC9", WEEKEND)
+ S._send_basket_notifications_in_background = lambda *a, **k: None
+ data = M.CreateBasket(
+ name="Holder", email="a@b.c",
+ items=[M.BasketItem(ticket_type_id=ADDON, quantity=1)],
+ promo_codes=["SRC9"],
+ )
+ basket, tickets, totals, charge = await S.create_basket_with_charge(
+ ev, data, "key", "https://example.test"
+ )
+ assert totals.total == 0, "the add-on is free"
+ assert charge is None, "no payment is requested for a zero-total basket"
+ assert basket.paid is True, "zero-total basket activates immediately"
+ assert len(tickets) == 1 and tickets[0].ticket_type_id == ADDON
+ src = await crud.get_ticket("SRC9")
+ assert src.extra.ticket_id_voucher_redemptions[0]["status"] == "completed", (
+ "the claim is finalized against the source ticket"
+ )
+ asyncio.run(run())
diff --git a/tests/test_promo_migration.py b/tests/test_promo_migration.py
new file mode 100644
index 0000000..85f3f33
--- /dev/null
+++ b/tests/test_promo_migration.py
@@ -0,0 +1,190 @@
+"""m010_promo_source_axis regression test.
+
+Runs the real migration body against a throwaway SQLite database (the same
+engine LNbits uses) with the ``events.events`` schema prefix stripped, since
+that prefix only resolves inside LNbits' ATTACHed connection.
+
+Asserts the legacy ``type`` axis is fully converted, the "TICKETID" sentinel
+code is cleared, codes/counters survive, and a second run is a no-op.
+15 assertions across three tests.
+"""
+import asyncio
+import json
+import sqlite3
+from pathlib import Path
+
+TESTS_DIR = Path(__file__).resolve().parent
+MIGRATIONS = TESTS_DIR.parent / "migrations.py"
+
+
+class MappingRow(dict):
+ """Mimics SQLAlchemy's RowMapping, which is what ``db.fetchall`` returns when
+ no ``model=`` is passed: values are read BY COLUMN NAME, and iterating the
+ row yields column NAMES rather than values. A plain-tuple stub hides the
+ bug where a migration unpacks rows positionally and silently converts
+ nothing."""
+
+
+class DB:
+ """Minimal stand-in for the migration's ``db`` handle."""
+
+ def __init__(self, conn):
+ self.conn = conn
+
+ async def fetchall(self, query):
+ cur = self.conn.execute(query)
+ cols = [d[0] for d in cur.description]
+ return [MappingRow(zip(cols, values)) for values in cur.fetchall()]
+
+ async def execute(self, query, params=None):
+ self.conn.execute(query, params or {})
+ self.conn.commit()
+
+
+def load_migration():
+ src = MIGRATIONS.read_text()
+ body = src.split("async def m010_promo_source_axis(db):", 1)[1]
+ body = body.replace("events.events", "events")
+ # Deliberately do NOT seed the namespace with anything: the real
+ # migrations module has no module-level imports, so a migration that
+ # relies on a name it never imports must fail here too.
+ namespace = {}
+ exec("async def m010(db):" + body, namespace) # noqa: S102 - test harness
+ return namespace["m010"]
+
+
+EVENTS = {
+ "EV-A": {
+ "promo_codes": [
+ {"code": "SAVE21", "discount_percent": 21.0, "active": True,
+ "combinable": False, "max_uses": 50, "used_count": 7, "type": "code",
+ "redemptions_per_ticket": 1, "ticket_type_ids": ["tt1"],
+ "max_tickets": None},
+ {"code": "TICKETID", "discount_percent": 50.0, "active": True,
+ "combinable": False, "max_uses": None, "used_count": 0,
+ "type": "ticket", "redemptions_per_ticket": 1,
+ "ticket_type_ids": None, "max_tickets": 1},
+ ]
+ },
+ # A hand-written promo with no `type` at all (missing type meant "code").
+ "EV-B": {"promo_codes": [{"code": "WELOVEJAM", "discount_percent": 100.0,
+ "active": True, "combinable": True, "max_uses": 1,
+ "used_count": 1}]},
+ # Already on the final axis: must be left exactly as-is.
+ "EV-C": {"promo_codes": [{"code": "NEW", "source": {"require_ticket": True,
+ "move": True}}]},
+ # The INTERMEDIATE revision (`source.kind`) — what the live DB holds after
+ # the first deploy. Must convert to require_ticket.
+ "EV-E": {"promo_codes": [{"code": "UPGRADE50", "discount_percent": 10.0,
+ "active": True, "combinable": False,
+ "source": {"kind": "ticket", "event_ids": ["X"],
+ "ticket_type_ids": None,
+ "move": True, "credit": "paid"}},
+ {"code": "PLAIN", "discount_percent": 5.0,
+ "active": True, "combinable": True,
+ "source": {"kind": "code", "event_ids": None,
+ "ticket_type_ids": None,
+ "move": False, "credit": "none"}}]},
+ "EV-D": {},
+}
+
+
+def _seed_conn():
+ conn = sqlite3.connect(":memory:")
+ conn.execute("CREATE TABLE events (id TEXT PRIMARY KEY, extra TEXT)")
+ for eid, extra in EVENTS.items():
+ conn.execute("INSERT INTO events VALUES (?, ?)", (eid, json.dumps(extra)))
+ conn.commit()
+ return conn
+
+
+def _rows(conn):
+ return dict(conn.execute("SELECT id, extra FROM events").fetchall())
+
+
+def _run_migration(conn):
+ m010 = load_migration()
+ asyncio.run(m010(DB(conn)))
+
+
+def test_migration_removes_legacy_and_intermediate_axes():
+ conn = _seed_conn()
+ _run_migration(conn)
+ rows = _rows(conn)
+ all_promos = [p for ev in rows.values() for p in json.loads(ev).get("promo_codes", [])]
+
+ assert not any("type" in p for p in all_promos), (
+ "no promo is left on the legacy type axis"
+ )
+ assert all("source" in p for p in all_promos), "every promo now has a source"
+ assert not any("kind" in p["source"] for p in all_promos), (
+ "no promo is left on the intermediate 'kind' discriminator"
+ )
+
+
+def test_migration_converts_type_values_and_preserves_settings():
+ conn = _seed_conn()
+ _run_migration(conn)
+ rows = _rows(conn)
+ a = json.loads(rows["EV-A"])
+ b = json.loads(rows["EV-B"])
+ c = json.loads(rows["EV-C"])
+ e = json.loads(rows["EV-E"])
+
+ code_promo, ticket_promo = a["promo_codes"]
+ assert code_promo["source"]["require_ticket"] is False, (
+ f"type=code -> require_ticket False ({json.dumps(code_promo['source'])})"
+ )
+ assert ticket_promo["source"]["require_ticket"] is True, (
+ f"type=ticket -> require_ticket True ({json.dumps(ticket_promo['source'])})"
+ )
+ assert (ticket_promo["source"]["move"] is False
+ and ticket_promo["source"]["credit"] == "none"
+ and ticket_promo["source"]["event_ids"] is None), (
+ "legacy ticket rule gets move=False, credit='none', same-event scope"
+ )
+ assert ticket_promo["code"] == "", (
+ f"TICKETID sentinel code cleared ({ticket_promo['code']!r})"
+ )
+ assert (code_promo["max_uses"] == 50 and code_promo["used_count"] == 7
+ and code_promo["ticket_type_ids"] == ["tt1"]
+ and ticket_promo["max_tickets"] == 1), (
+ "promo settings preserved (max_uses/used_count/target scope)"
+ )
+ assert b["promo_codes"][0]["source"]["require_ticket"] is False, (
+ f"hand-written promo with no type -> require_ticket False "
+ f"({json.dumps(b['promo_codes'][0].get('source'))})"
+ )
+ assert c["promo_codes"][0]["source"] == {"require_ticket": True, "move": True}, (
+ f"promo already on the final axis is untouched "
+ f"({json.dumps(c['promo_codes'][0]['source'])})"
+ )
+
+ upg, _plain = e["promo_codes"]
+ assert upg["source"]["require_ticket"] is True, (
+ f"intermediate kind='ticket' -> require_ticket True ({json.dumps(upg['source'])})"
+ )
+ assert (upg["source"]["move"] is True
+ and upg["source"]["credit"] == "paid"
+ and upg["source"]["event_ids"] == ["X"]
+ and upg["code"] == "UPGRADE50"), (
+ f"intermediate conversion keeps the rule's ticket settings ({json.dumps(upg)})"
+ )
+ assert e["promo_codes"][1]["source"]["require_ticket"] is False, (
+ f"intermediate kind='code' -> require_ticket False "
+ f"({json.dumps(e['promo_codes'][1]['source'])})"
+ )
+
+
+def test_migration_leaves_empty_event_and_is_idempotent():
+ conn = _seed_conn()
+ _run_migration(conn)
+ rows = _rows(conn)
+ assert json.loads(rows["EV-D"]) == {}, "event with no promos is left alone"
+
+ snapshot = json.dumps({k: json.loads(v) for k, v in rows.items()}, sort_keys=True)
+ _run_migration(conn)
+ again = json.dumps(
+ {k: json.loads(v) for k, v in _rows(conn).items()}, sort_keys=True
+ )
+ assert again == snapshot, "re-running the migration is a no-op"
diff --git a/tests/test_promo_scope.py b/tests/test_promo_scope.py
new file mode 100644
index 0000000..9ad7566
--- /dev/null
+++ b/tests/test_promo_scope.py
@@ -0,0 +1,156 @@
+"""PromoCode.ticket_type_ids scoping, ported from the standalone
+``test_scope.py`` suite: 11 assertions (T1-T11).
+"""
+import asyncio
+from datetime import datetime, timezone
+
+from events import crud
+from events import models as M
+from events import services as S
+
+TODAY = datetime.now(timezone.utc).date().isoformat()
+# Ticket types mirroring BitFest 2026 prices
+TB = {"TT_A": 149.0, "TT_B": 99.0, "TT_C": 35.0}
+
+
+def approx(a, b, tol=0.011):
+ return abs(a - b) <= tol
+
+
+def one(tid, qty=1):
+ return [{"ticket_type_id": tid, "quantity": qty}]
+
+
+def make_event(extra_static, ev_id="EVX"):
+ return M.Event(
+ id=ev_id, wallet="w1", name="ScopeTest", info="", canceled=False,
+ event_start_date="2026-01-01", event_end_date="2026-12-31",
+ currency="GBP", admin_email=None, allow_fiat=False, fiat_currency="GBP",
+ amount_tickets=1000, price_per_ticket=100, time=datetime.now(timezone.utc),
+ sold=0, banner=None,
+ extra=M.EventExtra(promo_codes=list(extra_static)))
+
+
+async def seed(ev):
+ await crud.create_event(ev)
+ for tid, price in TB.items():
+ await crud.create_ticket_type(
+ M.TicketType(id=tid, event_id=ev.id, name=tid, price=price,
+ available_from=TODAY, available_to="2026-12-31"))
+ return ev
+
+
+async def calc(ev, codes, items):
+ return await S.calculate_basket_total(ev, items, codes)
+
+
+def _promos():
+ return [
+ # fixed 50, scoped to TT_A, non-combinable, cap 10
+ M.PromoCode(code="LATERBIRD", discount_fixed=50, ticket_type_ids=["TT_A"],
+ combinable=False, max_uses=10),
+ M.PromoCode(code="FIX50C", discount_fixed=50, ticket_type_ids=["TT_A"],
+ combinable=True),
+ M.PromoCode(code="BASE21", discount_percent=21, ticket_type_ids=["TT_A"],
+ combinable=True),
+ M.PromoCode(code="CAP50", discount_fixed=50, ticket_type_ids=["TT_C"],
+ combinable=True),
+ M.PromoCode(code="ALL10", discount_percent=10, combinable=True),
+ M.PromoCode(code="ALLFIX20", discount_fixed=20, combinable=True),
+ ]
+
+
+async def _seeded_event():
+ return await seed(make_event(_promos()))
+
+
+def test_scoped_fixed_discount():
+ async def run():
+ ev = await _seeded_event()
+ r = await calc(ev, ["LATERBIRD"], one("TT_A"))
+ assert approx(r.discount, 50) and approx(r.total, 99), (
+ f"T1 scoped code discounts its base by 50 (disc={r.discount} total={r.total})"
+ )
+ r = await calc(ev, ["LATERBIRD"], one("TT_B"))
+ assert approx(r.discount, 0) and approx(r.total, 99), (
+ f"T2 out-of-scope basket gives no discount (disc={r.discount})"
+ )
+ r = await calc(ev, ["LATERBIRD"], [{"ticket_type_id": "TT_A", "quantity": 1},
+ {"ticket_type_id": "TT_B", "quantity": 1}])
+ assert approx(r.discount, 50) and approx(r.total, 198), (
+ f"T3 mixed basket: 50 off, rest full price (disc={r.discount} total={r.total})"
+ )
+ r = await calc(ev, ["LATERBIRD"], one("TT_A", qty=2))
+ assert approx(r.discount, 50) and approx(r.total, 248.0), (
+ f"T4 qty=2 scoped base 298 -> 50 off (total={r.total})"
+ )
+
+ asyncio.run(run())
+
+
+def test_fixed_capped_at_scoped_base():
+ async def run():
+ ev = await _seeded_event()
+ r = await calc(ev, ["CAP50"], one("TT_C"))
+ assert approx(r.discount, 35) and approx(r.total, 0), (
+ f"T5 fixed discount capped at scoped base (disc={r.discount} total={r.total})"
+ )
+
+ asyncio.run(run())
+
+
+def test_combinable_ordering():
+ # ORDERING: scoped base first, then basket-wide on what remains.
+ # FIX50C: 149 -> 99 remaining; ALL10: 10% of 99 = 9.9 -> 59.9 total discount.
+ # A fixed+percent mix is required to actually prove the ordering.
+ async def run():
+ ev = await _seeded_event()
+ r = await calc(ev, ["FIX50C", "ALL10"], one("TT_A"))
+ assert approx(r.discount, 59.9) and approx(r.total, 89.1), (
+ f"T6 combinable: base first then basket-wide (disc={r.discount} total={r.total})"
+ )
+ r = await calc(ev, ["LATERBIRD", "ALL10"], one("TT_A"))
+ assert len(r.discounts_applied) == 1, (
+ "T7 non-combinable scoped blocks basket-wide"
+ )
+
+ asyncio.run(run())
+
+
+def test_scoped_absent_type_is_noop():
+ async def run():
+ ev = await _seeded_event()
+ r = await calc(ev, ["LATERBIRD", "CAP50"], one("TT_A"))
+ assert approx(r.discount, 50), (
+ f"T8 scoped code on absent type is a no-op (disc={r.discount})"
+ )
+
+ asyncio.run(run())
+
+
+def test_basket_wide_percent_regression():
+ async def run():
+ ev = await _seeded_event()
+ r = await calc(ev, ["ALL10"], [{"ticket_type_id": "TT_A", "quantity": 1},
+ {"ticket_type_id": "TT_B", "quantity": 1}])
+ assert approx(r.discount, 24.8) and approx(r.total, 223.2), (
+ f"T9 basket-wide percent unchanged (disc={r.discount})"
+ )
+ r = await calc(ev, ["BASE21", "ALLFIX20"], one("TT_A"))
+ # BASE21: 149*0.21 = 31.29 -> 117.71 ; ALLFIX20: 20 -> 51.29
+ assert approx(r.discount, 51.29), (
+ f"T10 scoped percent then basket fixed (disc={r.discount})"
+ )
+
+ asyncio.run(run())
+
+
+def test_scoped_code_respects_max_uses():
+ async def run():
+ ev2 = await seed(make_event(
+ [M.PromoCode(code="MAXED", discount_fixed=50, ticket_type_ids=["TT_A"],
+ combinable=False, max_uses=1, used_count=1)], ev_id="EVY"))
+ r = await calc(ev2, ["MAXED"], one("TT_A"))
+ assert approx(r.discount, 0), f"T11 scoped code respects max_uses (disc={r.discount})"
+
+ asyncio.run(run())
diff --git a/tests/test_promo_upgrade.py b/tests/test_promo_upgrade.py
new file mode 100644
index 0000000..3e25d12
--- /dev/null
+++ b/tests/test_promo_upgrade.py
@@ -0,0 +1,582 @@
+"""Upgrade / source-axis suite (Option B: PromoCode.source), ported from the
+standalone ``test_upgrade.py`` suite: 37 assertions (U1-U21, V1-V9, W1-W8).
+
+Runs against the real ``events.services`` + ``events.models`` with the stubbed
+lnbits modules and in-memory CRUD provided by ``tests/conftest.py``.
+"""
+import asyncio
+from datetime import datetime, timezone
+
+import pytest
+
+from events import crud
+from events import models as M
+from events import services as S
+
+# Keep activation free of background email tasks.
+S._send_basket_notifications_in_background = lambda *a, **k: None
+
+TODAY = datetime.now(timezone.utc).date().isoformat()
+
+
+def approx(a, b, tol=0.011):
+ return abs(a - b) <= tol
+
+
+def mk_event(eid, currency="GBP", promos=None):
+ return M.Event(
+ id=eid, wallet="w1", name=eid, info="", canceled=False,
+ event_start_date="2026-01-01", event_end_date="2026-12-31", currency=currency,
+ admin_email=None, allow_fiat=False, fiat_currency="GBP", amount_tickets=1000,
+ price_per_ticket=100, time=datetime.now(timezone.utc), sold=0, banner=None,
+ extra=M.EventExtra(promo_codes=list(promos or [])))
+
+
+def mk_type(tid, eid, price):
+ return M.TicketType(id=tid, event_id=eid, name=tid, price=price,
+ available_from=TODAY, available_to="2026-12-31")
+
+
+def mk_ticket(tid, eid, type_id, paid=True, paid_value=None, registered=False,
+ deactivated=False, refunded=False, upgraded_to=None, basket_id=None):
+ return M.Ticket(
+ id=tid, wallet="w1", event=eid, name="Holder", email="a@b.c",
+ registered=registered, paid=paid, time=datetime.now(timezone.utc),
+ reg_timestamp=datetime.now(timezone.utc), ticket_type_id=type_id,
+ basket_id=basket_id,
+ extra=M.TicketExtra(paid_value=paid_value, deactivated=deactivated,
+ refunded=refunded, upgraded_to=upgraded_to))
+
+
+def one(tid, qty=1):
+ return [{"ticket_type_id": tid, "quantity": qty}]
+
+
+def two(tid, qty=2):
+ """A basket bigger than the single-ticket rules, for the shape gate."""
+ return [{"ticket_type_id": tid, "quantity": qty}]
+
+
+def upgrade(**over):
+ """A move upgrade: source credit + optional extra, scoped to TT_T1."""
+ base = dict(
+ source=M.PromoSource(require_ticket=True, event_ids=["EVS"], move=True, credit="paid"),
+ discount_percent=None, discount_fixed=None, combinable=False,
+ ticket_type_ids=["TT_T1"], redemptions_per_ticket=1,
+ )
+ base.update(over)
+ return M.PromoCode(**base)
+
+
+def friend_rule():
+ return M.PromoCode(source=M.PromoSource(require_ticket=True, event_ids=["EVS"]),
+ discount_percent=50, combinable=False, max_tickets=3)
+
+
+async def calc(ev, codes, items):
+ return await S.calculate_basket_total(ev, items, codes)
+
+
+async def _setup():
+ """Seed the source/target/other/currency events and SRC1, return (src, tgt)."""
+ src = mk_event("EVS", "GBP") # source event
+ tgt = mk_event("EVT", "GBP") # target event (rules live here)
+ other = mk_event("EVX", "GBP")
+ sat = mk_event("EVC", "sat") # different currency
+ for ev in (src, tgt, other, sat):
+ await crud.create_event(ev)
+ await crud.create_ticket_type(mk_type("TT_S", "EVS", 50))
+ await crud.create_ticket_type(mk_type("TT_T1", "EVT", 200))
+ await crud.create_ticket_type(mk_type("TT_T2", "EVT", 60))
+ await crud.create_ticket_type(mk_type("TT_C", "EVC", 80))
+ await crud.update_ticket(mk_ticket("SRC1", "EVS", "TT_S", paid_value=50))
+ return src, tgt
+
+
+# ---------------------------------------------------------------- U1-U2
+def test_upgrade_credit_only_and_extra_incentive():
+ async def run():
+ src, tgt = await _setup()
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T1"))
+ assert approx(r.discount, 50) and approx(r.total, 150), (
+ f"U1 upgrade pays the difference (200-50) ({r.discount}/{r.total})"
+ )
+ tgt.extra.promo_codes = [upgrade(discount_fixed=20)]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T1"))
+ assert approx(r.discount, 70) and approx(r.total, 130), (
+ f"U2 credit + extra fixed incentive ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U3
+def test_upgrade_credit_capped_at_target_price():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_BIG", "EVS", "TT_S", paid_value=500))
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_BIG"], one("TT_T1"))
+ assert approx(r.discount, 200) and approx(r.total, 0), (
+ f"U3 credit capped at target price (free, never negative) ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U4
+def test_upgrade_discounts_only_its_target_type():
+ async def run():
+ src, tgt = await _setup()
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T2"))
+ assert approx(r.discount, 0) and approx(r.total, 60), (
+ f"U4 upgrade discounts only its target type ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U5
+def test_unnamed_source_event_is_same_event_only():
+ async def run():
+ src, tgt = await _setup()
+ tgt.extra.promo_codes = [M.PromoCode(
+ source=M.PromoSource(require_ticket=True, move=True, credit="paid"),
+ combinable=False, ticket_type_ids=["TT_T1"])]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T1"))
+ assert approx(r.discount, 0), (
+ f"U5 unnamed source event is not cross-event eligible ({r.discount})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U6
+def test_cross_currency_source_is_rejected():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_SAT", "EVC", "TT_C", paid_value=80))
+ tgt.extra.promo_codes = [upgrade(source=M.PromoSource(
+ require_ticket=True, event_ids=["EVC"], move=True, credit="paid"))]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_SAT"], one("TT_T1"))
+ assert approx(r.discount, 0), f"U6 cross-currency source is rejected ({r.discount})"
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U7
+def test_consumed_source_cannot_be_reused():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_USED", "EVS", "TT_S", paid_value=50,
+ upgraded_to="SOMETHING"))
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_USED"], one("TT_T1"))
+ assert approx(r.discount, 0), f"U7 consumed source cannot be reused ({r.discount})"
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U8
+def test_deactivated_or_refunded_sources_rejected():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_DEAD", "EVS", "TT_S", paid_value=50,
+ deactivated=True))
+ await crud.update_ticket(mk_ticket("SRC_REF", "EVS", "TT_S", paid_value=50,
+ refunded=True))
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_DEAD"], one("TT_T1"))
+ r2 = await calc(tgt, ["SRC_REF"], one("TT_T1"))
+ assert approx(r.discount, 0) and approx(r2.discount, 0), (
+ f"U8 deactivated/refunded sources rejected ({r.discount}/{r2.discount})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U9
+def test_move_upgrade_refuses_multi_ticket_basket():
+ async def run():
+ src, tgt = await _setup()
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T1", qty=2))
+ assert approx(r.discount, 0), (
+ f"U9 move upgrade refuses a multi-ticket basket ({r.discount})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U10
+def test_upgrade_and_friend_rules_coexist():
+ async def run():
+ src, tgt = await _setup()
+ friend = friend_rule()
+ tgt.extra.promo_codes = [upgrade(), friend]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC1"], one("TT_T2", qty=2))
+ assert approx(r.discount, 60) and approx(r.total, 60), (
+ f"U10 multiple ticket-source rules coexist ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U11
+def test_chained_credit_uses_paid_value():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_CHAIN", "EVS", "TT_S", paid_value=150))
+ tgt.extra.promo_codes = [upgrade()]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_CHAIN"], one("TT_T1"))
+ assert approx(r.discount, 150) and approx(r.total, 50), (
+ f"U11 chained credit uses paid_value not face value ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U12
+def test_move_consumption_voids_source_and_releases_seat():
+ async def run():
+ src, tgt = await _setup()
+ friend = friend_rule()
+ src.sold = 1
+ await crud.update_event(src)
+ tt_s = await crud.get_ticket_type("TT_S")
+ tt_s.sold = 1
+ await crud.update_ticket_type(tt_s)
+ tgt.extra.promo_codes = [upgrade(), friend]
+ await crud.update_event(tgt)
+ match = await S._match_ticket_rule(tgt, "SRC1", 1)
+ await S._consume_upgraded_source(match[0], match[1], "NEW_TKT")
+ consumed = await crud.get_ticket("SRC1")
+ src_after = await crud.get_event("EVS")
+ tt_after = await crud.get_ticket_type("TT_S")
+ assert (consumed.extra.deactivated and consumed.extra.upgraded_to == "NEW_TKT"
+ and src_after.sold == 0 and tt_after.sold == 0), (
+ f"U12 move voids the source and releases the seat "
+ f"(deact={consumed.extra.deactivated} sold={src_after.sold}/{tt_after.sold})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U13
+def test_registered_source_keeps_its_seat():
+ async def run():
+ src, tgt = await _setup()
+ friend = friend_rule()
+ src.sold = 1
+ await crud.update_event(src)
+ tt_s = await crud.get_ticket_type("TT_S")
+ tt_s.sold = 1
+ await crud.update_ticket_type(tt_s)
+ tgt.extra.promo_codes = [upgrade(), friend]
+ await crud.update_event(tgt)
+ await crud.update_ticket(mk_ticket("SRC_REG", "EVS", "TT_S", paid_value=50,
+ registered=True))
+ match = await S._match_ticket_rule(tgt, "SRC_REG", 1)
+ await S._consume_upgraded_source(match[0], match[1], "NEW2")
+ src_after = await crud.get_event("EVS")
+ assert src_after.sold == 1, (
+ f"U13 attended/registered source keeps its seat (sold={src_after.sold})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U14
+def test_same_event_source_with_unnamed_scope():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_SAME", "EVT", "TT_T2", paid_value=60))
+ tgt.extra.promo_codes = [M.PromoCode(source=M.PromoSource(require_ticket=True),
+ discount_fixed=10, combinable=False)]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_SAME"], one("TT_T2"))
+ assert approx(r.discount, 10) and approx(r.total, 50), (
+ f"U14 same-event source works with an unnamed scope ({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U15
+def test_combinable_ticket_source_stacks_with_literal_code():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_STACK", "EVS", "TT_S", paid_value=50))
+ combo = M.PromoCode(source=M.PromoSource(require_ticket=True, event_ids=["EVS"]),
+ discount_fixed=50, combinable=True)
+ literal = M.PromoCode(code="EXTRA10", discount_percent=10, combinable=True)
+ tgt.extra.promo_codes = [combo, literal]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["SRC_STACK", "EXTRA10"], one("TT_T1"))
+ assert approx(r.discount, 65) and approx(r.total, 135), (
+ f"U15 combinable ticket-source promo stacks with a literal code "
+ f"({r.discount}/{r.total})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U16-U17
+def test_ticket_redemption_reservation_and_shape_gate():
+ # Regression for the ffcc559 gate bug: multi-ticket redemptions must still
+ # be RESERVED, and max_tickets must refuse an oversized basket.
+ async def run():
+ src, tgt = await _setup()
+ friend = friend_rule()
+ await crud.update_ticket(mk_ticket("SRC_MULTI", "EVS", "TT_S", paid_value=50))
+ await crud.update_ticket(mk_ticket("SRC_ONE", "EVS", "TT_S", paid_value=50))
+ tgt.extra.promo_codes = [friend]
+ await crud.update_event(tgt)
+ await S._reserve_ticket_voucher(tgt, "SRC_MULTI", "BK_M", item_count=2)
+ reserved = await crud.get_ticket("SRC_MULTI")
+ assert len(reserved.extra.ticket_id_voucher_redemptions) == 1, (
+ f"U16 multi-ticket redemption is reserved "
+ f"({reserved.extra.ticket_id_voucher_redemptions})"
+ )
+ tgt.extra.promo_codes = [M.PromoCode(
+ source=M.PromoSource(require_ticket=True, event_ids=["EVS"]),
+ discount_percent=50, combinable=False, max_tickets=1)]
+ await crud.update_event(tgt)
+ with pytest.raises(ValueError):
+ await S._reserve_ticket_voucher(tgt, "SRC_ONE", "BK_1", item_count=2) # U17 max_tickets=1 refuses a 2-ticket basket
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U18
+def test_activation_stamps_paid_value_from_basket_total():
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_ACT", "EVS", "TT_S"))
+ act_basket = M.Basket(id="BK_ACT", event_id="EVS", wallet="w1", email="", name="",
+ promo_codes=[], paid=False, total=42.0)
+ await crud.create_basket(act_basket)
+ act_ticket = await crud.get_ticket("SRC_ACT")
+ act_ticket.basket_id = "BK_ACT"
+ await crud.update_ticket(act_ticket)
+ await S._activate_basket_tickets(act_basket, [act_ticket], src)
+ refreshed = await crud.get_ticket("SRC_ACT")
+ assert approx(refreshed.extra.paid_value, 42.0), (
+ f"U18 activation stamps paid_value from the basket total "
+ f"({refreshed.extra.paid_value})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- U19-U21
+def test_backfill_paid_values():
+ async def run():
+ src, tgt = await _setup()
+ await crud.create_basket(M.Basket(id="BK_BF", event_id="EVS", wallet="w1",
+ email="", name="", promo_codes=[], paid=True,
+ total=40.0))
+ await crud.update_ticket(mk_ticket("SRC_B1", "EVS", "TT_S", basket_id="BK_BF"))
+ await crud.update_ticket(mk_ticket("SRC_B2", "EVS", "TT_S", basket_id="BK_BF"))
+ await crud.update_ticket(mk_ticket("SRC_NB", "EVS", "TT_S"))
+ summary = await S.backfill_paid_values("w1")
+ b1 = await crud.get_ticket("SRC_B1")
+ nb = await crud.get_ticket("SRC_NB")
+ before = (await crud.get_ticket("SRC1")).extra.paid_value
+ assert approx(b1.extra.paid_value, 20.0), (
+ f"U19 backfill splits a basket total across its tickets ({b1.extra.paid_value})"
+ )
+ assert approx(nb.extra.paid_value, 50.0), (
+ f"U20 backfill falls back to the ticket type price ({nb.extra.paid_value})"
+ )
+ assert approx(before, 50.0) and summary["filled"] >= 3, (
+ f"U21 backfill is idempotent (leaves existing values alone) "
+ f"({before}/{summary})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- V1-V9
+def test_offer_codes():
+ # Offer codes additionally require a ticket id, which makes several
+ # ticket-scoped offers identifiable to the buyer.
+ async def run():
+ src, tgt = await _setup()
+ await crud.update_ticket(mk_ticket("SRC_V1", "EVS", "TT_S", paid_value=50))
+ await crud.update_ticket(mk_ticket("SRC_V2", "EVS", "TT_S", paid_value=50))
+
+ named = M.PromoCode(
+ code="UPGRADE50",
+ source=M.PromoSource(require_ticket=True, event_ids=["EVS"], move=True,
+ credit="paid"),
+ combinable=False, ticket_type_ids=["TT_T1"])
+ tgt.extra.promo_codes = [named]
+ await crud.update_event(tgt)
+
+ # V1 the offer code alone is recognised and reports what it still needs
+ r = await calc(tgt, ["UPGRADE50"], one("TT_T1"))
+ assert (approx(r.discount, 0) and len(r.pending) == 1
+ and r.pending[0].code == "UPGRADE50"
+ and r.pending[0].requires == "ticket_id"), (
+ "V1 offer code alone is pending (drives the auto-prompt) "
+ f"(discount={r.discount} pending={[(p.code, p.requires) for p in r.pending]})"
+ )
+
+ # V2 code + ticket applies, and the applied code is the OFFER CODE
+ r = await calc(tgt, ["UPGRADE50", "SRC_V1"], one("TT_T1"))
+ assert (approx(r.discount, 50) and approx(r.total, 150)
+ and [d.code for d in r.discounts_applied] == ["UPGRADE50"]), (
+ f"V2 offer code + ticket applies the credit ({r.discount}/{r.total} "
+ f"{[d.code for d in r.discounts_applied]})"
+ )
+
+ # V3 a ticket alone does not activate a coded offer
+ r = await calc(tgt, ["SRC_V2"], one("TT_T1"))
+ assert (approx(r.discount, 0)
+ and [x.reason for x in r.rejections] == ["ticket_needs_offer_code"]), (
+ "V3 a ticket alone cannot use a coded offer "
+ f"({r.discount} {[(x.code, x.reason) for x in r.rejections]})"
+ )
+
+ # V4 an unknown token is explained, exactly once
+ r = await calc(tgt, ["UPGRADE50", "NOTATOKEN"], one("TT_T1"))
+ assert (approx(r.discount, 0)
+ and [x.reason for x in r.rejections] == ["ticket_not_found"]), (
+ "V4 an unknown token gets exactly one rejection reason "
+ f"({[(x.code, x.reason) for x in r.rejections]})"
+ )
+
+ # V5 two named offers on one event, distinguished by the code typed
+ await crud.update_ticket(mk_ticket("SRC_V5", "EVS", "TT_S", paid_value=50))
+ friend2 = M.PromoCode(
+ code="BRINGAFRIEND",
+ source=M.PromoSource(require_ticket=True, event_ids=["EVS"]),
+ discount_percent=50, combinable=False, max_tickets=1)
+ tgt.extra.promo_codes = [named, friend2]
+ await crud.update_event(tgt)
+ r_up = await calc(tgt, ["UPGRADE50", "SRC_V5"], one("TT_T1"))
+ r_fr = await calc(tgt, ["BRINGAFRIEND", "SRC_V5"], one("TT_T1"))
+ assert (approx(r_up.discount, 50)
+ and [d.code for d in r_up.discounts_applied] == ["UPGRADE50"]
+ and approx(r_fr.discount, 100)
+ and [d.code for d in r_fr.discounts_applied] == ["BRINGAFRIEND"]), (
+ f"V5 multiple named offers are selected by code "
+ f"(up={r_up.discount}{[d.code for d in r_up.discounts_applied]} "
+ f"fr={r_fr.discount}{[d.code for d in r_fr.discounts_applied]})"
+ )
+
+ # V6 an exhausted SOURCE TICKET is refused with a specific reason
+ await crud.update_ticket(mk_ticket("SRC_V6", "EVS", "TT_S", paid_value=50))
+ await S._reserve_ticket_voucher(tgt, "SRC_V6", "BK_V6", item_count=1)
+ r = await calc(tgt, ["UPGRADE50", "SRC_V6"], one("TT_T1"))
+ assert (approx(r.discount, 0)
+ and [x.reason for x in r.rejections] == ["ticket_redemptions_exhausted"]), (
+ "V6 an exhausted source ticket is refused with a reason "
+ f"({[(x.code, x.reason) for x in r.rejections]})"
+ )
+
+ # V8 a valid ticket that the BASKET SHAPE doesn't fit says so precisely
+ r = await calc(tgt, ["BRINGAFRIEND", "SRC_V5"], two("TT_T1"))
+ assert (approx(r.discount, 0)
+ and [x.reason for x in r.rejections] == ["ticket_count_not_eligible"]), (
+ "V8 basket-shape refusal gets its own reason, not a generic one "
+ f"({[(x.code, x.reason) for x in r.rejections]})"
+ )
+
+ # V9 max_uses is now enforceable on a ticket rule (the code is what counts)
+ await crud.update_ticket(mk_ticket("SRC_V7", "EVS", "TT_S", paid_value=50))
+ limited = M.PromoCode(
+ code="LIMITED",
+ source=M.PromoSource(require_ticket=True, event_ids=["EVS"]),
+ discount_fixed=10, combinable=False, max_uses=1, used_count=0)
+ tgt.extra.promo_codes = [limited]
+ await crud.update_event(tgt)
+ r = await calc(tgt, ["LIMITED", "SRC_V7"], one("TT_T2"))
+ await S._increment_promo_uses(tgt, r.discounts_applied)
+ refreshed = await crud.get_event("EVT")
+ limited_after = [p for p in refreshed.extra.promo_codes if p.code == "LIMITED"][0]
+ r2 = await calc(refreshed, ["LIMITED", "SRC_V7"], one("TT_T2"))
+ assert (approx(r.discount, 10) and limited_after.used_count == 1
+ and approx(r2.discount, 0)
+ and [x.reason for x in r2.rejections] == ["promo_uses_exhausted"]), (
+ f"V9 max_uses counts and caps a ticket rule (first={r.discount} "
+ f"used={limited_after.used_count} second={r2.discount} "
+ f"{[(x.code, x.reason) for x in r2.rejections]})"
+ )
+
+ asyncio.run(run())
+
+
+# ---------------------------------------------------------------- W1-W8
+def _pc(**kw):
+ return M.PromoCode(**kw)
+
+
+def _rejects(**kw):
+ try:
+ _pc(**kw)
+ return False
+ except Exception:
+ return True
+
+
+def test_model_invariants():
+ assert (_rejects(code="", discount_percent=10)
+ and _pc(code="HELLO", discount_percent=10).code == "HELLO"), (
+ "W1 a code is required when no ticket is involved"
+ )
+
+ assert _pc(code="", source=M.PromoSource(require_ticket=True)).code == "", (
+ "W2 a blank code is allowed when a ticket is required"
+ )
+
+ legacy = _pc(code="TICKETID", type="ticket", discount_percent=10)
+ assert legacy.source.require_ticket is True and legacy.code == "", (
+ f"W3 legacy type='ticket' maps onto require_ticket, sentinel cleared "
+ f"({legacy.source.require_ticket}/{legacy.code!r})"
+ )
+
+ interim = _pc(code="UPGRADE50", source={"kind": "ticket", "move": True})
+ assert (interim.source.require_ticket is True and interim.source.move is True
+ and interim.code == "UPGRADE50"), (
+ f"W4 interim source.kind maps onto require_ticket ({interim.source}/{interim.code!r})"
+ )
+
+ coerced = _pc(code="PLAIN", source={"require_ticket": False, "move": True,
+ "credit": "paid"})
+ assert coerced.source.move is False and coerced.source.credit == "none", (
+ f"W5 ticket-only flags are dropped when no ticket is involved ({coerced.source})"
+ )
+
+ assert (_rejects(code="X", source={"require_ticket": True, "credit": "maybe"})
+ and _pc(code="X", source={"require_ticket": True,
+ "credit": "paid"}).source.credit == "paid"), (
+ "W6 source credit must be 'none' or 'paid'"
+ )
+
+ both = _pc(code="BOTH", discount_percent=20, discount_fixed=5)
+ assert both.discount_fixed is None and both.discount_percent == 20, (
+ f"W7 percent and fixed discount stay exclusive "
+ f"({both.discount_percent}/{both.discount_fixed})"
+ )
+
+ capped = _pc(code="FIRST20", max_uses=20, used_count=0,
+ source=M.PromoSource(require_ticket=True))
+ assert capped.max_uses == 20 and capped.source.require_ticket is True, (
+ f"W8 max_uses and require_ticket now coexist ({capped.max_uses}/"
+ f"{capped.source.require_ticket})"
+ )
diff --git a/tests/test_promo_voucher.py b/tests/test_promo_voucher.py
new file mode 100644
index 0000000..0ae6865
--- /dev/null
+++ b/tests/test_promo_voucher.py
@@ -0,0 +1,154 @@
+"""Promo voucher (ticket-as-voucher) behaviour, ported from the standalone
+``test.py`` suite: 16 assertions across the A-F groups.
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from events import crud
+from events import models as M
+from events import services as S
+
+
+async def make_event(redemptions=1, extra_static=None, max_tickets=None):
+ ev = M.Event(
+ id="EV1", wallet="w1", name="Test", info="", canceled=False,
+ event_start_date="2026-01-01", event_end_date="2026-12-31",
+ currency="sat", admin_email=None, allow_fiat=False, fiat_currency="GBP",
+ amount_tickets=100, price_per_ticket=100, time=datetime.now(timezone.utc),
+ sold=0, banner=None,
+ extra=M.EventExtra(promo_codes=[
+ M.PromoCode(code="TICKETID", type="ticket", discount_percent=50,
+ combinable=False, redemptions_per_ticket=redemptions,
+ max_tickets=max_tickets)
+ ] + (extra_static or [])))
+ await crud.create_event(ev)
+ today = datetime.now(timezone.utc).date().isoformat()
+ tt = M.TicketType(id="TT1", event_id="EV1", name="GA", price=100,
+ available_from=today, available_to="2026-12-31")
+ await crud.create_ticket_type(tt)
+ return ev
+
+
+async def make_source(source_id, paid=True, extra=None, event="EV1"):
+ t = M.Ticket(
+ id=source_id, wallet="w1", event=event, name="Holder", email="a@b.c",
+ registered=False, paid=paid, time=datetime.now(timezone.utc),
+ reg_timestamp=datetime.now(timezone.utc), ticket_type_id="TT1",
+ basket_id=None, extra=extra or M.TicketExtra())
+ await crud.update_ticket(t)
+ return t
+
+
+async def calc(ev, codes, qty=1):
+ return await S.calculate_basket_total(
+ ev, [{"ticket_type_id": "TT1", "quantity": qty}], codes
+ )
+
+
+def test_a_voucher_lifecycle():
+ async def run():
+ ev = await make_event(redemptions=1)
+ await make_source("SRC1")
+ t1 = await calc(ev, ["SRC1"])
+ assert t1.total == 50, "A1 voucher applies on single ticket"
+ assert t1.discounts_applied[0].code == "SRC1", (
+ "A2 discount attributable to source id"
+ )
+ await S._reserve_ticket_voucher(ev, "SRC1", "basket-A")
+ src = await crud.get_ticket("SRC1")
+ assert len(src.extra.ticket_id_voucher_redemptions) == 1, (
+ "A3 reservation recorded pending"
+ )
+ assert src.extra.ticket_id_voucher_redemptions[0]["status"] == "pending", (
+ "A4 reservation pending status"
+ )
+ with pytest.raises(ValueError):
+ await S._reserve_ticket_voucher(ev, "SRC1", "basket-B") # A5 concurrent second reservation blocked
+ await S._finalize_ticket_voucher(src, "basket-A")
+ src = await crud.get_ticket("SRC1")
+ assert src.extra.ticket_id_voucher_redemptions[0]["status"] == "completed", (
+ "A6 finalized completed"
+ )
+ t1b = await calc(ev, ["SRC1"])
+ assert t1b.discount == 0, "A7 no discount after redemption used"
+
+ asyncio.run(run())
+
+
+def test_b_redemption_limit():
+ async def run():
+ ev3 = await make_event(redemptions=3)
+ await make_source("SRC3")
+ for i in range(3):
+ await S._reserve_ticket_voucher(ev3, "SRC3", f"b3-{i}")
+ t3 = await calc(ev3, ["SRC3"])
+ assert t3.discount == 0, "B1 voucher blocked at limit=3"
+ with pytest.raises(ValueError):
+ await S._reserve_ticket_voucher(ev3, "SRC3", "b3-x") # B2 4th reservation blocked
+
+ asyncio.run(run())
+
+
+def test_c_max_tickets_gate():
+ # The voucher applicability gate is max_tickets (None = any basket size),
+ # which replaced the old hardwired single-ticket basket rule.
+ async def run():
+ evc = await make_event(redemptions=1)
+ await make_source("SRCC")
+ tc = await calc(evc, ["SRCC"], qty=2)
+ assert tc.discount == 100 and tc.total == 100, (
+ "C1 voucher applies on multi-ticket basket when max_tickets is None"
+ )
+ evc2 = await make_event(redemptions=1, max_tickets=1)
+ await make_source("SRCC2")
+ tc2 = await calc(evc2, ["SRCC2"], qty=2)
+ assert tc2.discount == 0 and tc2.total == 200, (
+ "C2 voucher blocked when item_count exceeds max_tickets"
+ )
+
+ asyncio.run(run())
+
+
+def test_d_unpaid_ticket_is_not_a_voucher():
+ async def run():
+ evd = await make_event(redemptions=1)
+ await make_source("SRCD", paid=False)
+ td = await calc(evd, ["SRCD"])
+ assert td.discount == 0, "D1 unpaid ticket not a valid voucher"
+
+ asyncio.run(run())
+
+
+def test_e_expired_pending_pruned():
+ async def run():
+ eve = await make_event(redemptions=1)
+ old = M.TicketExtra(ticket_id_voucher_redemptions=[{
+ "basket_id": "dead", "status": "pending",
+ "time": (datetime.now(timezone.utc) - timedelta(days=2)).isoformat(),
+ "expires_at": (datetime.now(timezone.utc) - timedelta(days=1)).isoformat()}])
+ await make_source("SRCE", extra=old)
+ te = await calc(eve, ["SRCE"])
+ assert te.discount == 50, "E1 expired pending not counted -> voucher available"
+ await S._reserve_ticket_voucher(eve, "SRCE", "new-basket")
+ src = await crud.get_ticket("SRCE")
+ assert (
+ len(src.extra.ticket_id_voucher_redemptions) == 1
+ and src.extra.ticket_id_voucher_redemptions[0]["basket_id"] == "new-basket"
+ ), "E2 expired entry pruned on reserve"
+
+ asyncio.run(run())
+
+
+def test_f_static_promo_does_not_stack():
+ async def run():
+ evf = await make_event(redemptions=1, extra_static=[
+ M.PromoCode(code="CHEAP", discount_percent=10, combinable=True)])
+ await make_source("SRCF")
+ tf = await calc(evf, ["CHEAP"])
+ assert tf.discount == 10, "F1 static promo still applies"
+ tf2 = await calc(evf, ["CHEAP", "SRCF"])
+ assert tf2.discount == 50, "F2 static+voucher does not stack"
+
+ asyncio.run(run())
diff --git a/uv.lock b/uv.lock
index a9cdbab..1ec6d82 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1088,7 +1088,7 @@ wheels = [
[[package]]
name = "lnbits-events"
-version = "0.0.0"
+version = "2.0.0"
source = { virtual = "." }
dependencies = [
{ name = "lnbits" },
diff --git a/views.py b/views.py
index 4a3f142..a9c8a6c 100644
--- a/views.py
+++ b/views.py
@@ -1,6 +1,10 @@
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException, Request
from lnbits.core.views.generic import index, index_public
from lnbits.decorators import check_account_id_exists
+from lnbits.helpers import template_renderer
+from starlette.responses import HTMLResponse
+
+from .crud import get_event
events_generic_router = APIRouter()
@@ -11,8 +15,18 @@
dependencies=[Depends(check_account_id_exists)],
)
+
+async def event_public(request: Request, event_id: str) -> HTMLResponse:
+ event = await get_event(event_id)
+ return template_renderer(additional_folders=["events/templates"]).TemplateResponse(
+ request,
+ "event_public.html",
+ {"public": True, "event": event.dict() if event else None},
+ )
+
+
events_generic_router.add_api_route(
- "/{event_id}", methods=["GET"], endpoint=index_public
+ "/{event_id}", methods=["GET"], endpoint=event_public
)
events_generic_router.add_api_route(
@@ -22,3 +36,7 @@
events_generic_router.add_api_route(
"/register/{event_id}", methods=["GET"], endpoint=index_public
)
+
+events_generic_router.add_api_route(
+ "/basket/{basket_id}", methods=["GET"], endpoint=index_public
+)
diff --git a/views_api.py b/views_api.py
index c55ffa0..4fb1080 100644
--- a/views_api.py
+++ b/views_api.py
@@ -1,9 +1,6 @@
-import asyncio
from datetime import datetime, timezone
from http import HTTPStatus
from io import BytesIO
-from pathlib import Path
-from typing import Any
import pyqrcode # type: ignore[import-untyped]
from fastapi import (
@@ -12,116 +9,80 @@
HTTPException,
Query,
Request,
- WebSocket,
- WebSocketDisconnect,
)
-from fastapi.responses import StreamingResponse
+from fastapi.responses import HTMLResponse, StreamingResponse
from lnbits.core.crud import get_user
-from lnbits.core.crud.assets import get_public_asset
-from lnbits.core.crud.wallets import get_wallet
from lnbits.core.models import WalletTypeInfo
-from lnbits.core.models.payments import CreateInvoice
-from lnbits.core.services import create_payment_request
-from lnbits.db import Filters, Page
+from lnbits.db import Filters
from lnbits.decorators import (
parse_filters,
require_admin_key,
require_invoice_key,
)
-from lnbits.helpers import generate_filter_params_openapi, urlsafe_short_hash
+from lnbits.helpers import urlsafe_short_hash
from lnbits.settings import settings
-from lnbits.utils.exchange_rates import (
- fiat_amount_as_satoshis,
- get_fiat_rate_satoshis,
- satoshis_amount_as_fiat,
-)
from loguru import logger
from PIL import Image, ImageDraw
from .crud import (
- create_event,
- create_ticket,
- delete_event,
+ create_event as create_event_crud,
+ create_ticket_type,
+ delete_event as delete_event_crud,
+ delete_event_ticket_types,
delete_event_tickets,
delete_ticket,
+ delete_ticket_type,
+ get_basket,
+ get_basket_tickets,
get_event,
- get_events,
+ get_event_tickets,
get_ticket,
+ get_ticket_types,
get_tickets,
+ get_tickets_by_satspay_charge_id,
get_tickets_paginated,
purge_unpaid_tickets,
- update_event,
+ update_event as update_event_crud,
update_ticket,
+ update_ticket_type,
+ get_events,
)
from .models import (
+ BasketItem,
+ BasketResponse,
+ BasketTotals,
+ CreateBasket,
CreateEvent,
CreateTicket,
Event,
+ PromoValidateRequest,
PublicEvent,
PublicTicket,
Ticket,
TicketFilters,
TicketPaymentRequest,
- TicketResendResult,
- ensure_ticket_waves,
- get_active_ticket_waves,
+ TicketType,
)
from .services import (
- create_satspay_charge,
- fetch_watchonly_config,
- fetch_watchonly_wallets,
+ backfill_paid_values,
+ calculate_basket_total,
+ create_basket_with_charge,
get_satspay_charge,
+ handle_basket_payment,
+ preview_admin_email_html,
+ preview_ticket_email_html,
refund_tickets,
resend_ticket_email_notification,
- send_ticket_notification_in_background,
- set_ticket_paid,
-)
-from .tasks import (
- deregister_payment_listener,
- payment_listeners,
- register_payment_listener,
+ send_bulk_message,
+ send_bulk_ticket_emails,
+ toggle_ticket_deactivation,
)
-events_api_router = APIRouter(prefix="/api/v1/events")
-tickets_api_router = APIRouter(prefix="/api/v1/tickets")
+events_api_router = APIRouter(prefix="/api/v1")
qr_api_router = APIRouter(prefix="/api/v1")
-tickets_filters = parse_filters(TicketFilters)
-
-
-async def _get_watchonly_status(wallet) -> dict[str, Any]:
- try:
- config = await fetch_watchonly_config(wallet.inkey)
- network = config.get("network")
- if not network:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Watchonly extension returned an invalid network.",
- )
- wallets = await fetch_watchonly_wallets(wallet.inkey, network)
- except HTTPException:
- raise
- except Exception as exc:
- return {
- "available": False,
- "message": f"Watchonly extension is not reachable: {exc!s}",
- "network": None,
- "wallets": [],
- "mempool_endpoint": None,
- }
- return {
- "available": True,
- "message": None,
- "network": network,
- "wallets": wallets,
- "mempool_endpoint": config.get("mempool_endpoint"),
- }
-
-
-def _is_fiat_currency(currency: str | None) -> bool:
- return str(currency or "").lower() not in {"sat", "sats"}
-def make_qr_png(data: str, size: int = 235, border: int = 4) -> Image.Image:
+def _make_qr_png(data: str, size: int = 235, border: int = 4) -> Image.Image:
qr = pyqrcode.create(data)
matrix = qr.code
modules = len(matrix)
@@ -149,101 +110,112 @@ def make_qr_png(data: str, size: int = 235, border: int = 4) -> Image.Image:
return img
-@events_api_router.get("")
+@events_api_router.post("/events")
+async def api_create_event(
+ data: CreateEvent,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> Event:
+ event = await create_event_crud(data)
+ return event
+
+
+@events_api_router.put("/events/{event_id}")
+async def api_update_event(
+ event_id: str,
+ data: CreateEvent,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> Event:
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(
+ status_code=HTTPStatus.FORBIDDEN, detail="Not your event."
+ )
+ event = Event(
+ **{
+ **event.dict(),
+ **data.dict(),
+ "id": event.id,
+ "wallet": event.wallet,
+ "time": event.time,
+ "sold": event.sold,
+ "canceled": event.canceled,
+ }
+ )
+ event = await update_event_crud(event)
+ return event
+
+
+@events_api_router.get("/events")
async def api_events(
all_wallets: bool = Query(False),
wallet: WalletTypeInfo = Depends(require_invoice_key),
) -> list[Event]:
wallet_ids = [wallet.wallet.id]
-
if all_wallets:
user = await get_user(wallet.wallet.user)
wallet_ids = user.wallet_ids if user else []
-
return await get_events(wallet_ids)
-@events_api_router.get("/onchain/status")
-async def api_onchain_status(
- wallet: WalletTypeInfo = Depends(require_admin_key),
-) -> dict[str, Any]:
- return await _get_watchonly_status(wallet.wallet)
-
-
-@events_api_router.get("/{event_id}", response_model=PublicEvent)
+@events_api_router.get("/events/{event_id}", response_model=PublicEvent)
async def api_get_event(event_id: str) -> Event:
event = await get_event(event_id)
if not event:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
+ if event.canceled:
+ raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is canceled.")
+
await purge_unpaid_tickets(event_id)
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+
+ ticket_types = await get_ticket_types(event_id)
+ if not ticket_types:
+ raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is sold out.")
+ today_date = datetime.now(timezone.utc).date()
+ available_types = [
+ tt for tt in ticket_types
+ if datetime.strptime(tt.available_from, "%Y-%m-%d").date() <= today_date
+ and datetime.strptime(tt.available_to, "%Y-%m-%d").date() >= today_date
+ and (tt.max_tickets == 0 or tt.sold < tt.max_tickets)
+ ]
+ if not available_types:
+ raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is sold out.")
- today = datetime.now(timezone.utc).date()
- active_waves = get_active_ticket_waves(event, today)
- is_sales_closed = today > datetime.strptime(event.closing_date, "%Y-%m-%d").date()
+ latest_available_to = max(
+ datetime.strptime(tt.available_to, "%Y-%m-%d").date() for tt in ticket_types
+ )
+ is_sales_closed = today_date > latest_available_to
is_min_tickets_met = (
event.sold >= event.extra.min_tickets if event.extra.conditional else True
)
- if event.amount_tickets < 1:
- raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is sold out.")
if event.extra.conditional and not is_min_tickets_met and is_sales_closed:
event.canceled = True
- await update_event(event)
+ await update_event_crud(event)
await refund_tickets(event_id)
-
raise HTTPException(status_code=HTTPStatus.GONE, detail="Event canceled.")
- if not active_waves:
+ if is_sales_closed:
raise HTTPException(
- status_code=HTTPStatus.GONE,
- detail=(
- "Ticket closing date has passed."
- if is_sales_closed
- else "No ticket wave is currently open."
- ),
+ status_code=HTTPStatus.GONE, detail="Ticket closing date has passed."
)
- return event
-
-
-@events_api_router.post("")
-@events_api_router.put("/{event_id}")
-async def api_event_create(
- data: CreateEvent,
- wallet: WalletTypeInfo = Depends(require_admin_key),
- event_id: str | None = None,
-) -> Event:
- if event_id:
- event = await get_event(event_id)
- if not event:
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
- )
-
- if event.wallet != wallet.wallet.id:
- raise HTTPException(
- status_code=HTTPStatus.FORBIDDEN, detail="Not your event."
- )
- event = Event(
- **{
- **event.dict(),
- **data.dict(),
- "id": event.id,
- "wallet": event.wallet,
- "time": event.time,
- "sold": event.sold,
- "canceled": event.canceled,
- }
- )
- event = await update_event(event)
- else:
- event = await create_event(data)
-
- return event
+ public = Event(
+ **{**event.dict(), "extra": {**event.extra.dict(), "promo_codes": []}}
+ )
+ return public
-@events_api_router.put("/{event_id}/cancel")
+@events_api_router.put("/events/{event_id}/cancel")
async def api_event_cancel(
event_id: str,
wallet: WalletTypeInfo = Depends(require_admin_key),
@@ -253,18 +225,16 @@ async def api_event_cancel(
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
-
if event.wallet != wallet.wallet.id:
raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
event.canceled = True
- event = await update_event(event)
+ event = await update_event_crud(event)
await refund_tickets(event.id)
-
return event
-@events_api_router.delete("/{event_id}")
-async def api_form_delete(
+@events_api_router.delete("/events/{event_id}")
+async def api_event_delete(
event_id: str, wallet: WalletTypeInfo = Depends(require_admin_key)
) -> None:
event = await get_event(event_id)
@@ -272,405 +242,522 @@ async def api_form_delete(
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ await delete_event_tickets(event_id)
+ await delete_event_ticket_types(event_id)
+ await delete_event_crud(event_id)
+
+@events_api_router.post("/events/{event_id}/email-tickets")
+async def api_event_email_tickets(
+ event_id: str,
+ request: Request,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+ subject: str | None = Query(None),
+ body: str | None = Query(None),
+):
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
if event.wallet != wallet.wallet.id:
raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ base_url = str(request.base_url).rstrip("/")
+ results = await send_bulk_ticket_emails(event_id, base_url)
+ return {"results": [r.dict() for r in results]}
- await delete_event(event_id)
- await delete_event_tickets(event_id)
+@events_api_router.post("/events/{event_id}/email-message")
+async def api_event_email_message(
+ event_id: str,
+ body: dict,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+):
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ subject = body.get("subject")
+ message = body.get("message")
+ if not subject or not message:
+ raise HTTPException(
+ status_code=HTTPStatus.BAD_REQUEST,
+ detail="Both 'subject' and 'message' are required.",
+ )
+ results = await send_bulk_message(event_id, subject, message)
+ return {"results": results}
-@tickets_api_router.get("")
-async def api_tickets(
- all_wallets: bool = Query(False),
- key_info: WalletTypeInfo = Depends(require_admin_key),
-) -> list[Ticket]:
- wallet_ids = [key_info.wallet.id]
- if all_wallets:
- user = await get_user(key_info.wallet.user)
- wallet_ids = user.wallet_ids if user else []
+@events_api_router.get("/events/{event_id}/preview/ticket-email")
+async def api_preview_ticket_email(
+ event_id: str,
+ request: Request,
+ key: str | None = Query(None),
+ basket_id: str | None = Query(None),
+):
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ wallet = await _check_preview_key(event, key)
+ if not wallet:
+ raise HTTPException(
+ status_code=HTTPStatus.FORBIDDEN, detail="Invalid key."
+ )
+ base_url = str(request.base_url).rstrip("/")
+
+ basket = None
+ tickets: list[Ticket] = []
+ if basket_id:
+ basket = await get_basket(basket_id)
+ if basket and basket.event_id == event_id:
+ tickets = await get_basket_tickets(basket_id)
+ if not tickets:
+ tickets = [t for t in await get_event_tickets(event_id) if t.paid][:1]
+ if not tickets:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND,
+ detail="No paid tickets found for this event.",
+ )
+ for t in tickets:
+ if not t.extra.ticket_base_url:
+ t.extra.ticket_base_url = base_url
- return await get_tickets(wallet_ids)
+ subject, html = await preview_ticket_email_html(event, basket, tickets, base_url)
+ return HTMLResponse(content=html)
-@tickets_api_router.get(
- "/paginated",
- summary="Get paginated list of tickets",
- openapi_extra=generate_filter_params_openapi(TicketFilters),
- response_model=Page[Ticket],
-)
-async def api_tickets_paginated(
- all_wallets: bool = Query(False),
- filters: Filters = Depends(tickets_filters),
- key_info: WalletTypeInfo = Depends(require_admin_key),
-) -> Page[Ticket]:
- wallet_ids = [key_info.wallet.id]
+@events_api_router.get("/events/{event_id}/preview/admin-email")
+async def api_preview_admin_email(
+ event_id: str,
+ request: Request,
+ key: str | None = Query(None),
+ basket_id: str | None = Query(None),
+):
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ wallet = await _check_preview_key(event, key)
+ if not wallet:
+ raise HTTPException(
+ status_code=HTTPStatus.FORBIDDEN, detail="Invalid key."
+ )
+ base_url = str(request.base_url).rstrip("/")
+
+ basket = None
+ tickets: list[Ticket] = []
+ if basket_id:
+ basket = await get_basket(basket_id)
+ if basket and basket.event_id == event_id:
+ tickets = await get_basket_tickets(basket_id)
+ if not basket or not tickets:
+ all_tickets = [t for t in await get_event_tickets(event_id) if t.paid]
+ if all_tickets:
+ tickets = all_tickets[:2]
+ for t in all_tickets:
+ if t.basket_id:
+ basket = await get_basket(t.basket_id)
+ if basket:
+ break
+ if not basket or not tickets:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND,
+ detail="No paid tickets/baskets found for this event.",
+ )
+ for t in tickets:
+ if not t.extra.ticket_base_url:
+ t.extra.ticket_base_url = base_url
- if all_wallets:
- user = await get_user(key_info.wallet.user)
- wallet_ids = user.wallet_ids if user else []
+ subject, html = await preview_admin_email_html(event, basket, tickets, base_url)
+ return HTMLResponse(content=html)
- if not filters.sortby:
- filters.sortby = "time"
- if not filters.direction:
- filters.direction = "desc"
- return await get_tickets_paginated(wallet_ids, filters)
+async def _check_preview_key(event: Event, key: str | None):
+ from lnbits.core.crud import get_wallet_for_key
+ wallet = await get_wallet_for_key(key)
+ if wallet and wallet.id == event.wallet:
+ return wallet
+ return None
-@tickets_api_router.get("/{ticket_id}", response_model=PublicTicket)
-async def api_get_ticket(ticket_id: str) -> Ticket:
- ticket = await get_ticket(ticket_id)
- if not ticket:
+@events_api_router.get("/ticket-types/{event_id}")
+async def api_ticket_types(event_id: str) -> list[TicketType]:
+ return await get_ticket_types(event_id)
+
+
+@events_api_router.post("/ticket-types/{event_id}")
+async def api_create_ticket_type(
+ event_id: str,
+ data: TicketType,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> TicketType:
+ event = await get_event(event_id)
+ if not event:
raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- event = await get_event(ticket.event)
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ data.event_id = event_id
+ return await create_ticket_type(data)
+
+
+@events_api_router.put("/ticket-types/{event_id}/{type_id}")
+async def api_update_ticket_type(
+ event_id: str,
+ type_id: str,
+ data: TicketType,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> TicketType:
+ event = await get_event(event_id)
if not event:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- return ticket
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ data.event_id = event_id
+ data.id = type_id
+ return await update_ticket_type(data)
-@qr_api_router.get("/qr/{ticket_id}", response_class=StreamingResponse)
-async def api_ticket_qr(ticket_id: str):
- ticket = await get_ticket(ticket_id)
- if not ticket:
+@events_api_router.delete("/ticket-types/{event_id}/{type_id}")
+async def api_delete_ticket_type(
+ event_id: str,
+ type_id: str,
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> None:
+ event = await get_event(event_id)
+ if not event:
raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
+ if event.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your event.")
+ await delete_ticket_type(type_id)
- event = await get_event(ticket.event)
+
+@events_api_router.post("/baskets/{event_id}")
+async def api_create_basket(
+ event_id: str,
+ data: CreateBasket,
+ request: Request,
+) -> BasketResponse:
+ event = await get_event(event_id)
if not event:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
+ if event.canceled:
+ raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is canceled.")
- waves = ensure_ticket_waves(event)
- wave = next(
- (wave for wave in waves if wave.id == ticket.extra.ticket_wave_id),
- waves[0],
- )
+ from lnbits.core.crud.wallets import get_wallet as get_wallet_crud
- qr_img = make_qr_png(f"ticket://{ticket_id}", size=157)
- output = BytesIO()
+ wallet_record = await get_wallet_crud(event.wallet)
+ if not wallet_record:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Wallet does not exist."
+ )
- if not wave.use_ticket_image:
- qr_img.save(output, format="PNG")
- output.seek(0)
- return StreamingResponse(
- output,
- media_type="image/png",
- headers={
- "Cache-Control": "no-cache, no-store, must-revalidate",
- "Pragma": "no-cache",
- "Expires": "0",
- },
- )
-
- background_bytes = None
- if wave.ticket_image_id:
- asset = await get_public_asset(wave.ticket_image_id)
- if asset:
- background_bytes = asset.data
-
- if background_bytes:
- ticket_image = Image.open(BytesIO(background_bytes)).convert("RGBA")
- else:
- default_template = (
- Path(__file__).resolve().parent / "static" / "image" / "ticket.jpg"
- )
- ticket_image = Image.open(default_template).convert("RGBA")
-
- ticket_image.paste(qr_img, (122, 505))
- ticket_image.save(output, format="PNG")
- output.seek(0)
+ try:
+ basket, tickets, totals, charge = await create_basket_with_charge(
+ event=event,
+ data=data,
+ wallet_inkey=wallet_record.inkey,
+ base_url=str(request.base_url).rstrip("/"),
+ )
+ except ValueError as exc:
+ raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(exc))
- return StreamingResponse(
- output,
- media_type="image/png",
- headers={
- "Cache-Control": "no-cache, no-store, must-revalidate",
- "Pragma": "no-cache",
- "Expires": "0",
- },
+ payment_request = None
+ if charge:
+ payment_request = TicketPaymentRequest(
+ payment_hash=basket.id,
+ onchain_amount_sat=totals.total,
+ satspay_charge_url=f"/satspay/{charge['id']}",
+ basket_id=basket.id,
+ )
+
+ return BasketResponse(
+ basket=basket,
+ tickets=tickets,
+ totals=totals,
+ payment_request=payment_request,
+ event_name=event.name,
+ event_currency=event.currency,
+ event_fiat_currency=event.fiat_currency,
)
-@tickets_api_router.post("/{event_id}")
-async def api_ticket_create(
- event_id: str, data: CreateTicket, request: Request
-) -> TicketPaymentRequest:
- event = await get_event(event_id)
+@events_api_router.get("/baskets/{basket_id}")
+async def api_get_basket(basket_id: str) -> BasketResponse:
+ basket = await get_basket(basket_id)
+ if not basket:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Basket does not exist."
+ )
+
+ event = await get_event(basket.event_id)
if not event:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- if event.canceled:
- raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is canceled.")
+ if not basket.paid and basket.satspay_charge_id:
+ from lnbits.core.crud.wallets import get_wallet as get_wallet_crud
- if event.amount_tickets < 1:
- raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is sold out.")
+ wallet_record = await get_wallet_crud(event.wallet)
+ if wallet_record:
+ try:
+ charge = await get_satspay_charge(
+ wallet_record.inkey, basket.satspay_charge_id
+ )
+ if charge.get("paid"):
+ await handle_basket_payment(basket_id, wallet_record.inkey)
+ basket = await get_basket(basket_id) or basket
+ except Exception as exc:
+ logger.warning(f"Failed to reconcile basket {basket_id}: {exc}")
+
+ from .crud import get_basket_tickets as get_basket_tickets_crud
+ from .services import calculate_basket_total
+
+ tickets = await get_basket_tickets_crud(basket_id)
+ totals = await calculate_basket_total(
+ event,
+ [{"ticket_type_id": t.ticket_type_id or "", "quantity": 1}
+ for t in tickets],
+ basket.promo_codes,
+ allow_consumed=True,
+ )
- name = data.name
- email = data.email
- promo_code = data.promo_code.upper() if data.promo_code else None
- refund_address = data.refund_address
- nostr_identifier = data.nostr_identifier.strip() if data.nostr_identifier else None
- payment_method = (data.payment_method or "lightning").lower()
- if payment_method not in {"lightning", "fiat", "onchain"}:
+ return BasketResponse(
+ basket=basket,
+ tickets=tickets,
+ totals=totals,
+ payment_request=None,
+ event_name=event.name,
+ event_currency=event.currency,
+ event_fiat_currency=event.fiat_currency,
+ )
+
+
+@events_api_router.post("/promo/validate/{event_id}")
+async def api_validate_promo_codes(
+ event_id: str, data: PromoValidateRequest
+) -> BasketTotals:
+ event = await get_event(event_id)
+ if not event:
raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Unsupported payment method.",
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- if nostr_identifier and "@" not in nostr_identifier:
+ return await calculate_basket_total(event, data.items, data.codes)
+
+
+@events_api_router.post("/maintenance/backfill-paid-values")
+async def api_backfill_paid_values(
+ wallet: WalletTypeInfo = Depends(require_admin_key),
+) -> dict:
+ """Backfill ``TicketExtra.paid_value`` for tickets sold before it existed.
+
+ ``paid_value`` is the credit basis when a ticket is upgraded, so tickets
+ sold earlier need it reconstructed — from the basket that charged them,
+ falling back to the ticket type's price. Idempotent: tickets that already
+ have a value are left alone, so it is safe to re-run.
+ """
+ return await backfill_paid_values(wallet.wallet.id)
+
+
+@events_api_router.post("/baskets/{basket_id}/satspay-webhook")
+async def api_basket_satspay_webhook(
+ basket_id: str,
+ request: Request,
+):
+ body = await request.json()
+ charge_id = body.get("charge_id")
+ logger.debug(f"SatsPay webhook for basket {basket_id}, charge {charge_id}")
+
+ basket = await get_basket(basket_id)
+ if not basket:
raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Only NIP-05 Nostr identifiers are supported.",
+ status_code=HTTPStatus.NOT_FOUND, detail="Basket does not exist."
)
- active_waves = get_active_ticket_waves(event)
- if not active_waves:
+
+ event = await get_event(basket.event_id)
+ if not event:
raise HTTPException(
- status_code=HTTPStatus.GONE, detail="No ticket wave is currently open."
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- selected_wave = None
- if data.ticket_wave_id:
- selected_wave = next(
- (wave for wave in active_waves if wave.id == data.ticket_wave_id),
- None,
- )
- if not selected_wave:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Invalid ticket wave selected.",
- )
- elif len(active_waves) == 1:
- selected_wave = active_waves[0]
- else:
+ from lnbits.core.crud.wallets import get_wallet as get_wallet_crud
+
+ wallet_record = await get_wallet_crud(event.wallet)
+ if not wallet_record:
raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Please select a ticket wave.",
+ status_code=HTTPStatus.NOT_FOUND, detail="Wallet does not exist."
)
- price = selected_wave.price_per_ticket
- extra: dict[str, Any] = {"tag": "events", "name": name, "email": email}
+ await handle_basket_payment(basket_id, wallet_record.inkey)
+ return {"status": "ok"}
- if promo_code:
- # check if promo_code exists in event.extra.promo_codes
- if promo_code not in [pc.code for pc in event.extra.promo_codes]:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST, detail="Invalid promo code."
- )
- # get the promocode
- promo = next(pc for pc in event.extra.promo_codes if pc.code == promo_code)
- extra["promo_code"] = promo.code
- price = selected_wave.price_per_ticket * (1 - promo.discount_percent / 100)
- if payment_method == "fiat" and not selected_wave.allow_fiat:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Fiat payments are not enabled for this event.",
- )
-
- if _is_fiat_currency(selected_wave.currency):
- extra["fiat"] = True
- extra["currency"] = selected_wave.currency
- extra["fiatAmount"] = price
- extra["rate"] = await get_fiat_rate_satoshis(selected_wave.currency)
-
- if payment_method != "fiat":
- price = await fiat_amount_as_satoshis(price, selected_wave.currency)
-
- invoice_unit = selected_wave.currency
- fiat_amount = price
- fiat_provider = None
- onchain_amount_sat = None
-
- if payment_method == "fiat":
- if _is_fiat_currency(selected_wave.currency):
- invoice_unit = selected_wave.currency
- else:
- invoice_unit = selected_wave.fiat_currency
- fiat_amount = await satoshis_amount_as_fiat(price, invoice_unit)
- extra["fiat"] = True
- extra["currency"] = invoice_unit
- extra["fiatAmount"] = fiat_amount
- extra["rate"] = await get_fiat_rate_satoshis(invoice_unit)
- wallet = await get_wallet(event.wallet)
- if not wallet:
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND,
- detail="Event wallet does not exist.",
- )
- providers = settings.get_fiat_providers_for_user(wallet.user)
- fiat_provider = data.fiat_provider or (providers[0] if providers else None)
- if not fiat_provider:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="No fiat payment provider configured for this event.",
- )
- elif payment_method == "onchain":
- onchain_amount_sat = int(price)
- wallet_record = await get_wallet(event.wallet)
- if not wallet_record:
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND,
- detail="Event wallet does not exist.",
- )
- if not event.extra.onchain_enabled:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Onchain payments are not enabled for this event.",
- )
- if not event.extra.onchain_wallet_id:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="No onchain wallet configured for this event.",
- )
+@events_api_router.get("/tickets")
+async def api_tickets(
+ all_wallets: bool = Query(False),
+ key_info: WalletTypeInfo = Depends(require_invoice_key),
+) -> list[Ticket]:
+ wallet_ids = [key_info.wallet.id]
+ if all_wallets:
+ user = await get_user(key_info.wallet.user)
+ wallet_ids = user.wallet_ids if user else []
+ return await get_tickets(wallet_ids)
+
+
+@events_api_router.get("/tickets/paginated")
+async def api_tickets_paginated(
+ all_wallets: bool = Query(False),
+ filters: Filters = Depends(parse_filters(TicketFilters)),
+ key_info: WalletTypeInfo = Depends(require_invoice_key),
+):
+ wallet_ids = [key_info.wallet.id]
+ if all_wallets:
+ user = await get_user(key_info.wallet.user)
+ wallet_ids = user.wallet_ids if user else []
+ return await get_tickets_paginated(wallet_ids, filters)
+
- ticket_id = urlsafe_short_hash()
- base_url = str(request.base_url).rstrip("/")
- # Use internal address for webhook — SatsPay calls it server-side and
- # cannot reach the public domain from within the container.
- internal_base = f"http://{settings.host}:{settings.port}"
- webhook_url = (
- f"{internal_base}/events/api/v1/tickets/{ticket_id}/satspay-webhook"
- )
- complete_url = f"{base_url}/events/ticket/{ticket_id}"
- try:
- charge = await create_satspay_charge(
- api_key=wallet_record.inkey,
- data={
- "amount": onchain_amount_sat,
- "description": f"Ticket for {event.name}",
- "name": name,
- "onchainwallet": event.extra.onchain_wallet_id,
- "zeroconf": event.extra.onchain_zeroconf,
- "fasttrack": event.extra.onchain_fasttrack,
- "webhook": webhook_url,
- "completelink": complete_url,
- "completelinktext": "View your ticket",
- "time": 1440,
- },
+@events_api_router.get("/tickets/by-charge/{charge_id}")
+async def api_get_tickets_by_charge(charge_id: str) -> list[PublicTicket]:
+ tickets = await get_tickets_by_satspay_charge_id(charge_id)
+ if not tickets:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="No tickets found for this charge."
+ )
+ event = await get_event(tickets[0].event)
+ event_name = event.name if event else ""
+ from .crud import get_ticket_types as get_event_ticket_types
+ ticket_types = await get_event_ticket_types(tickets[0].event)
+ tt_map = {tt.id: tt.name for tt in ticket_types}
+ result = []
+ for ticket in tickets:
+ result.append(
+ PublicTicket(
+ event=ticket.event,
+ event_name=event_name,
+ ticket_type_name=tt_map.get(ticket.ticket_type_id or "", ""),
+ name=ticket.name,
+ registered=ticket.registered,
+ paid=ticket.paid,
+ time=ticket.time,
+ reg_timestamp=ticket.reg_timestamp,
)
- except Exception as exc:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail=f"Failed to create SatsPay charge: {exc}",
- ) from exc
-
- await create_ticket(
- payment_hash=ticket_id,
- wallet=event.wallet,
- event=event.id,
- name=name,
- email=email,
- extra={
- "applied_promo_code": promo_code,
- "ticket_wave_id": selected_wave.id,
- "ticket_wave_title": selected_wave.title,
- "refund_address": refund_address,
- "nostr_identifier": nostr_identifier,
- "ticket_base_url": base_url,
- "sats_paid": onchain_amount_sat,
- "onchain": True,
- "satspay_charge_id": charge["id"],
- },
)
+ return result
- return TicketPaymentRequest(
- payment_hash=ticket_id,
- onchain_amount_sat=onchain_amount_sat,
- satspay_charge_url=f"/satspay/{charge['id']}",
+
+@events_api_router.get("/tickets/{ticket_id}", response_model=PublicTicket)
+async def api_get_ticket(ticket_id: str) -> PublicTicket:
+ ticket = await get_ticket(ticket_id)
+ if not ticket:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
- else:
- invoice_unit = "sat"
-
- payment = await create_payment_request(
- wallet_id=event.wallet,
- invoice_data=CreateInvoice(
- out=False,
- amount=fiat_amount if payment_method == "fiat" else price,
- unit=invoice_unit,
- fiat_provider=fiat_provider,
- memo=f"{event_id}",
- extra=extra,
- ),
- )
- await create_ticket(
- payment_hash=payment.payment_hash,
- wallet=event.wallet,
- event=event.id,
- name=name,
- email=email,
- extra={
- "applied_promo_code": promo_code,
- "ticket_wave_id": selected_wave.id,
- "ticket_wave_title": selected_wave.title,
- "refund_address": refund_address,
- "nostr_identifier": nostr_identifier,
- "ticket_base_url": str(request.base_url).rstrip("/"),
- "sats_paid": payment.sat,
- "onchain": False,
- },
+ event = await get_event(ticket.event)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ from .crud import get_ticket_types as get_event_ticket_types
+
+ ticket_types = await get_event_ticket_types(ticket.event)
+ tt_map = {tt.id: tt.name for tt in ticket_types}
+ return PublicTicket(
+ event=ticket.event,
+ event_name=event.name,
+ ticket_type_name=tt_map.get(ticket.ticket_type_id or "", ""),
+ name=ticket.name,
+ email=ticket.email,
+ registered=ticket.registered,
+ paid=ticket.paid,
+ deactivated=ticket.extra.deactivated,
+ time=ticket.time,
+ reg_timestamp=ticket.reg_timestamp,
)
- return TicketPaymentRequest(
- payment_hash=payment.payment_hash,
- payment_request=getattr(payment, "bolt11", None),
- fiat_payment_request=getattr(payment, "extra", {}).get("fiat_payment_request"),
- fiat_provider=getattr(payment, "fiat_provider", None) or fiat_provider,
- is_fiat=bool(getattr(payment, "fiat_provider", None) or fiat_provider),
+
+@events_api_router.post("/tickets/{event_id}")
+async def api_ticket_create_legacy(
+ event_id: str, data: CreateTicket, request: Request
+) -> TicketPaymentRequest:
+ event = await get_event(event_id)
+ if not event:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
+ )
+ if event.canceled:
+ raise HTTPException(status_code=HTTPStatus.GONE, detail="Event is canceled.")
+
+ ticket_types = await get_ticket_types(event_id)
+ if not ticket_types:
+ raise HTTPException(
+ status_code=HTTPStatus.BAD_REQUEST, detail="No ticket types available."
+ )
+
+ selected_tt_id = data.ticket_type_id
+ if not selected_tt_id:
+ selected_tt_id = ticket_types[0].id
+
+ basket_data = CreateBasket(
+ name=data.name,
+ email=data.email,
+ items=[BasketItem(ticket_type_id=selected_tt_id, quantity=1)],
+ promo_codes=[data.promo_code] if data.promo_code else [],
+ payment_method=data.payment_method,
+ fiat_provider=data.fiat_provider,
+ nostr_identifier=data.nostr_identifier,
+ refund_address=data.refund_address,
)
+ from lnbits.core.crud.wallets import get_wallet as get_wallet_crud
-@tickets_api_router.websocket("/ws/{payment_hash}")
-async def websocket_endpoint(payment_hash: str, websocket: WebSocket) -> None:
- await websocket.accept()
- queue: asyncio.Queue[Ticket] = asyncio.Queue()
- register_payment_listener(payment_hash, queue)
- disconnect_task: asyncio.Task | None = None
- payment_task: asyncio.Task | None = None
+ wallet_record = await get_wallet_crud(event.wallet)
+ if not wallet_record:
+ raise HTTPException(
+ status_code=HTTPStatus.NOT_FOUND, detail="Wallet does not exist."
+ )
try:
- ticket = await get_ticket(payment_hash)
- if ticket and ticket.paid:
- await websocket.send_json({"paid": True})
- return
-
- while True:
- disconnect_task = asyncio.create_task(websocket.receive_text())
- payment_task = asyncio.create_task(queue.get())
- done, pending = await asyncio.wait(
- {disconnect_task, payment_task}, return_when=asyncio.FIRST_COMPLETED
- )
-
- for task in pending:
- task.cancel()
+ basket, tickets, totals, charge = await create_basket_with_charge(
+ event=event,
+ data=basket_data,
+ wallet_inkey=wallet_record.inkey,
+ base_url=str(request.base_url).rstrip("/"),
+ )
+ except ValueError as exc:
+ raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(exc))
- if disconnect_task in done:
- try:
- disconnect_task.result()
- except WebSocketDisconnect:
- pass
- break
+ if charge:
+ return TicketPaymentRequest(
+ payment_hash=basket.id,
+ onchain_amount_sat=totals.total,
+ satspay_charge_url=f"/satspay/{charge['id']}",
+ basket_id=basket.id,
+ )
- ticket = payment_task.result()
- await websocket.send_json({"paid": ticket.paid})
- if ticket.paid:
- break
- finally:
- for pending_task in (disconnect_task, payment_task):
- if pending_task and not pending_task.done():
- pending_task.cancel()
- deregister_payment_listener(payment_hash, queue)
+ return TicketPaymentRequest(
+ payment_hash=basket.id,
+ basket_id=basket.id,
+ )
-@tickets_api_router.delete("/{ticket_id}")
+@events_api_router.delete("/tickets/{ticket_id}")
async def api_ticket_delete(
ticket_id: str, wallet: WalletTypeInfo = Depends(require_admin_key)
) -> None:
@@ -679,131 +766,102 @@ async def api_ticket_delete(
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
-
if ticket.wallet != wallet.wallet.id:
raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your ticket.")
-
await delete_ticket(ticket_id)
-@tickets_api_router.post("/{ticket_id}/satspay-webhook")
-async def api_ticket_satspay_webhook(ticket_id: str) -> None:
+@events_api_router.put("/tickets/{ticket_id}/deactivate")
+async def api_ticket_deactivate(
+ ticket_id: str, wallet: WalletTypeInfo = Depends(require_admin_key)
+) -> Ticket:
ticket = await get_ticket(ticket_id)
if not ticket:
- logger.warning(f"SatsPay webhook: ticket {ticket_id} does not exist.")
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
- if ticket.paid:
- logger.warning(f"SatsPay webhook: ticket {ticket_id} already paid.")
- return
- if not ticket.extra.satspay_charge_id:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST, detail="Not a SatsPay ticket."
- )
- wallet = await get_wallet(ticket.wallet)
- if not wallet:
- raise HTTPException(
- status_code=HTTPStatus.NOT_FOUND, detail="Ticket wallet does not exist."
- )
- try:
- charge = await get_satspay_charge(wallet.inkey, ticket.extra.satspay_charge_id)
- except Exception as exc:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST, detail=f"Could not verify charge: {exc}"
- ) from exc
- if not charge.get("paid"):
- logger.warning(
- f"SatsPay webhook for ticket {ticket_id}: charge"
- f" {ticket.extra.satspay_charge_id} not paid."
- )
- return
-
- ticket = await set_ticket_paid(ticket)
- send_ticket_notification_in_background(ticket)
- for queue in payment_listeners.get(ticket_id, []):
- queue.put_nowait(ticket)
+ if ticket.wallet != wallet.wallet.id:
+ raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your ticket.")
+ return await toggle_ticket_deactivation(ticket)
-@tickets_api_router.put("/{payment_hash}/onchain-confirm")
-async def api_ticket_onchain_confirm(
- payment_hash: str,
- wallet: WalletTypeInfo = Depends(require_admin_key),
-) -> Ticket:
- ticket = await get_ticket(payment_hash)
+@events_api_router.put("/tickets/register/{ticket_id}")
+async def api_event_register_ticket(ticket_id: str) -> Ticket:
+ ticket = await get_ticket(ticket_id)
if not ticket:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
- if ticket.wallet != wallet.wallet.id:
- raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your ticket.")
- if ticket.paid:
+ if not ticket.paid:
raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST, detail="Ticket already paid."
+ status_code=HTTPStatus.FORBIDDEN, detail="Ticket not paid for."
)
- if not ticket.extra.onchain:
+ if ticket.extra.deactivated:
raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST,
- detail="Ticket is not an onchain payment.",
+ status_code=HTTPStatus.FORBIDDEN, detail="Ticket has been deactivated."
+ )
+ if ticket.registered:
+ raise HTTPException(
+ status_code=HTTPStatus.FORBIDDEN, detail="Ticket already registered."
)
- ticket = await set_ticket_paid(ticket)
- send_ticket_notification_in_background(ticket)
- for queue in payment_listeners.get(payment_hash, []):
- queue.put_nowait(ticket)
+ ticket.registered = True
+ ticket.reg_timestamp = datetime.now(timezone.utc)
+ ticket = await update_ticket(ticket)
return ticket
-@tickets_api_router.post("/{ticket_id}/resend-email", response_model=TicketResendResult)
+@events_api_router.post("/tickets/{ticket_id}/resend-email")
async def api_ticket_resend_email(
ticket_id: str,
request: Request,
wallet: WalletTypeInfo = Depends(require_admin_key),
-) -> TicketResendResult:
+):
ticket = await get_ticket(ticket_id)
if not ticket:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
-
if ticket.wallet != wallet.wallet.id:
raise HTTPException(status_code=HTTPStatus.FORBIDDEN, detail="Not your ticket.")
-
if not ticket.paid:
raise HTTPException(
status_code=HTTPStatus.FORBIDDEN,
detail="Only paid tickets can be resent by email.",
)
-
try:
- return await resend_ticket_email_notification(
+ result = await resend_ticket_email_notification(
ticket, str(request.base_url).rstrip("/")
)
+ return result.dict()
except ValueError as exc:
- raise HTTPException(
- status_code=HTTPStatus.BAD_REQUEST, detail=str(exc)
- ) from exc
+ raise HTTPException(status_code=HTTPStatus.BAD_REQUEST, detail=str(exc))
-@tickets_api_router.put("/register/{ticket_id}")
-async def api_event_register_ticket(ticket_id) -> Ticket:
+@qr_api_router.get("/qr/{ticket_id}", response_class=StreamingResponse)
+async def api_ticket_qr(ticket_id: str):
ticket = await get_ticket(ticket_id)
-
if not ticket:
raise HTTPException(
status_code=HTTPStatus.NOT_FOUND, detail="Ticket does not exist."
)
- if not ticket.paid:
+ event = await get_event(ticket.event)
+ if not event:
raise HTTPException(
- status_code=HTTPStatus.FORBIDDEN, detail="Ticket not paid for."
+ status_code=HTTPStatus.NOT_FOUND, detail="Event does not exist."
)
- if ticket.registered is True:
- raise HTTPException(
- status_code=HTTPStatus.FORBIDDEN, detail="Ticket already registered"
- )
+ qr_img = _make_qr_png(f"ticket://{ticket_id}", size=157)
+ output = BytesIO()
+ qr_img.save(output, format="PNG")
+ output.seek(0)
- ticket.registered = True
- ticket.reg_timestamp = datetime.now(timezone.utc)
- ticket = await update_ticket(ticket)
- return ticket
+ return StreamingResponse(
+ output,
+ media_type="image/png",
+ headers={
+ "Cache-Control": "no-cache, no-store, must-revalidate",
+ "Pragma": "no-cache",
+ "Expires": "0",
+ },
+ )