test(bookings): build booking fixtures with factories and split by behaviour

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)}"}

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"}

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

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)