stripe: Add 'get_event_status' method to the 'BillingSession' class.

This commit refactors 'event_status' view and adds
'BillingSession.get_event_status' method.

This refactoring will help in minimizing duplicate code
while supporting both realm and remote_server customers.

corporate/lib/stripe.py

@@ -493,6 +493,12 @@ class UpdatePlanRequest:
     licenses_at_next_renewal: Optional[int]
 
 
+@dataclass
+class EventStatusRequest:
+    stripe_session_id: Optional[str]
+    stripe_payment_intent_id: Optional[str]
+
+
 class AuditLogEventType(Enum):
     STRIPE_CUSTOMER_CREATED = 1
     STRIPE_CARD_CHANGED = 2
@@ -620,6 +626,10 @@ class BillingSession(ABC):
     def is_valid_plan_tier_switch(self, current_plan_tier: int, new_plan_tier: int) -> bool:
         pass
 
+    @abstractmethod
+    def has_billing_access(self) -> bool:
+        pass
+
     @catch_stripe_errors
     def create_stripe_customer(self) -> Customer:
         stripe_customer_data = self.get_data_for_stripe_customer()
@@ -1518,6 +1528,41 @@ class BillingSession(ABC):
         assert new_plan is not None  # for mypy
         invoice_plan(new_plan, plan_switch_time)
 
+    def get_event_status(self, event_status_request: EventStatusRequest) -> Dict[str, Any]:
+        customer = self.get_customer()
+
+        if customer is None:
+            raise JsonableError(_("No customer for this organization!"))
+
+        stripe_session_id = event_status_request.stripe_session_id
+        if stripe_session_id is not None:
+            try:
+                session = Session.objects.get(
+                    stripe_session_id=stripe_session_id, customer=customer
+                )
+            except Session.DoesNotExist:
+                raise JsonableError(_("Session not found"))
+
+            if (
+                session.type == Session.CARD_UPDATE_FROM_BILLING_PAGE
+                and not self.has_billing_access()
+            ):
+                raise JsonableError(_("Must be a billing administrator or an organization owner"))
+            return {"session": session.to_dict()}
+
+        stripe_payment_intent_id = event_status_request.stripe_payment_intent_id
+        if stripe_payment_intent_id is not None:
+            payment_intent = PaymentIntent.objects.filter(
+                stripe_payment_intent_id=stripe_payment_intent_id,
+                customer=customer,
+            ).last()
+
+            if payment_intent is None:
+                raise JsonableError(_("Payment intent not found"))
+            return {"payment_intent": payment_intent.to_dict()}
+
+        raise JsonableError(_("Pass stripe_session_id or stripe_payment_intent_id"))
+
 
 class RealmBillingSession(BillingSession):
     def __init__(
@@ -1758,6 +1803,11 @@ class RealmBillingSession(BillingSession):
             assert current_plan_tier == CustomerPlan.PLUS
             return new_plan_tier == CustomerPlan.STANDARD
 
+    @override
+    def has_billing_access(self) -> bool:
+        assert self.user is not None
+        return self.user.has_billing_access
+
 
 class RemoteRealmBillingSession(BillingSession):  # nocoverage
     def __init__(
@@ -1951,6 +2001,12 @@ class RemoteRealmBillingSession(BillingSession):  # nocoverage
         # TBD
         return False
 
+    @override
+    def has_billing_access(self) -> bool:
+        # We don't currently have a way to authenticate a remote
+        # session that isn't authorized for billing access.
+        return True
+
 
 class RemoteServerBillingSession(BillingSession):  # nocoverage
     """Billing session for pre-8.0 servers that do not yet support
@@ -2137,6 +2193,12 @@ class RemoteServerBillingSession(BillingSession):  # nocoverage
         # TBD
         return False
 
+    @override
+    def has_billing_access(self) -> bool:
+        # We don't currently have a way to authenticate a remote
+        # session that isn't authorized for billing access.
+        return True
+
 
 def stripe_customer_has_credit_card_as_default_payment_method(
     stripe_customer: stripe.Customer,

corporate/views/event_status.py

@@ -3,11 +3,9 @@ from typing import Optional
 
 from django.http import HttpRequest, HttpResponse
 from django.shortcuts import render
-from django.utils.translation import gettext as _
 
-from corporate.models import PaymentIntent, Session, get_customer_by_realm
+from corporate.lib.stripe import EventStatusRequest, RealmBillingSession
 from zerver.decorator import require_organization_member, zulip_login_required
-from zerver.lib.exceptions import JsonableError
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
 from zerver.models import UserProfile
@@ -23,30 +21,12 @@ def event_status(
     stripe_session_id: Optional[str] = REQ(default=None),
     stripe_payment_intent_id: Optional[str] = REQ(default=None),
 ) -> HttpResponse:
-    customer = get_customer_by_realm(user.realm)
-    if customer is None:
-        raise JsonableError(_("No customer for this organization!"))
-
-    if stripe_session_id is not None:
-        try:
-            session = Session.objects.get(stripe_session_id=stripe_session_id, customer=customer)
-        except Session.DoesNotExist:
-            raise JsonableError(_("Session not found"))
-
-        if session.type == Session.CARD_UPDATE_FROM_BILLING_PAGE and not user.has_billing_access:
-            raise JsonableError(_("Must be a billing administrator or an organization owner"))
-        return json_success(request, data={"session": session.to_dict()})
-
-    if stripe_payment_intent_id is not None:
-        payment_intent = PaymentIntent.objects.filter(
-            stripe_payment_intent_id=stripe_payment_intent_id,
-            customer=customer,
-        ).last()
-
-        if payment_intent is None:
-            raise JsonableError(_("Payment intent not found"))
-        return json_success(request, data={"payment_intent": payment_intent.to_dict()})
-    raise JsonableError(_("Pass stripe_session_id or stripe_payment_intent_id"))
+    event_status_request = EventStatusRequest(
+        stripe_session_id=stripe_session_id, stripe_payment_intent_id=stripe_payment_intent_id
+    )
+    billing_session = RealmBillingSession(user)
+    data = billing_session.get_event_status(event_status_request)
+    return json_success(request, data)
 
 
 @zulip_login_required