diff --git a/buzz/api/payments/test_payments.py b/buzz/api/payments/test_payments.py new file mode 100644 index 00000000..f47d0621 --- /dev/null +++ b/buzz/api/payments/test_payments.py @@ -0,0 +1,37 @@ +import frappe +from frappe.tests import IntegrationTestCase + +from buzz.api.payments import get_event_payment_gateways + + +class TestGetEventPaymentGateways(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = str(frappe.get_doc("Buzz Event", {"route": "test-route"}).name) + + def setUp(self): + frappe.set_user("Administrator") + gateway = {"doctype": "Payment Gateway", "gateway": f"Test Gateway {frappe.generate_hash(6)}"} + self.gateway = frappe.get_doc(gateway).insert().name + self.set_gateways([]) + + def tearDown(self): + frappe.db.rollback() + frappe.clear_document_cache("Buzz Event", self.event) + + def set_gateways(self, gateways: list[str]): + event = frappe.get_doc("Buzz Event", self.event) + event.payment_gateways = [] + for gateway in gateways: + event.append("payment_gateways", {"payment_gateway": gateway}) + event.save(ignore_permissions=True) + frappe.clear_document_cache("Buzz Event", self.event) + + def test_configured_gateways_are_returned(self): + self.set_gateways([self.gateway]) + + self.assertEqual(get_event_payment_gateways(self.event), [self.gateway]) + + def test_no_gateways_configured(self): + self.assertEqual(get_event_payment_gateways(self.event), []) diff --git a/buzz/api/sponsorships/__init__.py b/buzz/api/sponsorships/__init__.py index ff08ac21..d59ab1ec 100644 --- a/buzz/api/sponsorships/__init__.py +++ b/buzz/api/sponsorships/__init__.py @@ -1,127 +1,24 @@ import frappe -from buzz.payments import get_payment_link_for_sponsorship +from buzz.api.sponsorships.schemas import SponsorshipDetailsResponse, SponsorshipListItem +from buzz.api.sponsorships.services import SponsorshipService, list_user_enquiries @frappe.whitelist() -def get_sponsorship_details(enquiry_id: str) -> dict: - enquiry = frappe.get_doc("Sponsorship Enquiry", enquiry_id) - - if enquiry.owner != frappe.session.user and not frappe.has_permission( - "Sponsorship Enquiry", "read", enquiry - ): - frappe.throw(frappe._("Not permitted to view this sponsorship enquiry")) - - tier_title = "" - if enquiry.tier: - tier_title = frappe.db.get_value("Sponsorship Tier", enquiry.tier, "title") or enquiry.tier - - event_details = {} - if enquiry.event: - event = frappe.get_cached_doc("Buzz Event", enquiry.event) - event_details = { - "title": event.title, - "short_description": getattr(event, "short_description", ""), - "about": getattr(event, "about", ""), - "start_date": event.start_date, - "end_date": getattr(event, "end_date", ""), - "venue": getattr(event, "venue", ""), - "route": getattr(event, "route", ""), - } - - sponsor_details = None - sponsors = frappe.db.get_all( - "Event Sponsor", - filters={"enquiry": enquiry_id}, - fields=["name", "company_name", "company_logo", "creation", "event", "tier"], - limit=1, - ) - - if sponsors: - sponsor_details = sponsors[0] - if sponsor_details.get("tier"): - sponsor_tier_title = frappe.db.get_value("Sponsorship Tier", sponsor_details["tier"], "title") - sponsor_details["tier_title"] = sponsor_tier_title or sponsor_details["tier"] - - return { - "enquiry": { - "name": enquiry.name, - "company_name": enquiry.company_name, - "company_logo": enquiry.company_logo, - "event": enquiry.event, - "tier": enquiry.tier, - "tier_title": tier_title, - "status": enquiry.status, - "creation": enquiry.creation, - "owner": enquiry.owner, - }, - "event_details": event_details, - "sponsor_details": sponsor_details, - "has_sponsor": bool(sponsor_details), - } +def get_sponsorship_details(enquiry_id: str) -> SponsorshipDetailsResponse: + return SponsorshipService(enquiry_id).details() @frappe.whitelist() -def get_user_sponsorship_inquiries() -> list: - inquiries = frappe.db.get_all( - "Sponsorship Enquiry", - filters={"owner": frappe.session.user}, - fields=["name", "company_name", "event", "tier", "status", "creation"], - order_by="creation desc", - ) - - for inquiry in inquiries: - if inquiry.event: - event_title = frappe.db.get_value("Buzz Event", inquiry.event, "title") - inquiry["event_title"] = event_title - - if inquiry.tier: - tier_title = frappe.db.get_value("Sponsorship Tier", inquiry.tier, "title") - inquiry["tier_title"] = tier_title or inquiry.tier - else: - inquiry["tier_title"] = "" - - inquiry_names = [inquiry.name for inquiry in inquiries] - if inquiry_names: - sponsors = frappe.db.get_all( - "Event Sponsor", - filters={"enquiry": ["in", inquiry_names]}, - fields=["enquiry"], - ) - sponsored_inquiries = {sponsor.enquiry for sponsor in sponsors} - - for inquiry in inquiries: - inquiry["has_sponsor"] = inquiry.name in sponsored_inquiries - else: - for inquiry in inquiries: - inquiry["has_sponsor"] = False - - return inquiries +def get_user_sponsorship_inquiries() -> list[SponsorshipListItem]: + return list_user_enquiries() @frappe.whitelist() def create_sponsorship_payment_link(enquiry_id: str, tier_id: str, payment_gateway: str | None = None) -> str: - enquiry = frappe.get_doc("Sponsorship Enquiry", enquiry_id) - if enquiry.owner != frappe.session.user: - frappe.throw(frappe._("Not permitted to create payment for this enquiry")) - - redirect_url = f"/b/account/sponsorships/{enquiry_id}?success=true" - return get_payment_link_for_sponsorship( - enquiry_id, tier_id, redirect_url, payment_gateway=payment_gateway - ) + return SponsorshipService(enquiry_id).payment_link(tier_id, payment_gateway) @frappe.whitelist() -def withdraw_sponsorship_enquiry(enquiry_id: str): - enquiry = frappe.get_cached_doc("Sponsorship Enquiry", enquiry_id) - if enquiry.owner != frappe.session.user: - frappe.throw(frappe._("Not permitted to withdraw this enquiry")) - - if enquiry.status == "Paid": - frappe.throw(frappe._("Cannot withdraw a paid sponsorship enquiry")) - - if enquiry.status == "Withdrawn": - frappe.throw(frappe._("This sponsorship enquiry has already been withdrawn")) - - enquiry.status = "Withdrawn" - enquiry.save(ignore_permissions=True) +def withdraw_sponsorship_enquiry(enquiry_id: str) -> None: + SponsorshipService(enquiry_id).withdraw() diff --git a/buzz/api/sponsorships/exceptions.py b/buzz/api/sponsorships/exceptions.py new file mode 100644 index 00000000..14fb8bd7 --- /dev/null +++ b/buzz/api/sponsorships/exceptions.py @@ -0,0 +1,25 @@ +from frappe import _lt + +from buzz.api.exceptions import Conflict, NotPermitted + + +class EnquiryNotAccessible(NotPermitted): + message = _lt("You are not permitted to view this sponsorship enquiry.") + + +class PaymentNotPermitted(NotPermitted): + message = _lt("You are not permitted to pay for this sponsorship enquiry.") + + +class WithdrawalNotPermitted(NotPermitted): + message = _lt("You are not permitted to withdraw this sponsorship enquiry.") + + +class EnquiryAlreadyPaid(Conflict): + title = _lt("Already Paid") + message = _lt("A paid sponsorship enquiry cannot be withdrawn.") + + +class EnquiryAlreadyWithdrawn(Conflict): + title = _lt("Already Withdrawn") + message = _lt("This sponsorship enquiry has already been withdrawn.") diff --git a/buzz/api/sponsorships/schemas.py b/buzz/api/sponsorships/schemas.py new file mode 100644 index 00000000..8fd81b8d --- /dev/null +++ b/buzz/api/sponsorships/schemas.py @@ -0,0 +1,54 @@ +from datetime import date, datetime + +from buzz.api.schemas import APIResponse + + +class EnquirySummary(APIResponse): + name: str + company_name: str + company_logo: str | None + event: str + tier: str | None + tier_title: str + status: str + creation: datetime + owner: str + + +class EventSummary(APIResponse): + title: str + short_description: str | None + about: str | None + start_date: date + end_date: date | None + venue: str | None + route: str | None + + +class SponsorSummary(APIResponse): + name: str + company_name: str + company_logo: str | None + creation: datetime + event: str + tier: str + tier_title: str + + +class SponsorshipDetailsResponse(APIResponse): + enquiry: EnquirySummary + event_details: EventSummary + sponsor_details: SponsorSummary | None + has_sponsor: bool + + +class SponsorshipListItem(APIResponse): + name: str + company_name: str + event: str + tier: str | None + status: str + creation: datetime + event_title: str | None + tier_title: str + has_sponsor: bool diff --git a/buzz/api/sponsorships/services.py b/buzz/api/sponsorships/services.py new file mode 100644 index 00000000..4f27f906 --- /dev/null +++ b/buzz/api/sponsorships/services.py @@ -0,0 +1,161 @@ +from typing import TYPE_CHECKING + +import frappe + +from buzz.api.sponsorships.exceptions import ( + EnquiryAlreadyPaid, + EnquiryAlreadyWithdrawn, + EnquiryNotAccessible, + PaymentNotPermitted, + WithdrawalNotPermitted, +) +from buzz.api.sponsorships.schemas import ( + EnquirySummary, + EventSummary, + SponsorshipDetailsResponse, + SponsorshipListItem, + SponsorSummary, +) +from buzz.payments import get_payment_link_for_sponsorship + +if TYPE_CHECKING: + from buzz.events.doctype.buzz_event.buzz_event import BuzzEvent + from buzz.proposals.doctype.sponsorship_enquiry.sponsorship_enquiry import SponsorshipEnquiry + + +class SponsorshipService: + """Read and act on one Sponsorship Enquiry on behalf of its owner.""" + + def __init__(self, enquiry_id: str): + self.enquiry_id = enquiry_id + + @property + def enquiry(self) -> "SponsorshipEnquiry": + return frappe.get_cached_doc("Sponsorship Enquiry", self.enquiry_id) + + @property + def event(self) -> "BuzzEvent": + return frappe.get_cached_doc("Buzz Event", self.enquiry.event) + + @property + def is_owner(self) -> bool: + return self.enquiry.owner == frappe.session.user + + def details(self) -> SponsorshipDetailsResponse: + if not self.is_owner and not frappe.has_permission("Sponsorship Enquiry", "read", self.enquiry): + EnquiryNotAccessible.throw() + + sponsor = self.sponsor() + return SponsorshipDetailsResponse( + enquiry=self.enquiry_summary(), + event_details=self.event_summary(), + sponsor_details=sponsor, + has_sponsor=bool(sponsor), + ) + + def payment_link(self, tier_id: str, payment_gateway: str | None = None) -> str: + if not self.is_owner: + PaymentNotPermitted.throw() + + return get_payment_link_for_sponsorship( + self.enquiry_id, + tier_id, + f"/b/account/sponsorships/{self.enquiry_id}?success=true", + payment_gateway=payment_gateway, + ) + + def withdraw(self) -> None: + if not self.is_owner: + WithdrawalNotPermitted.throw() + + enquiry = self.enquiry + if enquiry.status == "Paid": + EnquiryAlreadyPaid.throw() + if enquiry.status == "Withdrawn": + EnquiryAlreadyWithdrawn.throw() + + enquiry.status = "Withdrawn" + enquiry.save(ignore_permissions=True) + + def enquiry_summary(self) -> EnquirySummary: + enquiry = self.enquiry + return EnquirySummary( + name=enquiry.name, + company_name=enquiry.company_name, + company_logo=enquiry.company_logo, + event=enquiry.event, + tier=enquiry.tier, + tier_title=get_tier_title(enquiry.tier) if enquiry.tier else "", + status=enquiry.status, + creation=enquiry.creation, + owner=enquiry.owner, + ) + + def event_summary(self) -> EventSummary: + event = self.event + return EventSummary( + title=event.title, + short_description=event.short_description, + about=event.about, + start_date=event.start_date, + end_date=event.end_date, + venue=event.venue, + route=event.route, + ) + + def sponsor(self) -> SponsorSummary | None: + sponsors = frappe.db.get_all( + "Event Sponsor", + filters={"enquiry": self.enquiry_id}, + fields=["name", "company_name", "company_logo", "creation", "event", "tier"], + limit=1, + ) + if not sponsors: + return None + + return SponsorSummary(**sponsors[0], tier_title=get_tier_title(sponsors[0].tier)) + + +def list_user_enquiries() -> list[SponsorshipListItem]: + enquiries = frappe.db.get_all( + "Sponsorship Enquiry", + filters={"owner": frappe.session.user}, + fields=["name", "company_name", "event", "tier", "status", "creation"], + order_by="creation desc", + ) + event_titles = get_titles("Buzz Event", {enquiry.event for enquiry in enquiries if enquiry.event}) + tier_titles = get_titles("Sponsorship Tier", {enquiry.tier for enquiry in enquiries if enquiry.tier}) + sponsored = get_sponsored_enquiries([enquiry.name for enquiry in enquiries]) + + return [ + SponsorshipListItem( + **enquiry, + event_title=event_titles.get(enquiry.event), + tier_title=(tier_titles.get(enquiry.tier) or enquiry.tier) if enquiry.tier else "", + has_sponsor=enquiry.name in sponsored, + ) + for enquiry in enquiries + ] + + +def get_titles(doctype: str, names: set[str]) -> dict[str, str]: + if not names: + return {} + + rows = frappe.get_all(doctype, filters={"name": ["in", list(names)]}, fields=["name", "title"]) + # Link fields arrive as strings even where the target autonames to an integer. + return {str(row.name): row.title for row in rows} + + +def get_sponsored_enquiries(enquiry_ids: list[str]) -> set[str]: + if not enquiry_ids: + return set() + + sponsors = frappe.db.get_all( + "Event Sponsor", filters={"enquiry": ["in", enquiry_ids]}, fields=["enquiry"] + ) + return {sponsor.enquiry for sponsor in sponsors} + + +def get_tier_title(tier: str) -> str: + return frappe.db.get_value("Sponsorship Tier", tier, "title") or tier diff --git a/buzz/api/sponsorships/test_sponsorships.py b/buzz/api/sponsorships/test_sponsorships.py new file mode 100644 index 00000000..d574420c --- /dev/null +++ b/buzz/api/sponsorships/test_sponsorships.py @@ -0,0 +1,210 @@ +import frappe +from frappe.tests import IntegrationTestCase + +from buzz.api.sponsorships import ( + create_sponsorship_payment_link, + get_sponsorship_details, + get_user_sponsorship_inquiries, + withdraw_sponsorship_enquiry, +) +from buzz.api.sponsorships.exceptions import ( + EnquiryAlreadyPaid, + EnquiryAlreadyWithdrawn, + EnquiryNotAccessible, + PaymentNotPermitted, + WithdrawalNotPermitted, +) + +ENQUIRY_FIELDS = { + "name", + "company_name", + "company_logo", + "event", + "tier", + "tier_title", + "status", + "creation", + "owner", +} +EVENT_FIELDS = {"title", "short_description", "about", "start_date", "end_date", "venue", "route"} +SPONSOR_FIELDS = {"name", "company_name", "company_logo", "creation", "event", "tier", "tier_title"} +LIST_FIELDS = { + "name", + "company_name", + "event", + "tier", + "status", + "creation", + "event_title", + "tier_title", + "has_sponsor", +} + + +class SponsorshipTestCase(IntegrationTestCase): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.event = str(frappe.get_doc("Buzz Event", {"route": "test-route"}).name) + + def setUp(self): + frappe.set_user("Administrator") + frappe.clear_messages() + + self.tier = frappe.get_doc( + { + "doctype": "Sponsorship Tier", + "event": self.event, + "title": f"Sponsorship Test {frappe.generate_hash(length=6)}", + "price": 5000, + "currency": "INR", + } + ).insert() + + self.enquiry = frappe.get_doc( + { + "doctype": "Sponsorship Enquiry", + "event": self.event, + "tier": self.tier.name, + "company_name": "Acme Corp", + "company_logo": "/files/acme.png", + "status": "Approval Pending", + } + ).insert() + + def tearDown(self): + frappe.set_user("Administrator") + frappe.db.rollback() + + def make_stranger(self) -> str: + email = f"stranger-{frappe.generate_hash(length=6)}@example.com" + user = frappe.new_doc("User") + user.email = email + user.first_name = "Stranger" + user.append("roles", {"role": "Buzz User"}) + user.insert(ignore_permissions=True) + return email + + def make_sponsor(self): + return frappe.get_doc( + { + "doctype": "Event Sponsor", + "event": self.event, + "tier": self.tier.name, + "company_name": "Acme Corp", + "company_logo": "/files/acme.png", + "enquiry": self.enquiry.name, + } + ).insert() + + +class TestGetSponsorshipDetails(SponsorshipTestCase): + def test_response_shape(self): + response = get_sponsorship_details(self.enquiry.name).__json__() + + self.assertEqual(set(response), {"enquiry", "event_details", "sponsor_details", "has_sponsor"}) + self.assertEqual(set(response["enquiry"]), ENQUIRY_FIELDS) + self.assertEqual(set(response["event_details"]), EVENT_FIELDS) + + def test_enquiry_carries_the_tier_title(self): + enquiry = get_sponsorship_details(self.enquiry.name).__json__()["enquiry"] + + self.assertEqual(enquiry["name"], self.enquiry.name) + self.assertEqual(enquiry["tier_title"], self.tier.title) + self.assertEqual(enquiry["owner"], "Administrator") + + def test_no_sponsor_yet(self): + response = get_sponsorship_details(self.enquiry.name).__json__() + + self.assertFalse(response["has_sponsor"]) + self.assertIsNone(response["sponsor_details"]) + + def test_sponsor_details_once_sponsored(self): + sponsor = self.make_sponsor() + response = get_sponsorship_details(self.enquiry.name).__json__() + + self.assertTrue(response["has_sponsor"]) + self.assertEqual(set(response["sponsor_details"]), SPONSOR_FIELDS) + self.assertEqual(response["sponsor_details"]["name"], sponsor.name) + self.assertEqual(response["sponsor_details"]["tier_title"], self.tier.title) + + def test_stranger_is_refused(self): + frappe.set_user(self.make_stranger()) + + with self.assertRaises(EnquiryNotAccessible): + get_sponsorship_details(self.enquiry.name) + + self.assertEqual(frappe.local.message_log[-1]["title"], "Not Permitted") + + def test_unknown_enquiry_raises_does_not_exist(self): + with self.assertRaises(frappe.DoesNotExistError): + get_sponsorship_details("no-such-enquiry") + + def test_status_codes(self): + self.assertEqual(EnquiryNotAccessible.http_status_code, 403) + self.assertEqual(PaymentNotPermitted.http_status_code, 403) + self.assertEqual(WithdrawalNotPermitted.http_status_code, 403) + self.assertEqual(EnquiryAlreadyPaid.http_status_code, 409) + self.assertEqual(EnquiryAlreadyWithdrawn.http_status_code, 409) + + +class TestGetUserSponsorshipInquiries(SponsorshipTestCase): + def test_response_shape(self): + rows = [row.__json__() for row in get_user_sponsorship_inquiries()] + mine = [row for row in rows if row["name"] == self.enquiry.name] + + self.assertEqual(len(mine), 1) + self.assertEqual(set(mine[0]), LIST_FIELDS) + self.assertEqual(mine[0]["tier_title"], self.tier.title) + self.assertEqual(mine[0]["event_title"], frappe.db.get_value("Buzz Event", self.event, "title")) + self.assertFalse(mine[0]["has_sponsor"]) + + def test_has_sponsor_flips_once_sponsored(self): + self.make_sponsor() + rows = {row.name: row for row in get_user_sponsorship_inquiries()} + + self.assertTrue(rows[self.enquiry.name].has_sponsor) + + def test_only_own_enquiries_are_listed(self): + frappe.set_user(self.make_stranger()) + + names = [row.name for row in get_user_sponsorship_inquiries()] + + self.assertNotIn(self.enquiry.name, names) + + +class TestWithdrawSponsorshipEnquiry(SponsorshipTestCase): + def test_withdraw_sets_the_status(self): + withdraw_sponsorship_enquiry(self.enquiry.name) + + self.assertEqual(frappe.db.get_value("Sponsorship Enquiry", self.enquiry.name, "status"), "Withdrawn") + + def test_second_withdrawal_is_rejected(self): + withdraw_sponsorship_enquiry(self.enquiry.name) + frappe.clear_messages() + + with self.assertRaises(EnquiryAlreadyWithdrawn): + withdraw_sponsorship_enquiry(self.enquiry.name) + + self.assertEqual(frappe.local.message_log[-1]["title"], "Already Withdrawn") + + def test_paid_enquiry_cannot_be_withdrawn(self): + frappe.db.set_value("Sponsorship Enquiry", self.enquiry.name, "status", "Paid") + frappe.clear_document_cache("Sponsorship Enquiry", self.enquiry.name) + + with self.assertRaises(EnquiryAlreadyPaid): + withdraw_sponsorship_enquiry(self.enquiry.name) + + def test_stranger_cannot_withdraw(self): + frappe.set_user(self.make_stranger()) + + with self.assertRaises(WithdrawalNotPermitted): + withdraw_sponsorship_enquiry(self.enquiry.name) + + +class TestCreateSponsorshipPaymentLink(SponsorshipTestCase): + def test_stranger_cannot_create_a_payment_link(self): + frappe.set_user(self.make_stranger()) + + with self.assertRaises(PaymentNotPermitted): + create_sponsorship_payment_link(self.enquiry.name, self.tier.name) diff --git a/buzz/payments.py b/buzz/payments.py index 0e78df65..5ba6748b 100644 --- a/buzz/payments.py +++ b/buzz/payments.py @@ -3,29 +3,19 @@ from payments.utils import get_payment_gateway_controller -def get_payment_gateway_for_event(event: str): - return frappe.get_cached_value("Buzz Event", event, "payment_gateway") - - def get_payment_gateways_for_event(event: str) -> list[str]: """Get all payment gateways configured for an event.""" - gateways = frappe.get_all( + return frappe.get_all( "Event Payment Gateway", filters={"parent": event, "parenttype": "Buzz Event"}, pluck="payment_gateway", ) - if not gateways: - # Fallback to legacy field - legacy = frappe.get_cached_value("Buzz Event", event, "payment_gateway") - return [legacy] if legacy else [] - return gateways def get_controller(payment_gateway): return get_payment_gateway_controller(payment_gateway) -@frappe.whitelist() def get_payment_link_for_booking( booking_id: str, redirect_to: str = "/events", payment_gateway: str | None = None ) -> str: @@ -47,7 +37,6 @@ def get_payment_link_for_booking( ) -@frappe.whitelist() def get_payment_link_for_sponsorship( sponsorship_enquiry: str, sponsorship_tier: str,