diff --git a/buzz/tests/factories/ticketing/event_payment_factory.py b/buzz/tests/factories/ticketing/event_payment_factory.py new file mode 100644 index 00000000..96a0f676 --- /dev/null +++ b/buzz/tests/factories/ticketing/event_payment_factory.py @@ -0,0 +1,24 @@ +from typing import Any + +import frappe +from frappe_factory_bot.frappe_factory_bot.base_factory import BaseFactory + +from buzz.ticketing.doctype.event_payment.event_payment import EventPayment + + +class EventPaymentFactory(BaseFactory[EventPayment]): + """An unpaid payment. Pass `reference_doctype`, `reference_docname` and `payment_gateway`.""" + + doctype = "Event Payment" + + @property + def default_attributes(self) -> dict[str, Any]: + from buzz.tests.factories import UserFactory + from buzz.tests.factories.ticketing.event_booking_factory import LINK_BOOKER + + return {"user": self.overrides.get("user") or UserFactory.create_once(LINK_BOOKER).name} + + @property + def received(self) -> dict[str, Any]: + # Webhooks find the payment by its id, so each one is unique. + return {"payment_received": 1, "payment_id": f"pay_{frappe.generate_hash(length=10)}"} diff --git a/buzz/tests/factories/ticketing/integration_request_factory.py b/buzz/tests/factories/ticketing/integration_request_factory.py new file mode 100644 index 00000000..39eaaefe --- /dev/null +++ b/buzz/tests/factories/ticketing/integration_request_factory.py @@ -0,0 +1,14 @@ +from typing import Any + +from frappe.integrations.doctype.integration_request.integration_request import IntegrationRequest +from frappe_factory_bot.frappe_factory_bot.base_factory import BaseFactory + + +class IntegrationRequestFactory(BaseFactory[IntegrationRequest]): + """A queued Razorpay webhook. Pass the payload as JSON in `data`.""" + + doctype = "Integration Request" + + @property + def default_attributes(self) -> dict[str, Any]: + return {"integration_request_service": "Razorpay", "is_remote_request": 1, "status": "Queued"} diff --git a/buzz/ticketing/doctype/event_booking/refund_test_case.py b/buzz/ticketing/doctype/event_booking/refund_test_case.py new file mode 100644 index 00000000..2f622808 --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/refund_test_case.py @@ -0,0 +1,130 @@ +from unittest.mock import MagicMock, patch + +import frappe +from frappe.tests import IntegrationTestCase + +from buzz.payments import handle_refund_notification +from buzz.tests.factories import ( + BuzzEventFactory, + EventBookingFactory, + EventTicketTypeFactory, + PaymentGatewayFactory, +) +from buzz.tests.factories.ticketing.event_payment_factory import EventPaymentFactory +from buzz.tests.factories.ticketing.integration_request_factory import IntegrationRequestFactory +from buzz.ticketing.doctype.event_booking.event_booking import RAZORPAY + +TICKET_PRICE = 500 +# 500 per ticket with 10% tax on top. +CHARGED_PER_TICKET = 550.0 +GATEWAY_CONTROLLER = "buzz.ticketing.doctype.event_booking.event_booking.get_controller" + + +class BookingRefundTestCase(IntegrationTestCase): + """A paid booking of one ticket per price in `ticket_prices`, on an event with 10% tax.""" + + ticket_prices = (TICKET_PRICE, TICKET_PRICE) + attendee_names = ("John", "Jenny") + + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = cls.create_event() + cls.ticket_types = [ + EventTicketTypeFactory.create(event=cls.event, prices=[{"currency": "INR", "price": price}]).name + for price in cls.ticket_prices + ] + cls.gateway = razorpay_gateway() + + @classmethod + def create_event(cls) -> str: + return BuzzEventFactory.create("with_tax", tax_percentage=10).name + + def setUp(self): + self.enterContext(self.set_user("Administrator")) + self.booking = self.create_paid_booking(self.attendee_names, self.ticket_types) + + def create_paid_booking(self, names, ticket_types): + attendees = [ + {"ticket_type": ticket_type, "first_name": name, "email": f"{name.lower()}@example.com"} + for name, ticket_type in zip(names, ticket_types, strict=True) + ] + booking = EventBookingFactory.create(event=self.event, attendees=attendees, payment_status="Paid") + booking.submit() + return booking + + def make_payment(self, gateway: str | None = None): + self.payment = EventPaymentFactory.create( + "received", + user=self.booking.user, + amount=self.booking.total_amount, + currency=self.booking.currency, + reference_doctype="Event Booking", + reference_docname=self.booking.name, + payment_gateway=gateway or self.gateway, + ) + + def initiate_refund(self, amount: float, tickets: list[str] | None = None, refund_id: str | None = None): + client = MagicMock() + client.refund_payment.return_value = { + "id": refund_id or self.refund_id(), + "status": "pending", + "amount": int(amount * 100), + } + with patch(GATEWAY_CONTROLLER, return_value=client): + self.booking.refund(amount=amount, tickets=tickets) + self.booking.reload() + return client + + def send_refund_webhook(self, refund_id: str, status: str, amount: float) -> str: + log = self.create_refund_log(self.refund_payload(refund_id, status, amount)) + self.handle_refund_log(log) + return log + + def refund_payload(self, refund_id: str, status: str, amount: float) -> dict: + entity = { + "id": refund_id, + "status": status, + "amount": int(amount * 100), + "payment_id": self.payment.payment_id, + } + return {"event": f"refund.{status}", "payload": {"refund": {"entity": entity}}} + + def create_refund_log(self, payload: dict) -> str: + return IntegrationRequestFactory.create(data=frappe.as_json(payload)).name + + def handle_refund_log(self, log: str) -> None: + # Cancelling a ticket emails the attendee, and delivery is not under test. + with patch("frappe.sendmail"): + handle_refund_notification("Integration Request", log) + self.booking.reload() + + def refund_id(self, suffix: str = "1") -> str: + """Refund ids are unique in the database, and rollback is per class.""" + return f"rfnd_{self._testMethodName}_{suffix}" + + def refunds(self) -> list: + names = frappe.get_all( + "Event Booking Refund", + filters={"booking": self.booking.name}, + order_by="creation asc", + pluck="name", + ) + return [frappe.get_doc("Event Booking Refund", name) for name in names] + + def refundable_tickets(self) -> list[str]: + return [ticket["ticket"] for ticket in self.booking.get_refund_summary()["tickets"]] + + def cancellation_requests(self) -> list: + return frappe.get_all( + "Ticket Cancellation Request", + filters={"booking": self.booking.name}, + fields=["name", "docstatus"], + ) + + +def razorpay_gateway() -> str: + """Refunds only go through a gateway named Razorpay, so this name is fixed.""" + if not frappe.db.exists("Payment Gateway", RAZORPAY): + PaymentGatewayFactory.create(gateway=RAZORPAY) + return RAZORPAY diff --git a/buzz/ticketing/doctype/event_booking/test_booking_confirmation.py b/buzz/ticketing/doctype/event_booking/test_booking_confirmation.py new file mode 100644 index 00000000..3edb72b5 --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/test_booking_confirmation.py @@ -0,0 +1,66 @@ +import frappe + +from buzz.api.booking import get_booking_confirmation, process_booking +from buzz.api.booking.services import get_booking_access_token, verify_booking_access_token +from buzz.tests.base_test_cases import BookingTestCase +from buzz.tests.factories import EventBookingFactory, EventTicketTypeFactory + + +class TestBookingConfirmation(BookingTestCase): + def test_access_token_roundtrip(self): + token = get_booking_access_token("B-TEST-001") + + self.assertTrue(verify_booking_access_token("B-TEST-001", token)) + for wrong_token in ("deadbeef", "", None): + self.assertFalse(verify_booking_access_token("B-TEST-001", wrong_token)) + self.assertFalse(verify_booking_access_token("B-OTHER-002", token)) + + def test_free_booking_redirects_to_the_token_gated_success_page(self): + payload = process_booking(self.booking_request()) + + path, token = payload.redirect_to.split("?token=") + self.assertEqual(path, f"/booking-success/{payload.booking_name}") + self.assertTrue(verify_booking_access_token(payload.booking_name, token)) + + def test_free_booking_confirms_on_submit(self): + booking = self.create_booking(self.free_ticket_type.name) + + booking.submit() + + self.assertEqual( + (booking.status, booking.payment_status, booking.total_amount), ("Confirmed", "Paid", 0) + ) + + def test_guest_with_a_valid_token_sees_the_booking(self): + booking = self.create_submitted_booking() + token = get_booking_access_token(booking.name) + + with self.set_user("Guest"): + result = get_booking_confirmation(booking.name, token=token) + + self.assertEqual(result.booking.name, booking.name) + self.assertEqual([ticket.attendee_name for ticket in result.tickets], ["Conf"]) + self.assertEqual(result.event.title, self.event.title) + + def test_guest_with_a_bad_token_is_refused(self): + booking = self.create_submitted_booking() + + with self.set_user("Guest"): + for token in ("wrong-token", ""): + with self.subTest(token=token), self.assertRaises(frappe.PermissionError): + get_booking_confirmation(booking.name, token=token) + + def test_a_user_with_access_needs_no_token(self): + booking = self.create_submitted_booking() + + self.assertEqual(get_booking_confirmation(booking.name).booking.name, booking.name) + + def create_submitted_booking(self): + paid_ticket_type = EventTicketTypeFactory.create("paid", event=self.event.name).name + booking = self.create_booking(paid_ticket_type) + booking.submit() + return booking + + def create_booking(self, ticket_type: str): + attendees = [{"ticket_type": ticket_type, "first_name": "Conf", "email": "conf@example.com"}] + return EventBookingFactory.create(event=self.event.name, attendees=attendees) diff --git a/buzz/ticketing/doctype/event_booking/test_booking_emails.py b/buzz/ticketing/doctype/event_booking/test_booking_emails.py new file mode 100644 index 00000000..80bf379a --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/test_booking_emails.py @@ -0,0 +1,197 @@ +from unittest.mock import patch + +import frappe + +from buzz.tests.base_test_cases import BookingTestCase +from buzz.tests.factories import ( + BuzzTeamFactory, + EmailTemplateFactory, + EventBookingFactory, + EventTicketTypeFactory, + UserFactory, +) + +OFFLINE_METHOD = "Bank Transfer" + + +class BookingEmailTestCase(BookingTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.email_booker = UserFactory.create_once("booking-email-booker@example.com").name + cls.ticket_type = EventTicketTypeFactory.create("paid", event=cls.event.name, title="Email Ticket") + + def setUp(self): + super().setUp() + # Ticket emails stay off, so `frappe.sendmail` only sees the booking email. + self.set_event( + { + "send_ticket_email": 0, + "send_booking_confirmation_email": 1, + "booking_confirmation_email_template": None, + "offline_acknowledgement_email_template": None, + } + ) + self.set_team_template(None) + + def create_booking(self, user: str, **overrides): + attendees = [ + {"ticket_type": self.ticket_type.name, "first_name": "John", "email": "john@example.com"} + ] + return EventBookingFactory.create(event=self.event.name, user=user, attendees=attendees, **overrides) + + def create_template(self, subject_prefix: str) -> str: + return EmailTemplateFactory.create(subject=f"{subject_prefix} - {{{{ event_title }}}}").name + + def set_team_template(self, template: str | None): + BuzzTeamFactory.set_settings( + self.event.team, {"default_booking_confirmation_email_template": template} + ) + + +@patch("frappe.sendmail") +class TestBookingConfirmationEmail(BookingEmailTestCase): + def test_sends_confirmation_to_booker(self, sendmail): + booking = self.submit_booking() + + sendmail.assert_called_once() + self.assertIn(self.email_booker, sendmail.call_args[1]["recipients"]) + self.assertEqual(sendmail.call_args[1]["reference_doctype"], "Event Booking") + self.assertEqual(sendmail.call_args[1]["reference_name"], booking.name) + + def test_uses_inline_template_when_none_configured(self, sendmail): + self.submit_booking() + + self.assertEqual(sendmail.call_args[1]["template"], "booking_confirmation") + + def test_skips_administrator(self, sendmail): + self.submit_booking("Administrator") + + sendmail.assert_not_called() + + def test_skips_guest(self, sendmail): + self.submit_booking("Guest") + + sendmail.assert_not_called() + + def test_respects_event_toggle_off(self, sendmail): + self.set_event({"send_booking_confirmation_email": 0}) + + self.submit_booking() + + sendmail.assert_not_called() + + def test_uses_event_template_when_set(self, sendmail): + self.set_event({"booking_confirmation_email_template": self.create_template("EVENT")}) + + self.submit_booking() + + self.assertIn("EVENT", sendmail.call_args[1]["subject"]) + + def test_falls_back_to_the_teams_default_template(self, sendmail): + self.set_team_template(self.create_template("TEAM")) + + self.submit_booking() + + self.assertIn("TEAM", sendmail.call_args[1]["subject"]) + + def test_support_email_comes_from_the_teams_settings(self, sendmail): + BuzzTeamFactory.set_settings(self.event.team, {"support_email": "booking-support@example.com"}) + + self.submit_booking() + + self.assertEqual(sendmail.call_args[1]["args"]["support_email"], "booking-support@example.com") + + def test_event_template_takes_precedence_over_the_team_default(self, sendmail): + self.set_event({"booking_confirmation_email_template": self.create_template("EVENT")}) + self.set_team_template(self.create_template("TEAM")) + + self.submit_booking() + + sendmail.assert_called_once() + self.assertIn("EVENT", sendmail.call_args[1]["subject"]) + self.assertNotIn("TEAM", sendmail.call_args[1]["subject"]) + + def submit_booking(self, user: str | None = None): + booking = self.create_booking(user or self.email_booker) + booking.submit() + return booking + + +@patch("frappe.sendmail") +class TestOfflineAcknowledgementEmail(BookingEmailTestCase): + def test_sends_acknowledgement_to_booker(self, sendmail): + booking = self.send_acknowledgement() + + sendmail.assert_called_once() + self.assertIn(self.email_booker, sendmail.call_args[1]["recipients"]) + self.assertEqual(sendmail.call_args[1]["reference_doctype"], "Event Booking") + self.assertEqual(sendmail.call_args[1]["reference_name"], booking.name) + + def test_uses_inline_template_when_none_configured(self, sendmail): + self.send_acknowledgement() + + self.assertEqual(sendmail.call_args[1]["template"], "offline_booking_acknowledgement") + + def test_carries_the_booking_summary(self, sendmail): + booking = self.send_acknowledgement() + + args = sendmail.call_args[1]["args"] + self.assertEqual(args["doc"].name, booking.name) + self.assertEqual(args["event_title"], self.event.title) + # Ticket types autoname to integers and arrive off the row as strings. + self.assertEqual([row["ticket_type_title"] for row in args["attendee_rows"]], ["Email Ticket"]) + + def test_builtin_template_renders(self, sendmail): + # Every other test mocks the send, so only this one catches a broken Jinja tag. + booking = self.create_offline_booking(self.email_booker) + + html = frappe.render_template( + "templates/emails/offline_booking_acknowledgement.html", booking.get_booking_email_args() + ) + + for text in ("Payment verification pending", booking.name, OFFLINE_METHOD, "Email Ticket"): + self.assertIn(text, html) + + def test_uses_event_template_when_set(self, sendmail): + self.set_event({"offline_acknowledgement_email_template": self.create_template("OFFLINE")}) + + self.send_acknowledgement() + + self.assertIn("OFFLINE", sendmail.call_args[1]["subject"]) + + def test_ignores_the_confirmation_template(self, sendmail): + self.set_event({"booking_confirmation_email_template": self.create_template("EVENT")}) + self.set_team_template(self.create_template("TEAM")) + + self.send_acknowledgement() + + self.assertEqual(sendmail.call_args[1]["template"], "offline_booking_acknowledgement") + + def test_respects_event_toggle_off(self, sendmail): + self.set_event({"send_booking_confirmation_email": 0}) + + self.send_acknowledgement() + + sendmail.assert_not_called() + + def test_skips_system_users(self, sendmail): + for user in ("Administrator", "Guest"): + with self.subTest(user=user): + self.send_acknowledgement(user) + + sendmail.assert_not_called() + + def create_offline_booking(self, user: str): + return self.create_booking( + user, + payment_method="Offline", + offline_payment_method=OFFLINE_METHOD, + status="Approval Pending", + payment_status="Verification Pending", + ) + + def send_acknowledgement(self, user: str | None = None): + booking = self.create_offline_booking(user or self.email_booker) + booking.send_offline_acknowledgement_email() + return booking diff --git a/buzz/ticketing/doctype/event_booking/test_booking_totals.py b/buzz/ticketing/doctype/event_booking/test_booking_totals.py new file mode 100644 index 00000000..daec2107 --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/test_booking_totals.py @@ -0,0 +1,60 @@ +from frappe.tests import IntegrationTestCase + +from buzz.api.booking.services import create_add_on_doc +from buzz.tests.factories import ( + BuzzEventFactory, + EventBookingFactory, + EventTicketTypeFactory, + TicketAddOnFactory, +) + +TICKET_PRICE = 500 +ADD_ON_PRICE = 100 + + +class TestBookingTotals(IntegrationTestCase): + def test_total_without_taxes_includes_add_ons(self): + event = BuzzEventFactory.create().name + booking = self.create_booking(event, TICKET_PRICE, attendees=2) + self.assertEqual(booking.total_amount, 1000) + + add_on = TicketAddOnFactory.create(event=event, price=ADD_ON_PRICE).name + booking.attendees[0].add_ons = create_add_on_doc("John", [{"add_on": add_on, "value": "XL"}]).name + booking.save() + + self.assertEqual(booking.attendees[0].number_of_add_ons, 1) + self.assertEqual(booking.attendees[0].add_on_total, ADD_ON_PRICE) + self.assertEqual((booking.net_amount, booking.total_amount), (1100, 1100)) + + def test_exclusive_tax_is_added_on_top(self): + event = BuzzEventFactory.create("with_tax", tax_label="GST", tax_percentage=18).name + + booking = self.create_booking(event, TICKET_PRICE, attendees=2) + + self.assertEqual((booking.tax_label, booking.tax_percentage), ("GST", 18)) + self.assertEqual((booking.net_amount, booking.tax_amount, booking.total_amount), (1000, 180, 1180)) + + def test_custom_tax_label(self): + event = BuzzEventFactory.create("with_tax", tax_label="VAT", tax_percentage=20).name + + booking = self.create_booking(event, 100) + + self.assertEqual((booking.tax_label, booking.tax_percentage), ("VAT", 20)) + self.assertEqual((booking.tax_amount, booking.total_amount), (20, 120)) + + def test_inclusive_tax_is_back_calculated_without_raising_the_total(self): + event = BuzzEventFactory.create("with_tax", tax_inclusive=1, tax_percentage=18).name + + booking = self.create_booking(event, TICKET_PRICE, attendees=2) + + self.assertEqual(booking.net_amount, 1000) + self.assertAlmostEqual(booking.tax_amount, round(1000 * 18 / 118, 2), places=2) + self.assertEqual(booking.total_amount, 1000) + + def create_booking(self, event: str, price: float, attendees: int = 1): + ticket_type = EventTicketTypeFactory.create(event=event, prices=[{"currency": "INR", "price": price}]) + rows = [ + {"ticket_type": ticket_type.name, "first_name": name, "email": f"{name.lower()}@example.com"} + for name in ("John", "Jenny")[:attendees] + ] + return EventBookingFactory.create(event=event, attendees=rows) diff --git a/buzz/ticketing/doctype/event_booking/test_event_booking.py b/buzz/ticketing/doctype/event_booking/test_event_booking.py index feae4165..c9a98c2e 100644 --- a/buzz/ticketing/doctype/event_booking/test_event_booking.py +++ b/buzz/ticketing/doctype/event_booking/test_event_booking.py @@ -1,1587 +1,74 @@ # Copyright (c) 2025, BWH Studios and Contributors # See license.txt -from unittest.mock import patch - import frappe -from frappe.tests import IntegrationTestCase - -from buzz.api.booking import get_booking_confirmation as get_booking_confirmation_endpoint -from buzz.api.booking import process_booking as process_booking_endpoint -from buzz.api.booking.schemas import BookingRequest -from buzz.events.doctype.buzz_team_settings.test_buzz_team_settings import set_team_settings - - -def process_booking(**kwargs): - return process_booking_endpoint(BookingRequest(**kwargs)).__json__() - - -def get_booking_confirmation(booking_id, token=None): - return get_booking_confirmation_endpoint(booking_id, token=token).__json__() - - -# On IntegrationTestCase, the doctype test records and all -# link-field test record dependencies are recursively loaded -# Use these module variables to add/remove to/from that list -EXTRA_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"] -IGNORE_TEST_RECORD_DEPENDENCIES = [] # eg. ["User"] - -TEST_ADD_ON_PRICE = 100 -TEST_VIP_TICKET_TYPE_PRICE = 500 - - -class IntegrationTestEventBooking(IntegrationTestCase): - """ - Integration tests for EventBooking. - Use this class for testing interactions between multiple components. - """ - - def test_total_calculation_without_taxes(self): - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - # Disable tax at event level - test_event.apply_tax = False - test_event.save() - - test_ticket_add_on = frappe.get_doc( - { - "doctype": "Ticket Add-on", - "event": test_event.name, - "title": "T-Shirt", - "price": TEST_ADD_ON_PRICE, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "VIP", - "prices": [{"currency": "INR", "price": TEST_VIP_TICKET_TYPE_PRICE}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - {"ticket_type": test_ticket_type.name, "first_name": "Jenny", "email": "jenny@email.com"}, - ], - } - ).insert() - - # without add ons - self.assertEqual(test_booking.total_amount, 1000) - - test_attendee_add_on = frappe.get_doc( - { - "doctype": "Attendee Ticket Add-on", - "add_ons": [{"add_on": test_ticket_add_on.name, "value": "XL"}], - } - ).insert() - - test_booking.attendees[0].add_ons = test_attendee_add_on.name - test_booking.save() - - # with one add-on - self.assertEqual(test_booking.attendees[0].number_of_add_ons, 1) - self.assertEqual(test_booking.attendees[0].add_on_total, TEST_ADD_ON_PRICE) - self.assertEqual(test_booking.net_amount, 1100) - self.assertEqual(test_booking.total_amount, 1100) - - def test_total_calculation_with_taxes(self): - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - # Enable tax at event level (exclusive) - test_event.apply_tax = True - test_event.tax_inclusive = False - test_event.tax_label = "GST" - test_event.tax_percentage = 18 - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "VIP", - "prices": [{"currency": "INR", "price": TEST_VIP_TICKET_TYPE_PRICE}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - {"ticket_type": test_ticket_type.name, "first_name": "Jenny", "email": "jenny@email.com"}, - ], - } - ).insert() - - self.assertEqual(test_booking.net_amount, 1000) - self.assertEqual(test_booking.tax_label, "GST") - self.assertEqual(test_booking.tax_percentage, 18) - self.assertEqual(test_booking.tax_amount, 180) - self.assertEqual(test_booking.total_amount, 1180) - - def test_total_calculation_with_custom_tax_label(self): - """Test that custom tax labels (e.g., VAT) work correctly.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - # Enable tax with custom VAT label (exclusive) - test_event.apply_tax = True - test_event.tax_inclusive = False - test_event.tax_label = "VAT" - test_event.tax_percentage = 20 - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Standard", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - ], - } - ).insert() - - self.assertEqual(test_booking.tax_label, "VAT") - self.assertEqual(test_booking.tax_percentage, 20) - self.assertEqual(test_booking.tax_amount, 20) - self.assertEqual(test_booking.total_amount, 120) - - def test_tax_inclusive_calculation(self): - """Test that tax-inclusive prices back-calculate tax without increasing total.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - # Enable tax-inclusive pricing - test_event.apply_tax = True - test_event.tax_inclusive = True - test_event.tax_label = "GST" - test_event.tax_percentage = 18 - test_event.save() +from buzz.api.booking import process_booking +from buzz.tests.base_test_cases import BookingTestCase +from buzz.tests.factories import EventBookingFactory, EventTicketFactory, EventTicketTypeFactory - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "VIP", - "prices": [{"currency": "INR", "price": TEST_VIP_TICKET_TYPE_PRICE}], - } - ).insert() +UTM_PARAMETERS = { + "utm_source": "facebook", + "utm_medium": "social", + "utm_campaign": "winter_promo", + "utm_content": "banner_ad", + "utm_term": "event tickets", +} - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - {"ticket_type": test_ticket_type.name, "first_name": "Jenny", "email": "jenny@email.com"}, - ], - } - ).insert() - - # net_amount = 2 * 500 = 1000 - self.assertEqual(test_booking.net_amount, 1000) - self.assertEqual(test_booking.tax_label, "GST") - self.assertEqual(test_booking.tax_percentage, 18) - # tax_amount = 1000 * 18 / 118 = 152.54 (rounded to 2 decimals) - self.assertAlmostEqual(test_booking.tax_amount, round(1000 * 18 / 118, 2), places=2) - # total stays the same — tax is included in the price - self.assertEqual(test_booking.total_amount, 1000) - - def test_tax_inclusive_with_single_ticket(self): - """Test tax-inclusive with a single ticket at a round price.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_event.apply_tax = True - test_event.tax_inclusive = True - test_event.tax_label = "VAT" - test_event.tax_percentage = 20 - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Standard", - "prices": [{"currency": "INR", "price": 120}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - ], - } - ).insert() - - # Price is 120 inclusive of 20% VAT - # tax_amount = 120 * 20 / 120 = 20 - self.assertEqual(test_booking.net_amount, 120) - self.assertEqual(test_booking.tax_amount, 20) - self.assertEqual(test_booking.total_amount, 120) - - def test_tax_exclusive_still_works(self): - """Ensure that when tax_inclusive is False, tax is still added on top.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_event.apply_tax = True - test_event.tax_inclusive = False - test_event.tax_label = "GST" - test_event.tax_percentage = 18 - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "VIP", - "prices": [{"currency": "INR", "price": TEST_VIP_TICKET_TYPE_PRICE}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - {"ticket_type": test_ticket_type.name, "first_name": "Jenny", "email": "jenny@email.com"}, - ], - } - ).insert() - - # Exclusive: tax added on top - self.assertEqual(test_booking.net_amount, 1000) - self.assertEqual(test_booking.tax_amount, 180) - self.assertEqual(test_booking.total_amount, 1180) +class TestTicketAvailability(BookingTestCase): def test_prevents_booking_if_tickets_unavailable(self): - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_vip_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "VIP", - "prices": [{"currency": "INR", "price": 500}], - "is_published": True, - "max_tickets_available": 2, - } - ).insert() - - test_normal_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Normal", - "prices": [{"currency": "INR", "price": 500}], - "is_published": True, - } - ).insert() + vip = EventTicketTypeFactory.create(event=self.event.name, max_tickets_available=2).name + normal = self.free_ticket_type.name + EventTicketFactory.create("submitted", event=self.event.name, ticket_type=vip) + self.create_booking(vip, normal).submit() - # VIP Ticket 1 - frappe.get_doc( - { - "doctype": "Event Ticket", - "ticket_type": test_vip_ticket_type.name, - "attendee_name": "John Doe", - "attendee_email": "john@email.com", - } - ).insert().submit() - - # VIP Ticket 2 with Normal Ticket 1 - frappe.get_doc( - { - "doctype": "Event Booking", - "user": frappe.session.user, - "event": test_event.name, - "attendees": [ - { - "first_name": "John", - "last_name": "Doe", - "ticket_type": test_vip_ticket_type.name, - "email": "john@email.com", - }, - { - "first_name": "Jenny", - "last_name": "Doe", - "ticket_type": test_normal_ticket_type.name, - "email": "jenny@email.com", - }, - ], - } - ).insert().submit() - - # VIP Ticket 3 with Normal Ticket 2 with self.assertRaises(frappe.ValidationError): - frappe.get_doc( - { - "doctype": "Event Booking", - "user": frappe.session.user, - "event": test_event.name, - "attendees": [ - { - "first_name": "John", - "last_name": "Doe", - "ticket_type": test_vip_ticket_type.name, - "email": "john@email.com", - }, - { - "first_name": "John", - "last_name": "Doe", - "ticket_type": test_normal_ticket_type.name, - "email": "john@email.com", - }, - ], - } - ).insert() - - # Unpublish normal ticket type - test_normal_ticket_type.is_published = False - test_normal_ticket_type.save() + self.create_booking(vip, normal) - # Booking with unpublished ticket type + frappe.db.set_value("Event Ticket Type", normal, "is_published", 0) + frappe.clear_document_cache("Event Ticket Type", normal) with self.assertRaises(frappe.ValidationError): - frappe.get_doc( - { - "doctype": "Event Booking", - "user": frappe.session.user, - "event": test_event.name, - "attendees": [ - { - "first_name": "John", - "last_name": "Doe", - "ticket_type": test_normal_ticket_type.name, - "email": "john@email.com", - } - ], - } - ).insert() - - def test_utm_parameters_are_saved(self): - """Test that UTM parameters are correctly saved with bookings.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Standard", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - ], - "utm_parameters": [ - {"utm_name": "utm_source", "value": "google"}, - {"utm_name": "utm_medium", "value": "cpc"}, - {"utm_name": "utm_campaign", "value": "summer_sale"}, - ], - } - ).insert() - - self.assertEqual(len(test_booking.utm_parameters), 3) - self.assertEqual(test_booking.utm_parameters[0].utm_name, "utm_source") - self.assertEqual(test_booking.utm_parameters[0].value, "google") - self.assertEqual(test_booking.utm_parameters[1].utm_name, "utm_medium") - self.assertEqual(test_booking.utm_parameters[1].value, "cpc") - self.assertEqual(test_booking.utm_parameters[2].utm_name, "utm_campaign") - self.assertEqual(test_booking.utm_parameters[2].value, "summer_sale") - - def test_booking_without_utm_parameters(self): - """Test that bookings work correctly without UTM parameters.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Standard", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - ], - } - ).insert() - - self.assertEqual(len(test_booking.utm_parameters), 0) - - def test_custom_utm_parameters(self): - """Test that custom UTM parameters (beyond standard ones) are saved.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Standard", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - - test_booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "John", "email": "john@email.com"}, - ], - "utm_parameters": [ - {"utm_name": "utm_source", "value": "newsletter"}, - {"utm_name": "utm_custom_param", "value": "special_offer"}, - ], - } - ).insert() - - self.assertEqual(len(test_booking.utm_parameters), 2) - # Check custom utm parameter is saved - custom_param = next( - (p for p in test_booking.utm_parameters if p.utm_name == "utm_custom_param"), None - ) - self.assertIsNotNone(custom_param) - self.assertEqual(custom_param.value, "special_offer") - - -class TestProcessBookingAPI(IntegrationTestCase): - """Test the process_booking API endpoint for UTM parameter handling.""" - - def test_process_booking_with_utm_parameters(self): - """Test that process_booking API correctly saves UTM parameters.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "API Test Ticket", - "is_published": True, - } - ).insert() - - # Disable tax at event level - test_event.apply_tax = False - test_event.save() - - attendees = [ - { - "first_name": "API Test User", - "email": "apitest@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } - ] - - utm_parameters = [ - {"utm_name": "utm_source", "value": "facebook"}, - {"utm_name": "utm_medium", "value": "social"}, - {"utm_name": "utm_campaign", "value": "winter_promo"}, - {"utm_name": "utm_content", "value": "banner_ad"}, - {"utm_name": "utm_term", "value": "event tickets"}, - ] - - result = process_booking( - attendees=attendees, - event=str(test_event.name), - utm_parameters=utm_parameters, - ) - - # Verify booking was created - self.assertIn("booking_name", result) - - # Fetch the booking and verify UTM parameters - booking = frappe.get_doc("Event Booking", result["booking_name"]) - self.assertEqual(len(booking.utm_parameters), 5) - - # Verify each UTM parameter - utm_dict = {p.utm_name: p.value for p in booking.utm_parameters} - self.assertEqual(utm_dict["utm_source"], "facebook") - self.assertEqual(utm_dict["utm_medium"], "social") - self.assertEqual(utm_dict["utm_campaign"], "winter_promo") - self.assertEqual(utm_dict["utm_content"], "banner_ad") - self.assertEqual(utm_dict["utm_term"], "event tickets") - - # ==================== Offline Payment Tests ==================== - - def _cleanup_offline_methods(self, event_name): - """Remove all Offline Payment Method records for the given event.""" - frappe.db.delete("Offline Payment Method", {"event": event_name}) - - def test_offline_booking_cannot_be_submitted_directly(self): - """Test that offline bookings cannot be submitted directly — must use approve/reject.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Offline Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - "payment_method": "Offline", - } - ).insert() - - with self.assertRaises(frappe.ValidationError): - booking.submit() - - def test_approve_offline_booking(self): - """Test approving an offline booking submits it and generates tickets.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Approval Test Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - "payment_method": "Offline", - "status": "Approval Pending", - "payment_status": "Verification Pending", - } - ).insert() - - # Booking should be in draft with no tickets - self.assertEqual(booking.docstatus, 0) - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 0) - - booking.approve_booking() - booking.reload() - - # After approval, booking should be submitted with tickets generated - self.assertEqual(booking.docstatus, 1) - self.assertEqual(booking.status, "Approved") - self.assertEqual(booking.payment_status, "Paid") - - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 1) - - def test_reject_offline_booking(self): - """Test rejecting an offline booking keeps it in draft with no tickets.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Rejection Test Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - "payment_method": "Offline", - "status": "Approval Pending", - "payment_status": "Verification Pending", - } - ).insert() - - self.assertEqual(booking.docstatus, 0) - - booking.reject_booking() - booking.reload() - - # Discarded (docstatus=2) and marked as Rejected - self.assertEqual(booking.docstatus, 2) - self.assertEqual(booking.status, "Rejected") - - # No tickets should be generated - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 0) - - def test_offline_with_coupon_code(self): - """Test offline payment with coupon code discount.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Coupon Test Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - coupon = frappe.get_doc( - { - "doctype": "Buzz Coupon Code", - "code": f"OFFLINE10-{frappe.generate_hash(length=4)}", - "coupon_type": "Discount", - "discount_type": "Percentage", - "discount_value": 10, - "is_active": True, - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "coupon_code": coupon.name, - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - "payment_method": "Offline", - } - ).insert() - - self.assertEqual(booking.net_amount, 500) - self.assertEqual(booking.discount_amount, 50) - self.assertEqual(booking.total_amount, 450) - - def test_offline_with_tax(self): - """Test offline payment with tax calculation.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.apply_tax = True - test_event.tax_percentage = 18 - test_event.tax_label = "GST" - test_event.save() - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Tax Test Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - "payment_method": "Offline", - } - ).insert() - - self.assertEqual(booking.net_amount, 500) - self.assertEqual(booking.tax_percentage, 18) - self.assertEqual(booking.tax_amount, 90) - self.assertEqual(booking.total_amount, 590) - - def test_offline_booking_requires_payment_method_field(self): - """Test that offline booking requires payment_method field.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self._cleanup_offline_methods(test_event.name) - - frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Payment Method Test", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - # Booking without payment_method field should default to normal flow - booking_without_method = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test", "email": "test@test.com"} - ], - } - ).insert() - - booking_without_method.submit() - # Without payment_method field, it should go to normal payment flow - self.assertEqual(booking_without_method.payment_status, "Unpaid") - - # Booking with payment_method = "Offline" should block direct submission - booking_with_method = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": test_ticket_type.name, "first_name": "Test2", "email": "test2@test.com"} - ], - "payment_method": "Offline", - } - ).insert() - - with self.assertRaises(frappe.ValidationError): - booking_with_method.submit() - - def test_process_booking_offline_stays_in_draft(self): - """Test that offline bookings via process_booking stay in draft with no tickets.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.apply_tax = False - test_event.is_published = True - test_event.save() - self._cleanup_offline_methods(test_event.name) - - offline_method = frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Offline Draft Test", - "prices": [{"currency": "INR", "price": 500}], - "is_published": True, - } - ).insert() - - result = process_booking( - attendees=[ - { - "first_name": "Offline User", - "email": "offline@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } - ], - event=str(test_event.name), - is_offline=True, - offline_payment_method=str(offline_method.name), - ) - - self.assertIn("booking_name", result) - self.assertTrue(result.get("offline_payment")) - - booking = frappe.get_doc("Event Booking", result["booking_name"]) - - # Booking must be in draft (not submitted) - self.assertEqual(booking.docstatus, 0) - self.assertEqual(booking.status, "Approval Pending") - self.assertEqual(booking.payment_status, "Verification Pending") - - # No tickets should exist - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 0) - - def test_process_booking_offline_generates_tickets_on_approval(self): - """Test that approving an offline booking created via API generates tickets.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.apply_tax = False - test_event.is_published = True - test_event.save() - self._cleanup_offline_methods(test_event.name) - - offline_method = frappe.get_doc( - { - "doctype": "Offline Payment Method", - "title": "Bank Transfer", - "event": test_event.name, - "enabled": 1, - } - ).insert() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Offline Approval Test", - "prices": [{"currency": "INR", "price": 500}], - "is_published": True, - } - ).insert() - - result = process_booking( - attendees=[ - { - "first_name": "Approval User", - "email": "approval@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } - ], - event=str(test_event.name), - is_offline=True, - offline_payment_method=str(offline_method.name), - ) - - booking = frappe.get_doc("Event Booking", result["booking_name"]) - - # No tickets before approval - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 0) - - # Approve the booking - booking.approve_booking() - booking.reload() - - # After approval: submitted, approved, tickets generated - self.assertEqual(booking.docstatus, 1) - self.assertEqual(booking.status, "Approved") - self.assertEqual(booking.payment_status, "Paid") - - tickets = frappe.db.get_all("Event Ticket", filters={"booking": booking.name}) - self.assertEqual(len(tickets), 1) - - def test_process_booking_free_event_returns_redirect_to(self): - """Free bookings (total_amount == 0) must get a redirect_to the new - token-gated booking-success screen, same as the paid-gateway branch.""" - from buzz.api.booking.services import verify_booking_access_token - - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.apply_tax = False - test_event.is_published = True - test_event.save() - - free_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Free Redirect Test Ticket", - "is_published": True, - } - ).insert() - - result = process_booking( - attendees=[ - { - "first_name": "Redirect Test User", - "email": "redirecttest@email.com", - "ticket_type": str(free_ticket_type.name), - "add_ons": [], - } - ], - event=str(test_event.name), - ) - - self.assertIn("booking_name", result) - self.assertIn("redirect_to", result) - self.assertTrue(result["redirect_to"].startswith(f"/booking-success/{result['booking_name']}?token=")) - - token = result["redirect_to"].split("token=")[1] - self.assertTrue(verify_booking_access_token(result["booking_name"], token)) - - def test_process_booking_without_utm_parameters(self): - """Test that process_booking API works without UTM parameters.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.is_published = True - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "API Test Ticket No UTM", - "is_published": True, - } - ).insert() - - # Disable tax at event level - test_event.apply_tax = False - test_event.save() + self.create_booking(normal) + def create_booking(self, *ticket_types: str): attendees = [ - { - "first_name": "No UTM User", - "email": "noutm@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } + {"first_name": "John", "ticket_type": ticket_type, "email": "john@example.com"} + for ticket_type in ticket_types ] + return EventBookingFactory.create(event=self.event.name, attendees=attendees) - result = process_booking( - attendees=attendees, - event=str(test_event.name), - utm_parameters=None, - ) - - self.assertIn("booking_name", result) - - booking = frappe.get_doc("Event Booking", result["booking_name"]) - self.assertEqual(len(booking.utm_parameters), 0) - - def test_process_booking_with_empty_utm_parameters(self): - """Test that process_booking API handles empty UTM list.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "API Test Ticket Empty UTM", - "is_published": True, - } - ).insert() - - # Disable tax at event level - test_event.apply_tax = False - test_event.is_published = True - test_event.save() - - attendees = [ - { - "first_name": "Empty UTM User", - "email": "emptyutm@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } - ] - - result = process_booking( - attendees=attendees, - event=str(test_event.name), - utm_parameters=[], - ) - - self.assertIn("booking_name", result) - - booking = frappe.get_doc("Event Booking", result["booking_name"]) - self.assertEqual(len(booking.utm_parameters), 0) - - def test_process_booking_failed_for_unpublished_event(self): - """Booking must fail when Buzz Event is not published.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.is_published = 0 - test_event.save() - - test_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Unpublished Test Ticket", - "is_published": True, - } - ).insert() - - attendees = [ - { - "first_name": "Failed User", - "email": "failed@email.com", - "ticket_type": str(test_ticket_type.name), - "add_ons": [], - } - ] - - with self.assertRaises(frappe.ValidationError) as ctx: - process_booking( - attendees=attendees, - event=str(test_event.name), - ) - - self.assertIn("Event is not live", str(ctx.exception)) - - def test_free_event_booking_auto_confirms(self): - """Test that free event bookings (total_amount = 0) automatically confirm on submit.""" - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - - free_ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Free Ticket", - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": frappe.session.user, - "attendees": [ - { - "first_name": "Free User", - "email": "free@email.com", - "ticket_type": free_ticket_type.name, - } - ], - } - ).insert() - - booking.submit() - - self.assertEqual(booking.status, "Confirmed", "Free booking should auto-confirm") - self.assertEqual(booking.payment_status, "Paid", "Free booking should be marked as Paid") - self.assertEqual(booking.total_amount, 0, "Free booking should have zero total") - - -class TestBookingConfirmation(IntegrationTestCase): - """Test token-gated guest booking confirmation (issue #167).""" - - def _make_submitted_booking(self): - test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - test_event.apply_tax = False - test_event.save() - - ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": test_event.name, - "title": "Confirmation Test Ticket", - "prices": [{"currency": "INR", "price": 500}], - } - ).insert() - - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": test_event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": ticket_type.name, "first_name": "Conf", "email": "conf@email.com"} - ], - } - ).insert() - booking.submit() - return booking - - def test_access_token_roundtrip(self): - from buzz.api.booking.services import get_booking_access_token, verify_booking_access_token - - token = get_booking_access_token("B-TEST-001") - self.assertTrue(verify_booking_access_token("B-TEST-001", token)) - # wrong token rejected - self.assertFalse(verify_booking_access_token("B-TEST-001", "deadbeef")) - # empty token rejected - self.assertFalse(verify_booking_access_token("B-TEST-001", "")) - self.assertFalse(verify_booking_access_token("B-TEST-001", None)) - # token is booking-specific - self.assertFalse(verify_booking_access_token("B-OTHER-002", token)) - - def test_get_booking_confirmation_valid_token_as_guest(self): - from buzz.api.booking.services import get_booking_access_token - - booking = self._make_submitted_booking() - token = get_booking_access_token(booking.name) - - frappe.set_user("Guest") - try: - result = get_booking_confirmation(booking.name, token=token) - finally: - frappe.set_user("Administrator") - - self.assertEqual(result["booking"]["name"], booking.name) - self.assertEqual(len(result["tickets"]), 1) - self.assertEqual(result["tickets"][0]["attendee_name"], "Conf") - self.assertTrue(result["event"]["title"]) - - def test_get_booking_confirmation_bad_token_as_guest_raises(self): - booking = self._make_submitted_booking() - - frappe.set_user("Guest") - try: - with self.assertRaises(frappe.PermissionError): - get_booking_confirmation(booking.name, token="wrong-token") - with self.assertRaises(frappe.PermissionError): - get_booking_confirmation(booking.name, token="") - finally: - frappe.set_user("Administrator") - - def test_get_booking_confirmation_owner_without_token(self): - booking = self._make_submitted_booking() - - # Administrator owns/has perm — no token needed - result = get_booking_confirmation(booking.name) - self.assertEqual(result["booking"]["name"], booking.name) - - -class TestBookingConfirmationEmail(IntegrationTestCase): - """Confirmation email sent to the booker with a booking summary (issue #56).""" - - BOOKER_EMAIL = "booker-56@example.com" - def setUp(self): - # Configure the shared event and create fixtures per test (not in - # setUpClass) so every mutation lives inside the test's own transaction: - # nothing leaks into other test classes, and the booker User is - # guaranteed to exist when the booking's user link is validated. - self.test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.test_event.apply_tax = False - self.test_event.send_booking_confirmation_email = 1 - self.test_event.booking_confirmation_email_template = None - # Keep per-attendee ticket emails off so frappe.sendmail is only invoked - # by the booking confirmation logic. - self.test_event.send_ticket_email = 0 - self.test_event.save() +class TestProcessBooking(BookingTestCase): + def test_saves_utm_parameters(self): + utm_parameters = [{"utm_name": name, "value": value} for name, value in UTM_PARAMETERS.items()] - set_team_settings(self.test_event.team, default_booking_confirmation_email_template=None) - # The rollback restores the settings row but not its cached copy. - self.addCleanup(frappe.clear_document_cache, "Buzz Team Settings", self.test_event.team) + payload = process_booking(self.booking_request(utm_parameters=utm_parameters)) - # A real (non-system) booker whose User email is a valid recipient. - if not frappe.db.exists("User", self.BOOKER_EMAIL): - frappe.get_doc( - { - "doctype": "User", - "email": self.BOOKER_EMAIL, - "first_name": "Booker", - "enabled": 1, - "user_type": "Website User", - "send_welcome_email": 0, - } - ).insert(ignore_permissions=True) + booking = frappe.get_doc("Event Booking", payload.booking_name) + self.assertEqual({row.utm_name: row.value for row in booking.utm_parameters}, UTM_PARAMETERS) - self.ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.test_event.name, - "title": "Confirmation Email Ticket", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() + def test_refuses_an_unpublished_event(self): + self.set_event({"is_published": 0}) - def tearDown(self): - frappe.delete_doc("Event Ticket Type", self.ticket_type.name, force=True) + with self.assertRaises(frappe.ValidationError) as raised: + process_booking(self.booking_request()) - def _make_booking(self, user): - return frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.test_event.name, - "user": user, - "attendees": [ - { - "ticket_type": self.ticket_type.name, - "first_name": "John", - "email": "john-56@example.com", - } - ], - } - ).insert() + self.assertIn("Event is not live", str(raised.exception)) - def _create_template(self, name, subject_prefix): - if frappe.db.exists("Email Template", name): - frappe.delete_doc("Email Template", name, force=True) - return frappe.get_doc( - { - "doctype": "Email Template", - "name": name, - "subject": f"{subject_prefix} - {{{{ event_title }}}}", - "response": f"
{subject_prefix} content
", - } - ).insert() - - @patch("frappe.sendmail") - def test_sends_confirmation_to_booker(self, mock_sendmail): - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_called_once() - self.assertIn(self.BOOKER_EMAIL, mock_sendmail.call_args[1]["recipients"]) - self.assertEqual(mock_sendmail.call_args[1]["reference_doctype"], "Event Booking") - self.assertEqual(mock_sendmail.call_args[1]["reference_name"], booking.name) - - @patch("frappe.sendmail") - def test_uses_inline_template_when_none_configured(self, mock_sendmail): - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_called_once() - self.assertEqual(mock_sendmail.call_args[1]["template"], "booking_confirmation") - - @patch("frappe.sendmail") - def test_skips_administrator(self, mock_sendmail): - booking = self._make_booking("Administrator") - booking.submit() - - mock_sendmail.assert_not_called() - - @patch("frappe.sendmail") - def test_skips_guest(self, mock_sendmail): - booking = self._make_booking("Guest") - booking.submit() - - mock_sendmail.assert_not_called() - - @patch("frappe.sendmail") - def test_respects_event_toggle_off(self, mock_sendmail): - self.test_event.send_booking_confirmation_email = 0 - self.test_event.save() - - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_not_called() - - @patch("frappe.sendmail") - def test_uses_event_template_when_set(self, mock_sendmail): - template = self._create_template("Booking Confirmation Event Template", "EVENT") - self.test_event.booking_confirmation_email_template = template.name - self.test_event.save() - - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_called_once() - self.assertIn("EVENT", mock_sendmail.call_args[1]["subject"]) - - @patch("frappe.sendmail") - def test_falls_back_to_the_teams_default_template(self, mock_sendmail): - template = self._create_template("Booking Confirmation Team Template", "TEAM") - set_team_settings(self.test_event.team, default_booking_confirmation_email_template=template.name) - - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_called_once() - self.assertIn("TEAM", mock_sendmail.call_args[1]["subject"]) - - @patch("frappe.sendmail") - def test_support_email_comes_from_the_teams_settings(self, mock_sendmail): - set_team_settings(self.test_event.team, support_email="booking-support@example.com") - - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - self.assertEqual(mock_sendmail.call_args[1]["args"]["support_email"], "booking-support@example.com") - - @patch("frappe.sendmail") - def test_event_template_takes_precedence_over_the_team_default(self, mock_sendmail): - event_template = self._create_template("Booking Event Precedence Template", "EVENT") - team_template = self._create_template("Booking Team Precedence Template", "TEAM") - self.test_event.booking_confirmation_email_template = event_template.name - self.test_event.save() - - set_team_settings( - self.test_event.team, default_booking_confirmation_email_template=team_template.name - ) - - booking = self._make_booking(self.BOOKER_EMAIL) - booking.submit() - - mock_sendmail.assert_called_once() - self.assertIn("EVENT", mock_sendmail.call_args[1]["subject"]) - self.assertNotIn("TEAM", mock_sendmail.call_args[1]["subject"]) - - -class TestOfflineAcknowledgementEmail(IntegrationTestCase): - """Acknowledgement sent when an offline booking is created, before verification.""" - - BOOKER_EMAIL = "offline-booker@example.com" - - def setUp(self): - self.test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.test_event.apply_tax = False - self.test_event.send_booking_confirmation_email = 1 - self.test_event.booking_confirmation_email_template = None - self.test_event.offline_acknowledgement_email_template = None - self.test_event.send_ticket_email = 0 - self.test_event.save() - - set_team_settings(self.test_event.team, default_booking_confirmation_email_template=None) - self.addCleanup(frappe.clear_document_cache, "Buzz Team Settings", self.test_event.team) - - if not frappe.db.exists("User", self.BOOKER_EMAIL): - frappe.get_doc( - { - "doctype": "User", - "email": self.BOOKER_EMAIL, - "first_name": "Offline", - "enabled": 1, - "user_type": "Website User", - "send_welcome_email": 0, - } - ).insert(ignore_permissions=True) - - self.ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.test_event.name, - "title": "Offline Acknowledgement Ticket", - "prices": [{"currency": "INR", "price": 100}], - } - ).insert() - - def tearDown(self): - frappe.delete_doc("Event Ticket Type", self.ticket_type.name, force=True) - - def _make_offline_booking(self, user): - """A booking in the state offline_booking_response leaves behind: a draft - awaiting verification, with no tickets.""" - booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.test_event.name, - "user": user, - "payment_method": "Offline", - "offline_payment_method": "Bank Transfer", - "status": "Approval Pending", - "payment_status": "Verification Pending", - "attendees": [ - { - "ticket_type": self.ticket_type.name, - "first_name": "John", - "email": "john-offline@example.com", - } - ], - } - ).insert() - return booking - - def _create_template(self, name, subject_prefix): - if frappe.db.exists("Email Template", name): - frappe.delete_doc("Email Template", name, force=True) - return frappe.get_doc( - { - "doctype": "Email Template", - "name": name, - "subject": f"{subject_prefix} - {{{{ event_title }}}}", - "response": f"{subject_prefix} content
", - } - ).insert() - - @patch("frappe.sendmail") - def test_sends_acknowledgement_to_booker(self, mock_sendmail): - booking = self._make_offline_booking(self.BOOKER_EMAIL) - booking.send_offline_acknowledgement_email() - - mock_sendmail.assert_called_once() - self.assertIn(self.BOOKER_EMAIL, mock_sendmail.call_args[1]["recipients"]) - self.assertEqual(mock_sendmail.call_args[1]["reference_doctype"], "Event Booking") - self.assertEqual(mock_sendmail.call_args[1]["reference_name"], booking.name) - - @patch("frappe.sendmail") - def test_uses_inline_template_when_none_configured(self, mock_sendmail): - self._make_offline_booking(self.BOOKER_EMAIL).send_offline_acknowledgement_email() - - self.assertEqual(mock_sendmail.call_args[1]["template"], "offline_booking_acknowledgement") - - @patch("frappe.sendmail") - def test_carries_the_booking_summary(self, mock_sendmail): - booking = self._make_offline_booking(self.BOOKER_EMAIL) - booking.send_offline_acknowledgement_email() - - args = mock_sendmail.call_args[1]["args"] - self.assertEqual(args["doc"].name, booking.name) - self.assertEqual(args["event_title"], self.test_event.title) - self.assertEqual(len(args["attendee_rows"]), 1) - # Ticket types autoname to integers and arrive off the row as strings, so a - # title lookup keyed on the raw value silently prints the id instead. - self.assertEqual(args["attendee_rows"][0]["ticket_type_title"], self.ticket_type.title) - - def test_builtin_template_renders(self): - """The template is only exercised end-to-end here: every other test mocks the - send, so a broken Jinja tag would otherwise reach production silently.""" - booking = self._make_offline_booking(self.BOOKER_EMAIL) - - html = frappe.render_template( - "templates/emails/offline_booking_acknowledgement.html", booking.get_booking_email_args() - ) - - self.assertIn("Payment verification pending", html) - self.assertIn(booking.name, html) - self.assertIn(booking.offline_payment_method, html) - self.assertIn(self.ticket_type.title, html) - - @patch("frappe.sendmail") - def test_uses_event_template_when_set(self, mock_sendmail): - template = self._create_template("Offline Acknowledgement Event Template", "OFFLINE") - self.test_event.offline_acknowledgement_email_template = template.name - self.test_event.save() - - self._make_offline_booking(self.BOOKER_EMAIL).send_offline_acknowledgement_email() - - mock_sendmail.assert_called_once() - self.assertIn("OFFLINE", mock_sendmail.call_args[1]["subject"]) - - @patch("frappe.sendmail") - def test_ignores_the_confirmation_template(self, mock_sendmail): - """The acknowledgement has its own template field; the confirmation's, event-level - or team-level, must not leak into it.""" - event_template = self._create_template("Offline Acknowledgement Confirmation Template", "EVENT") - team_template = self._create_template("Offline Acknowledgement Team Template", "TEAM") - self.test_event.booking_confirmation_email_template = event_template.name - self.test_event.save() - set_team_settings( - self.test_event.team, default_booking_confirmation_email_template=team_template.name - ) - - self._make_offline_booking(self.BOOKER_EMAIL).send_offline_acknowledgement_email() - - self.assertEqual(mock_sendmail.call_args[1]["template"], "offline_booking_acknowledgement") - - @patch("frappe.sendmail") - def test_respects_event_toggle_off(self, mock_sendmail): - self.test_event.send_booking_confirmation_email = 0 - self.test_event.save() - - self._make_offline_booking(self.BOOKER_EMAIL).send_offline_acknowledgement_email() - - mock_sendmail.assert_not_called() - - @patch("frappe.sendmail") - def test_skips_system_users(self, mock_sendmail): - for user in ("Administrator", "Guest"): - with self.subTest(user=user): - self._make_offline_booking(user).send_offline_acknowledgement_email() - - mock_sendmail.assert_not_called() - - -class TestZoomBackedCategoryBooking(IntegrationTestCase): - """Zoom needs a last name on every registrant, for meetings as much as webinars.""" - - def setUp(self): - super().setUp() - self.event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.event.name, - "title": "Zoom Category Ticket", - "is_published": True, - } - ).insert() - - def tearDown(self): - frappe.db.rollback() - - def _book_without_last_name(self, category): - self.event.db_set("category", category) - return process_booking( - attendees=[ - { - "first_name": "Nolast", - "email": "nolast@example.com", - "ticket_type": str(self.ticket_type.name), - "add_ons": [], - } - ], - event=str(self.event.name), - ) +class TestZoomBackedCategoryBooking(BookingTestCase): def test_last_name_required_for_webinar_category(self): - self.assertRaises(frappe.ValidationError, self._book_without_last_name, "Webinars") + self.assertRaises(frappe.ValidationError, self.book_without_last_name, "Webinars") def test_last_name_required_for_zoom_meeting_category(self): - self.assertRaises(frappe.ValidationError, self._book_without_last_name, "Zoom Meeting") + self.assertRaises(frappe.ValidationError, self.book_without_last_name, "Zoom Meeting") def test_last_name_not_required_for_other_categories(self): - result = self._book_without_last_name("Conferences") + payload = self.book_without_last_name(self.event.category) + + self.assertTrue(frappe.db.exists("Event Booking", payload.booking_name)) - self.assertIn("booking_name", result) + def book_without_last_name(self, category: str): + self.set_event({"category": category}) + return process_booking(self.booking_request()) diff --git a/buzz/ticketing/doctype/event_booking/test_event_booking_refund.py b/buzz/ticketing/doctype/event_booking/test_event_booking_refund.py index f4216f0d..e8921e7e 100644 --- a/buzz/ticketing/doctype/event_booking/test_event_booking_refund.py +++ b/buzz/ticketing/doctype/event_booking/test_event_booking_refund.py @@ -1,147 +1,51 @@ # Copyright (c) 2026, BWH Studios and Contributors # See license.txt -from unittest.mock import MagicMock, patch +from unittest.mock import patch import frappe -from frappe.tests import IntegrationTestCase -from frappe.utils import today -from pydantic import ValidationError -from buzz.api.checkin import validate_ticket_for_checkin +from buzz.api.checkin import checkin_ticket, validate_ticket_for_checkin from buzz.api.exceptions import Conflict -from buzz.payments import handle_refund_notification - -TICKET_PRICE = 500 -TAX_PERCENTAGE = 10 -CHARGED_PER_TICKET = 550.0 +from buzz.tests.factories import BuzzEventFactory, PaymentGatewayFactory +from buzz.ticketing.doctype.event_booking.refund_test_case import CHARGED_PER_TICKET, BookingRefundTestCase def make_payment_gateway(gateway: str) -> None: + """Kept for modules that still import it. New tests use `PaymentGatewayFactory`.""" if not frappe.db.exists("Payment Gateway", gateway): - frappe.get_doc({"doctype": "Payment Gateway", "gateway": gateway}).insert() - - -class BookingRefundTestCase(IntegrationTestCase): - def setUp(self): - self.event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.event.apply_tax = 1 - self.event.tax_inclusive = 0 - self.event.tax_percentage = TAX_PERCENTAGE - self.event.tax_label = "GST" - self.event.save() - - self.ticket_type = frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.event.name, - "title": "Refundable", - "prices": [{"currency": "INR", "price": TICKET_PRICE}], - "is_published": True, - } - ).insert() - - self.booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.event.name, - "user": "Administrator", - "attendees": [ - {"ticket_type": self.ticket_type.name, "first_name": "John", "email": "john@example.com"}, - { - "ticket_type": self.ticket_type.name, - "first_name": "Jenny", - "email": "jenny@example.com", - }, - ], - } - ).insert() - self.booking.payment_status = "Paid" - self.booking.submit() - - def refund_id(self, suffix: str = "1") -> str: - """Refund ids are unique in the database, so keep them unique per test too.""" - return f"rfnd_{self._testMethodName}_{suffix}" - - def refunds(self) -> list: - return [ - frappe.get_doc("Event Booking Refund", refund.name) - for refund in frappe.get_all( - "Event Booking Refund", filters={"booking": self.booking.name}, order_by="creation asc" - ) - ] - - def check_in(self, ticket: str) -> None: - frappe.get_doc({"doctype": "Event Check In", "ticket": ticket, "date": today()}).insert().submit() - - def make_payment(self, gateway: str = "Razorpay", payment_id: str = "pay_123") -> None: - make_payment_gateway(gateway) - frappe.get_doc( - { - "doctype": "Event Payment", - "user": "Administrator", - "amount": self.booking.total_amount, - "currency": self.booking.currency, - "reference_doctype": "Event Booking", - "reference_docname": self.booking.name, - "payment_gateway": gateway, - "payment_received": 1, - "payment_id": payment_id, - } - ).insert() + PaymentGatewayFactory.create(gateway=gateway) class TestRefundSummary(BookingRefundTestCase): def test_each_ticket_carries_its_share_of_the_charged_total(self): - # 500 + 500 booked, 10% tax on top, so each ticket cost the buyer 550. summary = self.booking.get_refund_summary() - self.assertEqual(len(summary["tickets"]), 2) self.assertEqual([ticket["amount"] for ticket in summary["tickets"]], [550.0, 550.0]) def test_tickets_are_named_and_linked_to_their_attendee(self): summary = self.booking.get_refund_summary() self.assertTrue(all(ticket["ticket"] for ticket in summary["tickets"])) - self.assertEqual( - sorted(ticket["attendee"] for ticket in summary["tickets"]), - ["Jenny", "John"], - ) + self.assertEqual(sorted(ticket["attendee"] for ticket in summary["tickets"]), ["Jenny", "John"]) def test_the_whole_total_is_refundable_before_any_refund(self): summary = self.booking.get_refund_summary() self.assertEqual(summary["committed"], 0) - self.assertEqual(summary["remaining"], self.booking.total_amount) + self.assertEqual(summary["remaining"], 1100) def test_a_checked_in_ticket_is_not_offered(self): - # The attendee has been through the door, so the ticket has been used. - used = self.booking.get_refund_summary()["tickets"][0]["ticket"] - self.check_in(used) - - summary = self.booking.get_refund_summary() + checked_in_ticket = self.refundable_tickets()[0] + checkin_ticket(checked_in_ticket) - self.assertNotIn(used, [ticket["ticket"] for ticket in summary["tickets"]]) - self.assertEqual(len(summary["tickets"]), 1) + self.assertEqual(len(self.refundable_tickets()), 1) + self.assertNotIn(checked_in_ticket, self.refundable_tickets()) class TestBookingRefund(BookingRefundTestCase): - def refund(self, **kwargs): - client = MagicMock() - client.refund_payment.return_value = { - "id": kwargs.pop("refund_id", self.refund_id()), - "status": "pending", - "amount": int(kwargs.get("amount", CHARGED_PER_TICKET) * 100), - } - - with patch("buzz.ticketing.doctype.event_booking.event_booking.get_controller", return_value=client): - result = self.booking.refund(**kwargs) - - self.booking.reload() - return result, client - def test_refund_is_refused_when_the_gateway_is_not_razorpay(self): - self.make_payment(gateway="PayPal") + self.make_payment(gateway=PaymentGatewayFactory.create().name) with self.assertRaises(frappe.ValidationError) as raised: self.booking.refund(amount=100) @@ -153,619 +57,141 @@ def test_refund_is_refused_without_a_received_payment(self): def test_refunding_tickets_records_the_refund_and_marks_it_initiated(self): self.make_payment() - tickets = self.booking.get_refund_summary()["tickets"] - _, client = self.refund(amount=CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) + client = self.initiate_refund(CHARGED_PER_TICKET, tickets=self.refundable_tickets()[:1]) - client.refund_payment.assert_called_once_with("pay_123", CHARGED_PER_TICKET) + client.refund_payment.assert_called_once_with(self.payment.payment_id, CHARGED_PER_TICKET) self.assertEqual(self.booking.refund_status, "Refund Initiated") - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.refunds()[0].refund_id, self.refund_id()) - self.assertEqual(self.refunds()[0].status, "Initiated") - self.assertEqual(self.refunds()[0].amount, CHARGED_PER_TICKET) + refund = self.refunds()[0] + self.assertEqual( + (refund.refund_id, refund.status, refund.amount), (self.refund_id(), "Initiated", 550) + ) def test_an_unsettled_refund_cancels_nothing_yet(self): - # Razorpay can take days. Nothing may cancel tickets before it answers, - # or a refund that fails leaves the customer with neither money nor seat. + # A refund can still fail, so tickets are only cancelled once it settles. self.make_payment() - tickets = self.booking.get_refund_summary()["tickets"] + ticket = self.refundable_tickets()[0] - self.refund(amount=CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) self.assertIsNone(self.refunds()[0].cancellation_request) - self.assertFalse(frappe.db.exists("Ticket Cancellation Request", {"booking": self.booking.name})) - self.assertEqual([row.ticket for row in self.refunds()[0].tickets], [tickets[0]["ticket"]]) + self.assertEqual(self.cancellation_requests(), []) + self.assertEqual([row.ticket for row in self.refunds()[0].tickets], [ticket]) def test_a_ticket_from_another_booking_is_refused(self): self.make_payment() - stranger = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.event.name, - "user": "Administrator", - "attendees": [ - { - "ticket_type": self.ticket_type.name, - "first_name": "Stranger", - "email": "stranger@example.com", - } - ], - } - ).insert() - stranger.payment_status = "Paid" - stranger.submit() - foreign_ticket = frappe.db.get_value("Event Ticket", {"booking": stranger.name}, "name") - - # Patch the gateway so nothing but the ownership check can refuse this. + other_booking = self.create_paid_booking(("Stranger",), self.ticket_types[:1]) + foreign_ticket = frappe.db.get_value("Event Ticket", {"booking": other_booking.name}, "name") + + # The gateway is patched, so only the ownership check can refuse this. with self.assertRaises(frappe.ValidationError): - self.refund(amount=CHARGED_PER_TICKET, tickets=[foreign_ticket]) + self.initiate_refund(CHARGED_PER_TICKET, tickets=[foreign_ticket]) - self.assertFalse(frappe.db.exists("Ticket Cancellation Request", {"booking": self.booking.name})) + self.assertEqual(self.cancellation_requests(), []) self.assertEqual(frappe.db.get_value("Event Ticket", foreign_ticket, "docstatus"), 1) def test_a_ticket_that_has_been_checked_in_is_refused(self): self.make_payment() - tickets = self.booking.get_refund_summary()["tickets"] - self.check_in(tickets[0]["ticket"]) + ticket = self.refundable_tickets()[0] + checkin_ticket(ticket) with self.assertRaises(frappe.ValidationError) as raised: - self.refund(amount=CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) self.assertIn("checked in", str(raised.exception)) - self.assertFalse(self.refunds()) + self.assertEqual(self.refunds(), []) def test_a_ticket_a_refund_holds_cannot_be_checked_in(self): - # The refund has not settled, so the ticket is still live, but the money - # is already on its way back and the door must not let it through. self.make_payment() - tickets = self.booking.get_refund_summary()["tickets"] - - self.refund(amount=CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) + ticket = self.refundable_tickets()[0] + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) with self.assertRaises(Conflict): - validate_ticket_for_checkin(tickets[0]["ticket"]) + validate_ticket_for_checkin(str(ticket)) self.assertEqual(frappe.local.message_log[-1]["title"], "Ticket Refunded") def test_a_custom_amount_refund_cancels_no_tickets(self): self.make_payment() - self.refund(amount=100) + self.initiate_refund(100) self.assertIsNone(self.refunds()[0].cancellation_request) - self.assertFalse(frappe.db.exists("Ticket Cancellation Request", {"booking": self.booking.name})) - - -class TestRefundNotification(BookingRefundTestCase): - def setUp(self): - super().setUp() - self.make_payment() - - def make_log(self, payload: dict): - return frappe.get_doc( - { - "doctype": "Integration Request", - "integration_request_service": "Razorpay", - "request_description": "Refund Notification", - "is_remote_request": 1, - "status": "Queued", - "data": frappe.as_json(payload), - } - ).insert(ignore_permissions=True) - - def notify(self, refund_id: str, status: str, amount: float) -> None: - log = self.make_log( - { - "event": f"refund.{status}", - "payload": { - "refund": { - "entity": { - "id": refund_id, - "status": status, - "amount": int(amount * 100), - "payment_id": "pay_123", - } - } - }, - } - ) - - # Cancelling a ticket emails the attendee, and the test site has no - # outgoing account. Delivery is not what these tests are about. - with patch("frappe.sendmail"): - handle_refund_notification("Integration Request", log.name) - - self.booking.reload() - return log - - def initiate(self, refund_id: str, amount: float, tickets: list[str] | None = None) -> None: - client = MagicMock() - client.refund_payment.return_value = { - "id": refund_id, - "status": "pending", - "amount": int(amount * 100), - } - - with patch("buzz.ticketing.doctype.event_booking.event_booking.get_controller", return_value=client): - self.booking.refund(amount=amount, tickets=tickets) - - self.booking.reload() - - def test_a_processed_refund_covering_the_total_marks_the_booking_refunded(self): - self.initiate(self.refund_id("1"), self.booking.total_amount) - - self.notify(self.refund_id("1"), "processed", self.booking.total_amount) - - self.assertEqual(self.booking.refund_status, "Refunded") - self.assertEqual(self.booking.refunded_amount, self.booking.total_amount) - self.assertEqual(self.refunds()[0].status, "Processed") - - def test_a_processed_refund_below_the_total_marks_the_booking_partially_refunded(self): - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET) - - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - self.assertEqual(self.booking.refund_status, "Partially Refunded") - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - - def test_a_full_refund_buzz_never_raised_cancels_every_remaining_ticket(self): - # Raised on the Razorpay dashboard: the webhook is the first Buzz hears of - # it, and it carries no tickets. - self.notify(self.refund_id("1"), "processed", self.booking.total_amount) - - self.assertEqual(self.booking.refund_status, "Refunded") - self.assertEqual(len(self.refunds()[0].tickets), 2) - self.assertEqual( - frappe.db.get_value( - "Ticket Cancellation Request", self.refunds()[0].cancellation_request, "docstatus" - ), - 1, - ) - - def test_a_partial_refund_buzz_never_raised_cancels_nothing(self): - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - self.assertEqual(self.refunds()[0].tickets, []) - self.assertFalse(frappe.db.exists("Ticket Cancellation Request", {"booking": self.booking.name})) - - def test_the_same_event_arriving_twice_is_not_counted_twice(self): - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET) - - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - - def test_a_failed_refund_is_not_counted_and_cancels_nothing(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) - - self.notify(self.refund_id("1"), "failed", CHARGED_PER_TICKET) - - self.assertEqual(self.refunds()[0].status, "Failed") - self.assertEqual(self.booking.refunded_amount, 0) - self.assertFalse(frappe.db.exists("Ticket Cancellation Request", {"booking": self.booking.name})) - self.assertEqual(frappe.db.get_value("Event Ticket", tickets[0]["ticket"], "docstatus"), 1) - - def test_a_webhook_processed_as_guest_still_updates_the_booking(self): - # The refund webhook is an allow_guest endpoint, so the job it enqueues - # runs as Guest — who has no write permission on Event Booking. - tickets = self.booking.get_refund_summary()["tickets"] - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) - frappe.set_user("Guest") - self.addCleanup(frappe.set_user, "Administrator") - - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - self.assertEqual(self.refunds()[0].status, "Processed") - self.assertEqual(self.booking.refund_status, "Partially Refunded") - self.assertEqual( - frappe.db.get_value( - "Ticket Cancellation Request", self.refunds()[0].cancellation_request, "docstatus" - ), - 1, - ) - - def test_a_processed_refund_raises_and_submits_the_cancellation(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) - - self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - request = frappe.get_doc("Ticket Cancellation Request", self.refunds()[0].cancellation_request) - self.assertEqual(request.status, "Accepted") - self.assertEqual(request.docstatus, 1) - self.assertEqual(frappe.db.get_value("Event Ticket", tickets[0]["ticket"], "docstatus"), 2) - - def test_a_cancellation_that_cannot_go_through_leaves_the_refund_recorded(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET, tickets=[tickets[0]["ticket"]]) - refund = self.refunds()[0] - - with patch("frappe.sendmail", side_effect=Exception("no outgoing email account")): - refund.apply_gateway_status("processed", CHARGED_PER_TICKET) - - self.booking.reload() - self.assertEqual(self.refunds()[0].status, "Processed") - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - - request = frappe.get_doc("Ticket Cancellation Request", self.refunds()[0].cancellation_request) - self.assertEqual(request.docstatus, 0) - self.assertEqual(request.status, "In Review") - self.assertEqual(frappe.db.get_value("Event Ticket", tickets[0]["ticket"], "docstatus"), 1) - - def test_the_integration_request_is_linked_back_to_the_booking(self): - self.initiate(self.refund_id("1"), CHARGED_PER_TICKET) - - log = self.notify(self.refund_id("1"), "processed", CHARGED_PER_TICKET) - - self.assertEqual( - frappe.db.get_value("Integration Request", log.name, "reference_docname"), self.booking.name - ) - - def test_a_refund_for_an_unknown_payment_is_ignored(self): - log = self.make_log( - { - "event": "refund.processed", - "payload": { - "refund": { - "entity": { - "id": self.refund_id("x"), - "payment_id": "pay_nope", - "status": "processed", - "amount": 55000, - } - } - }, - } - ) - - self.assertIsNone(handle_refund_notification("Integration Request", log.name)) - self.booking.reload() - self.assertFalse(self.refunds()) - - def test_a_payload_without_a_refund_is_refused(self): - # Only refund events reach this handler, so one carrying no refund is a - # defect for the caller to log, not an event to pass over quietly. - log = self.make_log({"event": "refund.processed", "payload": {}}) + self.assertEqual(self.cancellation_requests(), []) - self.assertRaises(ValidationError, handle_refund_notification, "Integration Request", log.name) +class TestRefundCeiling(BookingRefundTestCase): + ticket_prices = (2000, 3000) + attendee_names = ("Cheap", "Pricey") -class TestRefundCeiling(IntegrationTestCase): - """A booking of 2000 + 3000 must never give back more than 5000.""" + @classmethod + def create_event(cls) -> str: + return BuzzEventFactory.create().name def setUp(self): - self.event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.event.apply_tax = 0 - self.event.save() - - self.ticket_types = [ - frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.event.name, - "title": f"Tier {price}", - "prices": [{"currency": "INR", "price": price}], - "is_published": True, - } - ).insert() - for price in (2000, 3000) - ] - - self.booking = frappe.get_doc( - { - "doctype": "Event Booking", - "event": self.event.name, - "user": "Administrator", - "attendees": [ - { - "ticket_type": self.ticket_types[0].name, - "first_name": "Cheap", - "email": "cheap@example.com", - }, - { - "ticket_type": self.ticket_types[1].name, - "first_name": "Pricey", - "email": "pricey@example.com", - }, - ], - } - ).insert() - self.booking.payment_status = "Paid" - self.booking.submit() - - make_payment_gateway("Razorpay") - frappe.get_doc( - { - "doctype": "Event Payment", - "user": "Administrator", - "amount": self.booking.total_amount, - "currency": self.booking.currency, - "reference_doctype": "Event Booking", - "reference_docname": self.booking.name, - "payment_gateway": "Razorpay", - "payment_received": 1, - "payment_id": "pay_123", - } - ).insert() - - def refund_id(self, suffix: str = "1") -> str: - """Refund ids are unique in the database, so keep them unique per test too.""" - return f"rfnd_{self._testMethodName}_{suffix}" - - def refund(self, amount: float, refund_id: str | None = None, tickets: list[str] | None = None): - refund_id = refund_id or self.refund_id() - client = MagicMock() - client.refund_payment.return_value = { - "id": refund_id, - "status": "pending", - "amount": int(amount * 100), - } - - with patch("buzz.ticketing.doctype.event_booking.event_booking.get_controller", return_value=client): - return self.booking.refund(amount=amount, tickets=tickets) - - def settle(self, refund_id: str, amount: float) -> None: - refund = frappe.get_doc("Event Booking Refund", {"refund_id": refund_id}) - with patch("frappe.sendmail"): - refund.apply_gateway_status("processed", amount) - - self.booking.reload() + super().setUp() + self.make_payment() def test_the_booking_total_is_the_ceiling(self): self.assertEqual(self.booking.total_amount, 5000) def test_a_second_refund_cannot_exceed_what_is_left_after_a_settled_one(self): - self.refund(3000, refund_id=self.refund_id("1")) - self.settle(self.refund_id("1"), 3000) + self.initiate_refund(3000) + self.settle_refund(self.refund_id(), 3000) with self.assertRaises(frappe.ValidationError) as raised: - self.refund(3000, refund_id=self.refund_id("2")) + self.initiate_refund(3000, refund_id=self.refund_id("2")) self.assertIn("2,000", str(raised.exception)) def test_a_second_refund_cannot_exceed_what_is_left_while_the_first_is_still_initiated(self): - # The gateway has not settled the first refund yet, so it cannot be relied - # on to reject the second one either. - self.refund(3000, refund_id=self.refund_id("1")) + self.initiate_refund(3000) - self.assertRaises(frappe.ValidationError, self.refund, 3000, self.refund_id("2")) + with self.assertRaises(frappe.ValidationError): + self.initiate_refund(3000, refund_id=self.refund_id("2")) def test_a_failed_refund_frees_its_amount_again(self): - self.refund(3000, refund_id=self.refund_id("1")) - frappe.get_doc("Event Booking Refund", {"refund_id": self.refund_id("1")}).apply_gateway_status( - "failed", 3000 - ) - self.booking.reload() + self.initiate_refund(3000) + self.fail_refund(self.refund_id(), 3000) - self.refund(3000, refund_id=self.refund_id("2")) + self.initiate_refund(3000, refund_id=self.refund_id("2")) - self.assertEqual(len(frappe.get_all("Event Booking Refund", {"booking": self.booking.name})), 2) + self.assertEqual(len(self.refunds()), 2) def test_refunding_exactly_what_is_left_is_allowed(self): - self.refund(3000, refund_id=self.refund_id("1")) - self.settle(self.refund_id("1"), 3000) + self.initiate_refund(3000) + self.settle_refund(self.refund_id(), 3000) - self.refund(2000, refund_id=self.refund_id("2")) - self.settle(self.refund_id("2"), 2000) + self.initiate_refund(2000, refund_id=self.refund_id("2")) + self.settle_refund(self.refund_id("2"), 2000) self.assertEqual(self.booking.refund_status, "Refunded") self.assertEqual(self.booking.refunded_amount, 5000) def test_a_ticket_already_refunded_is_not_offered_again(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.refund(3000, refund_id=self.refund_id("1"), tickets=[tickets[1]["ticket"]]) - self.settle(self.refund_id("1"), 3000) - - remaining = self.booking.get_refund_summary() + low_price_ticket, high_price_ticket = self.refundable_tickets() + self.initiate_refund(3000, tickets=[high_price_ticket]) + self.settle_refund(self.refund_id(), 3000) - self.assertEqual([ticket["ticket"] for ticket in remaining["tickets"]], [tickets[0]["ticket"]]) - self.assertEqual(remaining["remaining"], 2000) + self.assertEqual(self.refundable_tickets(), [low_price_ticket]) + self.assertEqual(self.booking.get_refund_summary()["remaining"], 2000) def test_a_ticket_whose_refund_failed_is_offered_again(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.refund(3000, refund_id=self.refund_id("1"), tickets=[tickets[1]["ticket"]]) - - frappe.get_doc("Event Booking Refund", {"refund_id": self.refund_id("1")}).apply_gateway_status( - "failed", 3000 - ) - self.booking.reload() - - self.assertEqual(len(self.booking.get_refund_summary()["tickets"]), 2) - - -class TestSyncRefunds(BookingRefundTestCase): - def setUp(self): - super().setUp() - self.make_payment() - - def gateway_refund(self, refund_id: str, status: str, amount: float) -> dict: - return {"id": refund_id, "status": status, "amount": int(amount * 100)} - - def sync(self, *gateway_refunds: dict) -> dict: - client = MagicMock() - client.fetch_refunds.return_value = list(gateway_refunds) - - with ( - patch("buzz.ticketing.doctype.event_booking.event_booking.get_controller", return_value=client), - patch("frappe.sendmail"), - ): - result = self.booking.sync_refunds() - - self.booking.reload() - return result - - def initiate(self, refund_id: str, amount: float, tickets: list[str] | None = None) -> None: - client = MagicMock() - client.refund_payment.return_value = { - "id": refund_id, - "status": "pending", - "amount": int(amount * 100), - } - - with patch("buzz.ticketing.doctype.event_booking.event_booking.get_controller", return_value=client): - self.booking.refund(amount=amount, tickets=tickets) - - self.booking.reload() - - def cancellation_requests(self) -> list: - return frappe.get_all( - "Ticket Cancellation Request", - filters={"booking": self.booking.name}, - fields=["name", "docstatus"], - ) - - def test_a_refund_raised_outside_buzz_is_recorded(self): - result = self.sync(self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET)) - - self.assertEqual(result, {"refunds": 1, "created": 1}) - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.refunds()[0].status, "Processed") - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - self.assertEqual(self.booking.refund_status, "Partially Refunded") - - def test_a_payment_with_no_refunds_records_nothing(self): - result = self.sync() + high_price_ticket = self.refundable_tickets()[1] + self.initiate_refund(3000, tickets=[high_price_ticket]) - self.assertEqual(result, {"refunds": 0, "created": 0}) - self.assertEqual(self.refunds(), []) - self.assertEqual(self.booking.refund_status, "") - - def test_a_full_refund_raised_outside_buzz_cancels_every_remaining_ticket(self): - self.sync(self.gateway_refund(self.refund_id(), "processed", self.booking.total_amount)) - - self.assertEqual(self.booking.refund_status, "Refunded") - self.assertEqual(len(self.refunds()[0].tickets), 2) - self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) - - def test_a_partial_refund_raised_outside_buzz_cancels_nothing(self): - self.sync(self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET)) + self.fail_refund(self.refund_id(), 3000) - self.assertEqual(self.refunds()[0].tickets, []) - self.assertEqual(self.cancellation_requests(), []) - - def test_a_refund_the_gateway_still_calls_pending_is_initiated(self): - self.sync(self.gateway_refund(self.refund_id(), "pending", self.booking.total_amount)) - - self.assertEqual(self.refunds()[0].status, "Initiated") - self.assertEqual(self.booking.refund_status, "Refund Initiated") - self.assertEqual(self.booking.refunded_amount, 0) - self.assertEqual(self.cancellation_requests(), []) + self.assertEqual(len(self.refundable_tickets()), 2) - def test_a_refund_buzz_already_holds_is_updated_not_duplicated(self): - ticket = self.booking.get_refund_summary()["tickets"][0]["ticket"] - self.initiate(self.refund_id(), CHARGED_PER_TICKET, tickets=[ticket]) - - result = self.sync(self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET)) - - self.assertEqual(result, {"refunds": 1, "created": 0}) - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.refunds()[0].status, "Processed") - self.assertEqual([row.ticket for row in self.refunds()[0].tickets], [ticket]) - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - - def test_a_settled_refund_is_not_knocked_back_by_a_stale_pending(self): - self.sync(self.gateway_refund(self.refund_id(), "processed", self.booking.total_amount)) - - self.sync(self.gateway_refund(self.refund_id(), "pending", self.booking.total_amount)) - - self.assertEqual(self.refunds()[0].status, "Processed") - self.assertEqual(self.booking.refund_status, "Refunded") - self.assertEqual(self.booking.refunded_amount, self.booking.total_amount) - self.assertEqual(len(self.cancellation_requests()), 1) - - def test_a_failed_refund_is_not_revived_by_a_later_sighting(self): - self.sync(self.gateway_refund(self.refund_id(), "failed", CHARGED_PER_TICKET)) - - self.sync(self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET)) - - self.assertEqual(self.refunds()[0].status, "Failed") - self.assertEqual(self.booking.refunded_amount, 0) - - def test_a_refund_the_webhook_recorded_is_not_duplicated_by_a_sync(self): - log = frappe.get_doc( - { - "doctype": "Integration Request", - "integration_request_service": "Razorpay", - "request_description": "Refund Notification", - "is_remote_request": 1, - "status": "Queued", - "data": frappe.as_json( - { - "payload": { - "refund": { - "entity": { - "id": self.refund_id(), - "status": "processed", - "amount": int(CHARGED_PER_TICKET * 100), - "payment_id": "pay_123", - } - } - } - } - ), - } - ).insert(ignore_permissions=True) + def settle_refund(self, refund_id: str, amount: float) -> None: with patch("frappe.sendmail"): - handle_refund_notification("Integration Request", log.name) - - result = self.sync(self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET)) - - self.assertEqual(result, {"refunds": 1, "created": 0}) - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + self.apply_gateway_status(refund_id, "processed", amount) - def test_syncing_twice_does_not_count_the_same_refund_twice(self): - refund = self.gateway_refund(self.refund_id(), "processed", CHARGED_PER_TICKET) + def fail_refund(self, refund_id: str, amount: float) -> None: + self.apply_gateway_status(refund_id, "failed", amount) - self.sync(refund) - self.sync(refund) - - self.assertEqual(len(self.refunds()), 1) - self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) - - def test_a_checked_in_ticket_is_never_cancelled_by_a_synced_refund(self): - tickets = self.booking.get_refund_summary()["tickets"] - self.check_in(tickets[0]["ticket"]) - - self.sync(self.gateway_refund(self.refund_id(), "processed", self.booking.total_amount)) - - self.assertEqual( - [row.ticket for row in self.refunds()[0].tickets], - [tickets[1]["ticket"]], - ) - - def test_two_refunds_in_one_batch_cancel_the_tickets_once_they_cover_the_total(self): - # Neither half covers the booking on its own; the second one, seeing the - # first already recorded, is what pays back the last of it. - self.sync( - self.gateway_refund(self.refund_id("1"), "processed", CHARGED_PER_TICKET), - self.gateway_refund(self.refund_id("2"), "processed", CHARGED_PER_TICKET), - ) - - refunds = self.refunds() - self.assertEqual(len(refunds), 2) - self.assertEqual(refunds[0].tickets, []) - self.assertEqual(len(refunds[1].tickets), 2) - self.assertEqual(self.booking.refund_status, "Refunded") - self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) - - def test_sync_is_refused_when_the_gateway_is_not_razorpay(self): - frappe.db.set_value( - "Event Payment", - {"reference_docname": self.booking.name}, - "payment_gateway", - "PayPal", - ) - - with self.assertRaises(frappe.ValidationError) as raised: - self.booking.sync_refunds() - - self.assertIn("Razorpay", str(raised.exception)) - - def test_sync_is_refused_without_a_received_payment(self): - frappe.db.set_value("Event Payment", {"reference_docname": self.booking.name}, "payment_received", 0) - - self.assertRaises(frappe.ValidationError, self.booking.sync_refunds) + def apply_gateway_status(self, refund_id: str, status: str, amount: float) -> None: + refund = frappe.get_doc("Event Booking Refund", {"refund_id": refund_id}) + refund.apply_gateway_status(status, amount) + self.booking.reload() diff --git a/buzz/ticketing/doctype/event_booking/test_event_eligibility.py b/buzz/ticketing/doctype/event_booking/test_event_eligibility.py index 9b059324..6e585454 100644 --- a/buzz/ticketing/doctype/event_booking/test_event_eligibility.py +++ b/buzz/ticketing/doctype/event_booking/test_event_eligibility.py @@ -1,91 +1,48 @@ # Copyright (c) 2026, BWH Studios and contributors # For license information, please see license.txt -"""Who may create an Event Booking, and for which events. - -Eligibility (event published, registrations open) used to live only in -`BookingService`, so the generic document API reached `Document.insert()` -without it. These pin both halves of the fix: ordinary users cannot write -bookings at all, and the ones who can are held to their own team's events. -""" +"""The generic document API must not skip the eligibility checks `BookingService` runs: +ordinary users cannot write bookings, and Event Managers only book their own team's events.""" import frappe -from frappe.tests import IntegrationTestCase +from frappe.utils import add_days, now_datetime, today from buzz.api.booking import process_booking from buzz.api.booking.exceptions import RegistrationsClosed from buzz.api.booking.schemas import BookingRequest -from buzz.api.forms.test_forms import ensure_event_host, ensure_prompt_named_record -from buzz.events.doctype.buzz_team.test_buzz_team import create_owned_team, create_user -from buzz.events.doctype.buzz_team_membership.buzz_team_membership import upsert_membership +from buzz.tests.base_test_cases import BookingTestCase +from buzz.tests.factories import ( + BuzzEventFactory, + BuzzTeamFactory, + BuzzTeamMembershipFactory, + EventBookingFactory, + EventTicketTypeFactory, + UserFactory, +) ATTENDEE = "eligibility-attendee@example.com" ORGANISER = "eligibility-organiser@example.com" OUTSIDER = "eligibility-outsider@example.com" -class EligibilityTestCase(IntegrationTestCase): +class EligibilityTestCase(BookingTestCase): @classmethod def setUpClass(cls): super().setUpClass() - cls.category = ensure_prompt_named_record("Event Category", "Eligibility Category") - cls.host = ensure_event_host("Eligibility Host") - owner = create_user("eligibility-team-owner@example.com", "Eligibility") - cls.team = create_owned_team(f"Eligibility Team {frappe.generate_hash(length=6)}", owner) - cls.event = cls.make_event() - cls.event.reload() - - @classmethod - def make_event(cls, **overrides): - values = { - "doctype": "Buzz Event", - "title": f"Eligibility Event {frappe.generate_hash(length=6)}", - "team": cls.team, - "start_date": "2030-01-01", - "end_date": "2030-01-01", - "start_time": "10:00:00", - "end_time": "18:00:00", - "medium": "Online", - "category": cls.category, - "host": cls.host, - "is_published": 1, - } - values.update(overrides) - return frappe.get_doc(values).insert(ignore_permissions=True) + cls.attendee = UserFactory.create_once(ATTENDEE).name + cls.organiser = UserFactory.create_once(ORGANISER).name + BuzzTeamMembershipFactory.create(team=cls.event.team, user=cls.organiser) + # An Event Manager by role, but on a team that does not own `cls.event`. + cls.outsider = UserFactory.create_once(OUTSIDER).name + BuzzTeamFactory.create_owned_by(cls.outsider) def setUp(self): - frappe.set_user("Administrator") - frappe.clear_messages() - self.addCleanup(frappe.set_user, "Administrator") - self.addCleanup(frappe.clear_document_cache, "Buzz Event", self.event.name) - - self.attendee = create_user(ATTENDEE, "Attendee") - self.organiser = create_user(ORGANISER, "Organiser") - upsert_membership(self.team, self.organiser, "Manager") - self.set_event({"is_published": 1, "registrations_close_at": None, "allow_guest_booking": 0}) - - self.paid_ticket_type = self.make_ticket_type(price=5000) - - def make_ticket_type(self, event=None, price=0): - return frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": event or self.event.name, - "title": f"Eligibility {frappe.generate_hash(length=6)}", - "prices": [{"currency": "INR", "price": price}], - "is_published": 1, - } - ).insert(ignore_permissions=True) - - def outsider_managing_another_team(self) -> str: - """An Event Manager by role, but on a team that does not own `self.event`.""" - outsider = create_user(OUTSIDER, "Outsider") - other_team = create_owned_team(f"Eligibility Other {frappe.generate_hash(length=6)}", outsider) - upsert_membership(other_team, outsider, "Manager") - return outsider - - def set_event(self, values): - frappe.db.set_value("Buzz Event", self.event.name, values) - frappe.clear_document_cache("Buzz Event", self.event.name) + super().setUp() + self.paid_ticket_type = EventTicketTypeFactory.create( + event=self.event.name, prices=[{"currency": "INR", "price": 5000}] + ) + + def close_registrations(self): + self.set_event({"registrations_close_at": add_days(now_datetime(), -1)}) def attendee_row(self, **overrides) -> dict: row = { @@ -103,27 +60,14 @@ def booking_request(self, **overrides) -> BookingRequest: return BookingRequest(**values) def insert_as(self, session_user: str, **overrides): - """Insert under the caller's own permissions, like the generic document API — - no `ignore_permissions`, so role permissions and `validate` decide.""" - payload = { - "doctype": "Event Booking", - "event": str(self.event.name), - "user": session_user, - "currency": "INR", - "attendees": [self.attendee_row()], - } - payload.update(overrides) - - frappe.set_user(session_user) - try: - return frappe.get_doc(payload).insert() - finally: - frappe.set_user("Administrator") + """No `ignore_permissions`, like the generic document API.""" + values = {"user": session_user, "currency": "INR", "attendees": [self.attendee_row()], **overrides} + with self.set_user(session_user): + return EventBookingFactory.create(event=str(self.event.name), **values) class TestOrdinaryUsersCannotWriteBookings(EligibilityTestCase): - """Bookings are only ever written by the service flow, which runs with - `ignore_permissions`. Nothing user-facing needs create or write.""" + """Only the service flow writes bookings, and it runs with `ignore_permissions`.""" def test_every_new_user_is_granted_the_buzz_user_role(self): self.assertIn("Buzz User", frappe.get_roles(self.attendee)) @@ -146,38 +90,34 @@ def test_a_direct_insert_by_a_buzz_user_is_refused(self): class TestEventManagersAreHeldToTheirOwnTeam(EligibilityTestCase): - """`Event Manager` is granted by membership of any team, so the role alone must - not open another team's events.""" + """Membership of any team grants `Event Manager`, so the role alone must not open other teams' events.""" def test_an_outsider_may_not_book_an_unpublished_event(self): - outsider = self.outsider_managing_another_team() self.set_event({"is_published": 0}) with self.assertRaises(frappe.ValidationError): - self.insert_as(outsider, user=outsider) + self.insert_as(self.outsider) - self.assertIn("Event Manager", frappe.get_roles(outsider)) + self.assertIn("Event Manager", frappe.get_roles(self.outsider)) def test_an_outsider_may_not_book_after_registrations_close(self): - outsider = self.outsider_managing_another_team() - self.set_event({"registrations_close_at": "2020-01-01 00:00:00"}) + self.close_registrations() with self.assertRaises(RegistrationsClosed): - self.insert_as(outsider, user=outsider) + self.insert_as(self.outsider) def test_an_outsider_may_not_book_an_event_that_already_ended(self): # No explicit cutoff, so `are_registrations_closed` falls back to the end datetime. - outsider = self.outsider_managing_another_team() - self.set_event({"start_date": "2020-01-01", "end_date": "2020-01-01"}) - self.addCleanup(self.set_event, {"start_date": "2030-01-01", "end_date": "2030-01-01"}) + self.addCleanup( + self.set_event, {"start_date": self.event.start_date, "end_date": self.event.end_date} + ) + self.set_event({"start_date": add_days(today(), -10), "end_date": add_days(today(), -10)}) with self.assertRaises(RegistrationsClosed): - self.insert_as(outsider, user=outsider) + self.insert_as(self.outsider) def test_an_outsider_may_book_an_open_published_event(self): - outsider = self.outsider_managing_another_team() - - booking = self.insert_as(outsider, user=outsider) + booking = self.insert_as(self.outsider) self.assertTrue(frappe.db.exists("Event Booking", booking.name)) @@ -185,47 +125,42 @@ def test_an_outsider_may_book_an_open_published_event(self): class TestTheServiceFlowRefusesIneligibleEvents(EligibilityTestCase): def test_service_refuses_an_unpublished_event(self): self.set_event({"is_published": 0}) - frappe.set_user(self.attendee) - with self.assertRaises(frappe.ValidationError): + with self.set_user(self.attendee), self.assertRaises(frappe.ValidationError): process_booking(self.booking_request()) self.assertIn("Event is not live", frappe.local.message_log[-1]["message"]) def test_service_refuses_once_registrations_have_closed(self): - self.set_event({"registrations_close_at": "2020-01-01 00:00:00"}) - frappe.set_user(self.attendee) + self.close_registrations() - with self.assertRaises(RegistrationsClosed): + with self.set_user(self.attendee), self.assertRaises(RegistrationsClosed): process_booking(self.booking_request()) class TestLegitimateFlowsStillWork(EligibilityTestCase): def test_the_guard_does_not_apply_to_the_vetted_service_flow(self): # Free ticket keeps the flow off the payment gateway. - free_ticket_type = self.make_ticket_type(price=0) + free_ticket_type = str(self.free_ticket_type.name) + request = self.booking_request(attendees=[self.attendee_row(ticket_type=free_ticket_type)]) - frappe.set_user(self.attendee) - payload = process_booking( - self.booking_request(attendees=[self.attendee_row(ticket_type=str(free_ticket_type.name))]) - ).__json__() + with self.set_user(self.attendee): + payload = process_booking(request) - self.assertIn("booking_name", payload) - self.assertTrue(frappe.db.exists("Event Booking", payload["booking_name"])) + self.assertTrue(frappe.db.exists("Event Booking", payload.booking_name)) def test_an_event_organiser_may_book_their_own_closed_event(self): - # Organisers are exempt (comp tickets, pre-launch testing). - self.set_event({"is_published": 0, "registrations_close_at": "2020-01-01 00:00:00"}) + self.set_event({"is_published": 0}) + self.close_registrations() - booking = self.insert_as(self.organiser, user=self.organiser) + booking = self.insert_as(self.organiser) self.assertTrue(frappe.db.exists("Event Booking", booking.name)) def test_a_draft_booked_while_open_survives_registrations_closing(self): - # A draft made while open must stay writable via a trusted flow (payment - # authorisation, offline approval) after close, or a paid booking is stranded. - booking = self.insert_as(self.organiser, user=self.organiser) - self.set_event({"registrations_close_at": "2020-01-01 00:00:00"}) + # Payment authorisation and offline approval resave the draft after close. + booking = self.insert_as(self.organiser) + self.close_registrations() booking.reload() booking.flags.ignore_permissions = True @@ -235,48 +170,36 @@ def test_a_draft_booked_while_open_survives_registrations_closing(self): class TestTheGuardCannotBeSteppedAround(EligibilityTestCase): - """The guard runs on every write, so a draft cannot outlive its event's - eligibility by being edited or repointed after the fact.""" - def test_an_outsider_cannot_grow_a_draft_after_registrations_close(self): - free_type = self.make_ticket_type(price=0) - outsider = self.outsider_managing_another_team() - booking = self.insert_as( - outsider, - user=outsider, - attendees=[self.attendee_row(ticket_type=str(free_type.name))], - ) - self.set_event({"registrations_close_at": "2020-01-01 00:00:00"}) - - frappe.set_user(outsider) - booking.reload() - booking.append( - "attendees", - self.attendee_row(email="smuggled@example.com", ticket_type=str(free_type.name)), - ) - - with self.assertRaises(RegistrationsClosed): - booking.save() + free_type = str(self.free_ticket_type.name) + booking = self.insert_as(self.outsider, attendees=[self.attendee_row(ticket_type=free_type)]) + self.close_registrations() + + with self.set_user(self.outsider): + booking.reload() + booking.append( + "attendees", self.attendee_row(email="smuggled@example.com", ticket_type=free_type) + ) + with self.assertRaises(RegistrationsClosed): + booking.save() self.assertEqual(frappe.db.count("Event Booking Attendee", {"parent": booking.name}), 1) def test_a_draft_cannot_be_repointed_at_a_closed_event(self): - # The guard only runs on creation, so a draft must not be able to reach a - # closed event afterwards. Ticket types are what pin it down. - booking = self.insert_as(self.organiser, user=self.organiser) - closed_event = self.make_event(registrations_close_at="2020-01-01 00:00:00") + # Ticket types pin a draft to its event. + booking = self.insert_as(self.organiser) + closed_event = BuzzEventFactory.create( + team=self.event.team, registrations_close_at=add_days(now_datetime(), -1) + ) - frappe.set_user(self.organiser) - booking.event = closed_event.name - with self.assertRaises(frappe.ValidationError): + with self.set_user(self.organiser), self.assertRaises(frappe.ValidationError): + booking.event = closed_event.name booking.save() self.assertEqual(frappe.db.get_value("Event Booking", booking.name, "event"), str(self.event.name)) class TestWhatValidateAlreadyEnforced(EligibilityTestCase): - """Rules that predate the eligibility guard, kept as guardrails.""" - def test_prices_are_refetched_from_the_ticket_type(self): booking = self.insert_as(self.organiser, attendees=[self.attendee_row(amount=0)]) diff --git a/buzz/ticketing/doctype/event_booking/test_guest_booking.py b/buzz/ticketing/doctype/event_booking/test_guest_booking.py index dca08fea..eaee83a2 100644 --- a/buzz/ticketing/doctype/event_booking/test_guest_booking.py +++ b/buzz/ticketing/doctype/event_booking/test_guest_booking.py @@ -1,211 +1,100 @@ # Copyright (c) 2025, BWH Studios and Contributors # See license.txt -import uuid from base64 import b32encode import frappe import pyotp -from frappe.tests import IntegrationTestCase -from buzz.api.booking import process_booking as process_booking_endpoint -from buzz.api.booking.schemas import BookingRequest +from buzz.api.booking import process_booking +from buzz.tests.base_test_cases import BookingTestCase +OTP_SECRET = b32encode(b"TESTSECRET").decode("utf-8") +WRONG_OTP = "000000" -def process_booking(**kwargs): - return process_booking_endpoint(BookingRequest(**kwargs)).__json__() - -class TestGuestBooking(IntegrationTestCase): +class TestGuestBooking(BookingTestCase): def setUp(self): - frappe.set_user("Administrator") - self.test_event = frappe.get_doc("Buzz Event", {"route": "test-route"}) - self.event_name = str(self.test_event.name) - self.original_allow_guest = self.test_event.allow_guest_booking - self.original_verification = self.test_event.guest_verification_method - self.test_event.allow_guest_booking = True - self.test_event.guest_verification_method = "None" - self.test_event.is_published = True - self.test_event.save() - self.ticket_type = str(self._get_or_create_free_ticket_type()) - - def tearDown(self): - frappe.set_user("Administrator") - self.test_event.reload() - self.test_event.allow_guest_booking = self.original_allow_guest - self.test_event.guest_verification_method = self.original_verification - self.test_event.is_published = False - self.test_event.save() - - # --- helpers --- - - def _generate_test_email(self): - return f"testguest-{uuid.uuid4().hex[:8]}@example.com" - - def _cleanup_test_user(self, email): - frappe.set_user("Administrator") - if frappe.db.exists("User", email): - frappe.delete_doc("User", email, force=True) - - def _get_or_create_free_ticket_type(self): - existing = frappe.db.get_value( - "Event Ticket Type", - {"event": self.test_event.name, "title": "Free (Test)"}, - "name", - ) - if existing: - return existing - return ( - frappe.get_doc( - { - "doctype": "Event Ticket Type", - "event": self.test_event.name, - "title": "Free (Test)", - } - ) - .insert() - .name - ) - - def _make_attendees(self, email): - return [{"ticket_type": self.ticket_type, "first_name": "Test", "last_name": "Guest", "email": email}] - - # --- tests --- - - def test_guest_booking_without_otp(self): - """Full happy path: guest books with verification='None', booking + user created.""" - email = self._generate_test_email() - try: - frappe.set_user("Guest") - result = process_booking( - attendees=self._make_attendees(email), - event=self.event_name, - guest_email=email, - guest_full_name="Test Guest", - ) - self.assertIn("booking_name", result) - self.assertTrue(frappe.db.exists("Event Booking", result["booking_name"])) - self.assertTrue(frappe.db.exists("User", email)) - self.assertIn("Buzz User", frappe.get_roles(email)) - finally: - self._cleanup_test_user(email) - - def test_guest_booking_with_otp(self): - """OTP happy path: cache OTP, pass it in, booking succeeds, OTP cache cleared.""" - email = self._generate_test_email() - self.test_event.guest_verification_method = "Email OTP" - self.test_event.save() - try: - # Simulate OTP generation (same as send_guest_booking_otp) - otp_secret = b32encode(b"TESTSECRET").decode("utf-8") - otp_code = pyotp.HOTP(otp_secret).at(0) - frappe.cache.set_value( - f"guest_booking_otp:email:{email.lower().strip()}", otp_secret, expires_in_sec=600 - ) - - frappe.set_user("Guest") - result = process_booking( - attendees=self._make_attendees(email), - event=self.event_name, - guest_email=email, - guest_full_name="Test Guest", - otp=str(otp_code), - ) - self.assertIn("booking_name", result) - # OTP cache should be cleared after successful verification - self.assertIsNone(frappe.cache.get_value(f"guest_booking_otp:email:{email.lower().strip()}")) - finally: - self._cleanup_test_user(email) + super().setUp() + self.set_event({"allow_guest_booking": 1, "guest_verification_method": "None"}) + self.email = f"testguest-{frappe.generate_hash(length=8)}@example.com" + # process_booking commits, so the guest's User outlives the rollback. + self.addCleanup(self.delete_guest_user) + + def test_guest_booking_without_otp_creates_the_booking_and_user(self): + with self.set_user("Guest"): + payload = self.book_as_guest() + + self.assertTrue(frappe.db.exists("Event Booking", payload.booking_name)) + self.assertIn("Buzz User", frappe.get_roles(self.email)) + + def test_guest_booking_with_otp_clears_the_code(self): + self.enable_email_otp() + self.store_otp() + + with self.set_user("Guest"): + payload = self.book_as_guest(otp=pyotp.HOTP(OTP_SECRET).at(0)) + + self.assertTrue(frappe.db.exists("Event Booking", payload.booking_name)) + self.assertIsNone(frappe.cache.get_value(self.otp_cache_key())) def test_guest_booking_rejected_when_disabled(self): - """Security gate: allow_guest_booking=False raises AuthenticationError.""" - self.test_event.allow_guest_booking = False - self.test_event.save() - - frappe.set_user("Guest") - with self.assertRaises(frappe.AuthenticationError): - process_booking( - attendees=self._make_attendees("nobody@example.com"), - event=self.event_name, - guest_email="nobody@example.com", - guest_full_name="Nobody", - ) + self.set_event({"allow_guest_booking": 0}) + + with self.set_user("Guest"), self.assertRaises(frappe.AuthenticationError): + self.book_as_guest() def test_invalid_otp_rejected(self): - """Wrong OTP code raises ValidationError.""" - email = self._generate_test_email() - self.test_event.guest_verification_method = "Email OTP" - self.test_event.save() - - otp_secret = b32encode(b"TESTSECRET").decode("utf-8") - frappe.cache.set_value( - f"guest_booking_otp:email:{email.lower().strip()}", otp_secret, expires_in_sec=600 - ) - - frappe.set_user("Guest") - with self.assertRaises(frappe.ValidationError): - process_booking( - attendees=self._make_attendees(email), - event=self.event_name, - guest_email=email, - guest_full_name="Test Guest", - otp="000000", - ) + self.enable_email_otp() + self.store_otp() + + with self.set_user("Guest"), self.assertRaises(frappe.ValidationError): + self.book_as_guest(otp=WRONG_OTP) def test_guest_booking_requires_email(self): - """Missing email raises ValidationError.""" - frappe.set_user("Guest") - with self.assertRaises(frappe.ValidationError): - process_booking( - attendees=[ - { - "ticket_type": self.ticket_type, - "first_name": "Test", - "last_name": "Guest", - "email": "t@e.com", - } - ], - event=self.event_name, - guest_email="", - guest_full_name="Test Guest", - ) + with self.set_user("Guest"), self.assertRaises(frappe.ValidationError): + self.book_as_guest(guest_email="") def test_brute_force_lockout(self): - """Repeated wrong OTPs locks out subsequent attempts. - - LoginAttemptTracker(max_consecutive_login_attempts=5) uses strict > - comparison, so lockout triggers after count exceeds the threshold. - """ - email = self._generate_test_email() - self.test_event.guest_verification_method = "Email OTP" - self.test_event.save() - - otp_secret = b32encode(b"TESTSECRET").decode("utf-8") - cache_key = f"guest_booking_otp:email:{email.lower().strip()}" - - frappe.set_user("Guest") - - # Exhaust allowed attempts (Frappe's tracker uses > comparison, - # so we need max_consecutive_login_attempts + 1 failures to trigger lockout) - for _ in range(6): - frappe.cache.set_value(cache_key, otp_secret, expires_in_sec=600) - with self.assertRaises(frappe.ValidationError): - process_booking( - attendees=self._make_attendees(email), - event=self.event_name, - guest_email=email, - guest_full_name="Test Guest", - otp="000000", - ) - - # Next attempt should hit "Too many failed attempts" - frappe.cache.set_value(cache_key, otp_secret, expires_in_sec=600) - with self.assertRaises(frappe.ValidationError) as ctx: - process_booking( - attendees=self._make_attendees(email), - event=self.event_name, - guest_email=email, - guest_full_name="Test Guest", - otp="000000", - ) - self.assertIn("Too many failed attempts", str(ctx.exception)) + # LoginAttemptTracker(max_consecutive_login_attempts=5) compares with `>`, + # so the lockout starts after the sixth failure. + self.enable_email_otp() + + with self.set_user("Guest"): + for _ in range(6): + self.store_otp() + with self.assertRaises(frappe.ValidationError): + self.book_as_guest(otp=WRONG_OTP) + + self.store_otp() + with self.assertRaises(frappe.ValidationError) as raised: + self.book_as_guest(otp=WRONG_OTP) + + self.assertIn("Too many failed attempts", str(raised.exception)) + + def book_as_guest(self, **overrides): + attendees = [ + { + "ticket_type": str(self.free_ticket_type.name), + "first_name": "Test", + "last_name": "Guest", + "email": self.email, + } + ] + values = {"guest_email": self.email, "guest_full_name": "Test Guest", **overrides} + return process_booking(self.booking_request(attendees=attendees, **values)) + + def enable_email_otp(self): + self.set_event({"guest_verification_method": "Email OTP"}) + + def store_otp(self): + """What `send_guest_booking_otp` leaves in the cache.""" + frappe.cache.set_value(self.otp_cache_key(), OTP_SECRET, expires_in_sec=600) + self.addCleanup(frappe.cache.delete_value, self.otp_cache_key()) + + def otp_cache_key(self) -> str: + return f"guest_booking_otp:email:{self.email}" + + def delete_guest_user(self): + if frappe.db.exists("User", self.email): + frappe.delete_doc("User", self.email, force=True, ignore_permissions=True) diff --git a/buzz/ticketing/doctype/event_booking/test_offline_booking.py b/buzz/ticketing/doctype/event_booking/test_offline_booking.py new file mode 100644 index 00000000..a5a4f681 --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/test_offline_booking.py @@ -0,0 +1,111 @@ +import frappe + +from buzz.api.booking import process_booking +from buzz.tests.base_test_cases import BookingTestCase +from buzz.tests.factories import ( + BuzzCouponCodeFactory, + EventBookingFactory, + EventTicketTypeFactory, + OfflinePaymentMethodFactory, +) + +TICKET_PRICE = 500 +AWAITING_VERIFICATION = {"status": "Approval Pending", "payment_status": "Verification Pending"} + + +class TestOfflineBooking(BookingTestCase): + def setUp(self): + super().setUp() + self.set_event({"apply_tax": 0}) + self.paid_ticket_type = EventTicketTypeFactory.create( + event=self.event.name, prices=[{"currency": "INR", "price": TICKET_PRICE}] + ).name + + def test_offline_booking_cannot_be_submitted_directly(self): + booking = self.create_offline_booking() + + with self.assertRaises(frappe.ValidationError): + booking.submit() + + def test_booking_without_a_payment_method_takes_the_normal_flow(self): + booking = self.create_booking() + + booking.submit() + + self.assertEqual(booking.payment_status, "Unpaid") + + def test_approving_submits_the_booking_and_issues_tickets(self): + booking = self.create_offline_booking(**AWAITING_VERIFICATION) + self.assertEqual((booking.docstatus, self.ticket_count(booking.name)), (0, 0)) + + booking.approve_booking() + booking.reload() + + self.assertEqual((booking.docstatus, booking.status, booking.payment_status), (1, "Approved", "Paid")) + self.assertEqual(self.ticket_count(booking.name), 1) + + def test_rejecting_discards_the_booking_without_tickets(self): + booking = self.create_offline_booking(**AWAITING_VERIFICATION) + + booking.reject_booking() + booking.reload() + + self.assertEqual((booking.docstatus, booking.status), (2, "Rejected")) + self.assertEqual(self.ticket_count(booking.name), 0) + + def test_offline_booking_applies_a_coupon(self): + coupon = BuzzCouponCodeFactory.create(discount_type="Percentage", discount_value=10).name + + booking = self.create_offline_booking(coupon_code=coupon) + + self.assertEqual((booking.net_amount, booking.discount_amount, booking.total_amount), (500, 50, 450)) + + def test_offline_booking_applies_tax(self): + self.set_event({"apply_tax": 1, "tax_inclusive": 0, "tax_label": "GST", "tax_percentage": 18}) + + booking = self.create_offline_booking() + + self.assertEqual((booking.net_amount, booking.tax_percentage), (500, 18)) + self.assertEqual((booking.tax_amount, booking.total_amount), (90, 590)) + + def test_process_booking_leaves_an_offline_booking_in_draft(self): + payload = self.book_offline() + + self.assertTrue(payload.offline_payment) + booking = frappe.get_doc("Event Booking", payload.booking_name) + self.assertEqual(booking.docstatus, 0) + self.assertEqual((booking.status, booking.payment_status), tuple(AWAITING_VERIFICATION.values())) + self.assertEqual(self.ticket_count(booking.name), 0) + + def test_approving_a_booking_from_process_booking_issues_tickets(self): + booking = frappe.get_doc("Event Booking", self.book_offline().booking_name) + + booking.approve_booking() + booking.reload() + + self.assertEqual((booking.docstatus, booking.status, booking.payment_status), (1, "Approved", "Paid")) + self.assertEqual(self.ticket_count(booking.name), 1) + + def create_booking(self, **overrides): + attendees = [ + {"ticket_type": self.paid_ticket_type, "first_name": "Test", "email": "test@example.com"} + ] + return EventBookingFactory.create(event=self.event.name, attendees=attendees, **overrides) + + def create_offline_booking(self, **overrides): + return self.create_booking(payment_method="Offline", **overrides) + + def book_offline(self): + method = str(OfflinePaymentMethodFactory.create(event=self.event.name).name) + attendees = [ + { + "first_name": "Offline", + "email": "offline@example.com", + "ticket_type": str(self.paid_ticket_type), + } + ] + request = self.booking_request(attendees=attendees, is_offline=True, offline_payment_method=method) + return process_booking(request) + + def ticket_count(self, booking: str) -> int: + return frappe.db.count("Event Ticket", {"booking": booking}) diff --git a/buzz/ticketing/doctype/event_booking/test_refund_sync.py b/buzz/ticketing/doctype/event_booking/test_refund_sync.py new file mode 100644 index 00000000..e5f01ae9 --- /dev/null +++ b/buzz/ticketing/doctype/event_booking/test_refund_sync.py @@ -0,0 +1,265 @@ +from unittest.mock import MagicMock, patch + +import frappe +from pydantic import ValidationError + +from buzz.api.checkin import checkin_ticket +from buzz.payments import handle_refund_notification +from buzz.tests.factories import PaymentGatewayFactory +from buzz.ticketing.doctype.event_booking.refund_test_case import ( + CHARGED_PER_TICKET, + GATEWAY_CONTROLLER, + BookingRefundTestCase, +) + +BOOKING_TOTAL = 1100 + + +class TestRefundNotification(BookingRefundTestCase): + def setUp(self): + super().setUp() + self.make_payment() + + def test_a_processed_refund_covering_the_total_marks_the_booking_refunded(self): + self.initiate_refund(BOOKING_TOTAL) + + self.send_refund_webhook(self.refund_id(), "processed", BOOKING_TOTAL) + + self.assertEqual(self.booking.refund_status, "Refunded") + self.assertEqual(self.booking.refunded_amount, BOOKING_TOTAL) + self.assertEqual(self.refunds()[0].status, "Processed") + + def test_a_processed_refund_below_the_total_marks_the_booking_partially_refunded(self): + self.initiate_refund(CHARGED_PER_TICKET) + + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + self.assertEqual(self.booking.refund_status, "Partially Refunded") + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + + def test_a_full_refund_buzz_never_raised_cancels_every_remaining_ticket(self): + # Raised on the Razorpay dashboard: the webhook carries no tickets. + self.send_refund_webhook(self.refund_id(), "processed", BOOKING_TOTAL) + + self.assertEqual(self.booking.refund_status, "Refunded") + self.assertEqual(len(self.refunds()[0].tickets), 2) + self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) + + def test_a_partial_refund_buzz_never_raised_cancels_nothing(self): + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + self.assertEqual(self.refunds()[0].tickets, []) + self.assertEqual(self.cancellation_requests(), []) + + def test_the_same_event_arriving_twice_is_not_counted_twice(self): + self.initiate_refund(CHARGED_PER_TICKET) + + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + self.assertEqual(len(self.refunds()), 1) + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + + def test_a_failed_refund_is_not_counted_and_cancels_nothing(self): + ticket = self.refundable_tickets()[0] + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) + + self.send_refund_webhook(self.refund_id(), "failed", CHARGED_PER_TICKET) + + self.assertEqual(self.refunds()[0].status, "Failed") + self.assertEqual(self.booking.refunded_amount, 0) + self.assertEqual(self.cancellation_requests(), []) + self.assertEqual(frappe.db.get_value("Event Ticket", ticket, "docstatus"), 1) + + def test_a_webhook_processed_as_guest_still_updates_the_booking(self): + # The webhook is an allow_guest endpoint, so its job runs as Guest. + self.initiate_refund(CHARGED_PER_TICKET, tickets=self.refundable_tickets()[:1]) + log = self.create_refund_log(self.refund_payload(self.refund_id(), "processed", CHARGED_PER_TICKET)) + + with self.set_user("Guest"): + self.handle_refund_log(log) + + self.assertEqual(self.refunds()[0].status, "Processed") + self.assertEqual(self.booking.refund_status, "Partially Refunded") + self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) + + def test_a_processed_refund_raises_and_submits_the_cancellation(self): + ticket = self.refundable_tickets()[0] + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) + + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + request = frappe.get_doc("Ticket Cancellation Request", self.refunds()[0].cancellation_request) + self.assertEqual((request.status, request.docstatus), ("Accepted", 1)) + self.assertEqual(frappe.db.get_value("Event Ticket", ticket, "docstatus"), 2) + + def test_a_cancellation_that_cannot_go_through_leaves_the_refund_recorded(self): + ticket = self.refundable_tickets()[0] + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) + + with patch("frappe.sendmail", side_effect=Exception("no outgoing email account")): + self.refunds()[0].apply_gateway_status("processed", CHARGED_PER_TICKET) + + self.booking.reload() + self.assertEqual(self.refunds()[0].status, "Processed") + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + request = frappe.get_doc("Ticket Cancellation Request", self.refunds()[0].cancellation_request) + self.assertEqual((request.status, request.docstatus), ("In Review", 0)) + self.assertEqual(frappe.db.get_value("Event Ticket", ticket, "docstatus"), 1) + + def test_the_integration_request_is_linked_back_to_the_booking(self): + self.initiate_refund(CHARGED_PER_TICKET) + + log = self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + self.assertEqual( + frappe.db.get_value("Integration Request", log, "reference_docname"), self.booking.name + ) + + def test_a_refund_for_an_unknown_payment_is_ignored(self): + payload = self.refund_payload(self.refund_id(), "processed", CHARGED_PER_TICKET) + payload["payload"]["refund"]["entity"]["payment_id"] = "pay_unknown" + + self.assertIsNone(handle_refund_notification("Integration Request", self.create_refund_log(payload))) + self.assertEqual(self.refunds(), []) + + def test_a_payload_without_a_refund_is_refused(self): + log = self.create_refund_log({"event": "refund.processed", "payload": {}}) + + self.assertRaises(ValidationError, handle_refund_notification, "Integration Request", log) + + +class TestSyncRefunds(BookingRefundTestCase): + def setUp(self): + super().setUp() + self.make_payment() + + def test_a_refund_raised_outside_buzz_is_recorded(self): + result = self.sync(self.gateway_refund("processed", CHARGED_PER_TICKET)) + + self.assertEqual(result, {"refunds": 1, "created": 1}) + self.assertEqual([refund.status for refund in self.refunds()], ["Processed"]) + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + self.assertEqual(self.booking.refund_status, "Partially Refunded") + + def test_a_payment_with_no_refunds_records_nothing(self): + result = self.sync() + + self.assertEqual(result, {"refunds": 0, "created": 0}) + self.assertEqual(self.refunds(), []) + self.assertEqual(self.booking.refund_status, "") + + def test_a_full_refund_raised_outside_buzz_cancels_every_remaining_ticket(self): + self.sync(self.gateway_refund("processed", BOOKING_TOTAL)) + + self.assertEqual(self.booking.refund_status, "Refunded") + self.assertEqual(len(self.refunds()[0].tickets), 2) + self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) + + def test_a_partial_refund_raised_outside_buzz_cancels_nothing(self): + self.sync(self.gateway_refund("processed", CHARGED_PER_TICKET)) + + self.assertEqual(self.refunds()[0].tickets, []) + self.assertEqual(self.cancellation_requests(), []) + + def test_a_refund_the_gateway_still_calls_pending_is_initiated(self): + self.sync(self.gateway_refund("pending", BOOKING_TOTAL)) + + self.assertEqual(self.refunds()[0].status, "Initiated") + self.assertEqual(self.booking.refund_status, "Refund Initiated") + self.assertEqual(self.booking.refunded_amount, 0) + self.assertEqual(self.cancellation_requests(), []) + + def test_a_refund_buzz_already_holds_is_updated_not_duplicated(self): + ticket = self.refundable_tickets()[0] + self.initiate_refund(CHARGED_PER_TICKET, tickets=[ticket]) + + result = self.sync(self.gateway_refund("processed", CHARGED_PER_TICKET)) + + self.assertEqual(result, {"refunds": 1, "created": 0}) + self.assertEqual([refund.status for refund in self.refunds()], ["Processed"]) + self.assertEqual([row.ticket for row in self.refunds()[0].tickets], [ticket]) + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + + def test_a_settled_refund_is_not_knocked_back_by_a_stale_pending(self): + self.sync(self.gateway_refund("processed", BOOKING_TOTAL)) + + self.sync(self.gateway_refund("pending", BOOKING_TOTAL)) + + self.assertEqual(self.refunds()[0].status, "Processed") + self.assertEqual(self.booking.refund_status, "Refunded") + self.assertEqual(self.booking.refunded_amount, BOOKING_TOTAL) + self.assertEqual(len(self.cancellation_requests()), 1) + + def test_a_failed_refund_is_not_revived_by_a_later_sighting(self): + self.sync(self.gateway_refund("failed", CHARGED_PER_TICKET)) + + self.sync(self.gateway_refund("processed", CHARGED_PER_TICKET)) + + self.assertEqual(self.refunds()[0].status, "Failed") + self.assertEqual(self.booking.refunded_amount, 0) + + def test_a_refund_the_webhook_recorded_is_not_duplicated_by_a_sync(self): + self.send_refund_webhook(self.refund_id(), "processed", CHARGED_PER_TICKET) + + result = self.sync(self.gateway_refund("processed", CHARGED_PER_TICKET)) + + self.assertEqual(result, {"refunds": 1, "created": 0}) + self.assertEqual(len(self.refunds()), 1) + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + + def test_syncing_twice_does_not_count_the_same_refund_twice(self): + refund = self.gateway_refund("processed", CHARGED_PER_TICKET) + + self.sync(refund) + self.sync(refund) + + self.assertEqual(len(self.refunds()), 1) + self.assertEqual(self.booking.refunded_amount, CHARGED_PER_TICKET) + + def test_a_checked_in_ticket_is_never_cancelled_by_a_synced_refund(self): + checked_in_ticket, other_ticket = self.refundable_tickets() + checkin_ticket(checked_in_ticket) + + self.sync(self.gateway_refund("processed", BOOKING_TOTAL)) + + self.assertEqual([row.ticket for row in self.refunds()[0].tickets], [other_ticket]) + + def test_two_refunds_in_one_batch_cancel_the_tickets_once_they_cover_the_total(self): + # Only the second refund brings the total up to the booking amount. + self.sync( + self.gateway_refund("processed", CHARGED_PER_TICKET, suffix="1"), + self.gateway_refund("processed", CHARGED_PER_TICKET, suffix="2"), + ) + + first, second = self.refunds() + self.assertEqual((first.tickets, len(second.tickets)), ([], 2)) + self.assertEqual(self.booking.refund_status, "Refunded") + self.assertEqual([request.docstatus for request in self.cancellation_requests()], [1]) + + def test_sync_is_refused_when_the_gateway_is_not_razorpay(self): + other_gateway = PaymentGatewayFactory.create().name + frappe.db.set_value("Event Payment", self.payment.name, "payment_gateway", other_gateway) + + with self.assertRaises(frappe.ValidationError) as raised: + self.booking.sync_refunds() + + self.assertIn("Razorpay", str(raised.exception)) + + def test_sync_is_refused_without_a_received_payment(self): + frappe.db.set_value("Event Payment", self.payment.name, "payment_received", 0) + + self.assertRaises(frappe.ValidationError, self.booking.sync_refunds) + + def gateway_refund(self, status: str, amount: float, suffix: str = "1") -> dict: + return {"id": self.refund_id(suffix), "status": status, "amount": int(amount * 100)} + + def sync(self, *gateway_refunds: dict) -> dict: + client = MagicMock() + client.fetch_refunds.return_value = list(gateway_refunds) + + with patch(GATEWAY_CONTROLLER, return_value=client), patch("frappe.sendmail"): + result = self.booking.sync_refunds() + + self.booking.reload() + return result