diff --git a/stripe/__init__.py b/stripe/__init__.py index dac404516..e1a009d83 100644 --- a/stripe/__init__.py +++ b/stripe/__init__.py @@ -349,6 +349,10 @@ def add_beta_version( OAuthErrorObject as OAuthErrorObject, ) from stripe._event import Event as Event + from stripe._event_notification_handler import ( + StripeEventNotificationHandler as StripeEventNotificationHandler, + UnhandledNotificationDetails as UnhandledNotificationDetails, + ) from stripe._event_service import EventService as EventService from stripe._exchange_rate import ExchangeRate as ExchangeRate from stripe._exchange_rate_service import ( @@ -791,6 +795,14 @@ def add_beta_version( "ErrorObject": ("stripe._error_object", False), "OAuthErrorObject": ("stripe._error_object", False), "Event": ("stripe._event", False), + "StripeEventNotificationHandler": ( + "stripe._event_notification_handler", + False, + ), + "UnhandledNotificationDetails": ( + "stripe._event_notification_handler", + False, + ), "EventService": ("stripe._event_service", False), "ExchangeRate": ("stripe._exchange_rate", False), "ExchangeRateService": ("stripe._exchange_rate_service", False), diff --git a/stripe/_event_notification_handler.py b/stripe/_event_notification_handler.py new file mode 100644 index 000000000..58397bfbf --- /dev/null +++ b/stripe/_event_notification_handler.py @@ -0,0 +1,1021 @@ +from dataclasses import dataclass +from typing_extensions import TYPE_CHECKING + +from typing import TypeVar, Callable, List + +# Import at runtime for isinstance check and type annotations +from stripe.v2.core._event import EventNotification, UnknownEventNotification + +if TYPE_CHECKING: + from stripe._stripe_client import StripeClient + + # event-notification-types: The beginning of the section generated from our OpenAPI spec + from stripe.events._v1_billing_meter_error_report_triggered_event import ( + V1BillingMeterErrorReportTriggeredEventNotification, + ) + from stripe.events._v1_billing_meter_no_meter_found_event import ( + V1BillingMeterNoMeterFoundEventNotification, + ) + from stripe.events._v2_core_account_closed_event import ( + V2CoreAccountClosedEventNotification, + ) + from stripe.events._v2_core_account_created_event import ( + V2CoreAccountCreatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_customer_capability_status_updated_event import ( + V2CoreAccountIncludingConfigurationCustomerCapabilityStatusUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_customer_updated_event import ( + V2CoreAccountIncludingConfigurationCustomerUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_merchant_capability_status_updated_event import ( + V2CoreAccountIncludingConfigurationMerchantCapabilityStatusUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_merchant_updated_event import ( + V2CoreAccountIncludingConfigurationMerchantUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_recipient_capability_status_updated_event import ( + V2CoreAccountIncludingConfigurationRecipientCapabilityStatusUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_recipient_updated_event import ( + V2CoreAccountIncludingConfigurationRecipientUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_storer_capability_status_updated_event import ( + V2CoreAccountIncludingConfigurationStorerCapabilityStatusUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_configuration_storer_updated_event import ( + V2CoreAccountIncludingConfigurationStorerUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_defaults_updated_event import ( + V2CoreAccountIncludingDefaultsUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_identity_updated_event import ( + V2CoreAccountIncludingIdentityUpdatedEventNotification, + ) + from stripe.events._v2_core_account_including_requirements_updated_event import ( + V2CoreAccountIncludingRequirementsUpdatedEventNotification, + ) + from stripe.events._v2_core_account_link_returned_event import ( + V2CoreAccountLinkReturnedEventNotification, + ) + from stripe.events._v2_core_account_person_created_event import ( + V2CoreAccountPersonCreatedEventNotification, + ) + from stripe.events._v2_core_account_person_deleted_event import ( + V2CoreAccountPersonDeletedEventNotification, + ) + from stripe.events._v2_core_account_person_updated_event import ( + V2CoreAccountPersonUpdatedEventNotification, + ) + from stripe.events._v2_core_account_updated_event import ( + V2CoreAccountUpdatedEventNotification, + ) + from stripe.events._v2_core_event_destination_ping_event import ( + V2CoreEventDestinationPingEventNotification, + ) + from stripe.events._v2_core_health_event_generation_failure_resolved_event import ( + V2CoreHealthEventGenerationFailureResolvedEventNotification, + ) + from stripe.events._v2_money_management_adjustment_created_event import ( + V2MoneyManagementAdjustmentCreatedEventNotification, + ) + from stripe.events._v2_money_management_financial_account_created_event import ( + V2MoneyManagementFinancialAccountCreatedEventNotification, + ) + from stripe.events._v2_money_management_financial_account_updated_event import ( + V2MoneyManagementFinancialAccountUpdatedEventNotification, + ) + from stripe.events._v2_money_management_financial_address_activated_event import ( + V2MoneyManagementFinancialAddressActivatedEventNotification, + ) + from stripe.events._v2_money_management_financial_address_failed_event import ( + V2MoneyManagementFinancialAddressFailedEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_available_event import ( + V2MoneyManagementInboundTransferAvailableEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_bank_debit_failed_event import ( + V2MoneyManagementInboundTransferBankDebitFailedEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_bank_debit_processing_event import ( + V2MoneyManagementInboundTransferBankDebitProcessingEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_bank_debit_queued_event import ( + V2MoneyManagementInboundTransferBankDebitQueuedEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_bank_debit_returned_event import ( + V2MoneyManagementInboundTransferBankDebitReturnedEventNotification, + ) + from stripe.events._v2_money_management_inbound_transfer_bank_debit_succeeded_event import ( + V2MoneyManagementInboundTransferBankDebitSucceededEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_canceled_event import ( + V2MoneyManagementOutboundPaymentCanceledEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_created_event import ( + V2MoneyManagementOutboundPaymentCreatedEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_failed_event import ( + V2MoneyManagementOutboundPaymentFailedEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_posted_event import ( + V2MoneyManagementOutboundPaymentPostedEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_returned_event import ( + V2MoneyManagementOutboundPaymentReturnedEventNotification, + ) + from stripe.events._v2_money_management_outbound_payment_updated_event import ( + V2MoneyManagementOutboundPaymentUpdatedEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_canceled_event import ( + V2MoneyManagementOutboundTransferCanceledEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_created_event import ( + V2MoneyManagementOutboundTransferCreatedEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_failed_event import ( + V2MoneyManagementOutboundTransferFailedEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_posted_event import ( + V2MoneyManagementOutboundTransferPostedEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_returned_event import ( + V2MoneyManagementOutboundTransferReturnedEventNotification, + ) + from stripe.events._v2_money_management_outbound_transfer_updated_event import ( + V2MoneyManagementOutboundTransferUpdatedEventNotification, + ) + from stripe.events._v2_money_management_payout_method_updated_event import ( + V2MoneyManagementPayoutMethodUpdatedEventNotification, + ) + from stripe.events._v2_money_management_received_credit_available_event import ( + V2MoneyManagementReceivedCreditAvailableEventNotification, + ) + from stripe.events._v2_money_management_received_credit_failed_event import ( + V2MoneyManagementReceivedCreditFailedEventNotification, + ) + from stripe.events._v2_money_management_received_credit_returned_event import ( + V2MoneyManagementReceivedCreditReturnedEventNotification, + ) + from stripe.events._v2_money_management_received_credit_succeeded_event import ( + V2MoneyManagementReceivedCreditSucceededEventNotification, + ) + from stripe.events._v2_money_management_received_debit_canceled_event import ( + V2MoneyManagementReceivedDebitCanceledEventNotification, + ) + from stripe.events._v2_money_management_received_debit_failed_event import ( + V2MoneyManagementReceivedDebitFailedEventNotification, + ) + from stripe.events._v2_money_management_received_debit_pending_event import ( + V2MoneyManagementReceivedDebitPendingEventNotification, + ) + from stripe.events._v2_money_management_received_debit_succeeded_event import ( + V2MoneyManagementReceivedDebitSucceededEventNotification, + ) + from stripe.events._v2_money_management_received_debit_updated_event import ( + V2MoneyManagementReceivedDebitUpdatedEventNotification, + ) + from stripe.events._v2_money_management_transaction_created_event import ( + V2MoneyManagementTransactionCreatedEventNotification, + ) + from stripe.events._v2_money_management_transaction_updated_event import ( + V2MoneyManagementTransactionUpdatedEventNotification, + ) + # event-notification-types: The end of the section generated from our OpenAPI spec + +# internal type to represent any EventNotification subclass +EventNotificationChild = TypeVar( + "EventNotificationChild", bound="EventNotification" +) + + +@dataclass +class UnhandledNotificationDetails: + """ + Information about an unhandled event notification to make it easier to respond (and potentially update your integration). + """ + + is_known_event_type: bool + """ + If true, the unhandled event's type is known to the SDK (i.e., it was successfully deserialized into a specific `EventNotification` subclass). + """ + + +FallbackCallback = Callable[ + [EventNotification, "StripeClient", UnhandledNotificationDetails], None +] +""" +This function is called when no other callback is registered for a given event notification type. +""" + + +class StripeEventNotificationHandler: + def __init__( + self, + client: "StripeClient", + webhook_secret: str, + fallback_callback: FallbackCallback, + ) -> None: + self._registered_handlers = {} + self._client = client + self._webhook_secret = webhook_secret + self.fallback_callback = fallback_callback + # once this is true, adding additional handlers results in an error + self._has_handled_events = False + + def handle(self, webhook_body: str, sig_header: str): + # isn't thread-safe, but we expect these to get registered synchronously at startup + self._has_handled_events = True + + event_notif = self._client.parse_event_notification( + webhook_body, sig_header, self._webhook_secret + ) + + # Create a new client with the event's context. + # This is thread-safe since we're not modifying the original client. + # The new client reuses the HTTP client to avoid TLS handshake overhead. + client_with_event_context = self._client.with_stripe_context( + event_notif.context + ) + + if event_notif.type in self._registered_handlers: + self._registered_handlers[event_notif.type]( + event_notif, client_with_event_context + ) + else: + self.fallback_callback( + event_notif, + client_with_event_context, + UnhandledNotificationDetails( + is_known_event_type=not isinstance( + event_notif, UnknownEventNotification + ) + ), + ) + + def _register( + self, + event_type: str, + func: "Callable[[EventNotificationChild, StripeClient], None]", + ) -> None: + if self._has_handled_events: + raise RuntimeError( + "Cannot register new event handlers after .handle() has been called. This is indicative of a bug." + ) + if event_type in self._registered_handlers: + raise ValueError( + f'Handler for event type "{event_type}" already registered.' + ) + + self._registered_handlers[event_type] = func + + @property + def registered_event_types(self) -> List[str]: + """ + Returns an alphabetized list of all event types that have registered handlers. + """ + return sorted(self._registered_handlers.keys()) + + # event-notification-registration-methods: The beginning of the section generated from our OpenAPI spec + def on_v1_billing_meter_error_report_triggered( + self, + func: "Callable[[V1BillingMeterErrorReportTriggeredEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V1BillingMeterErrorReportTriggeredEvent` (`v1.billing.meter.error_report_triggered`) event notification. + """ + self._register( + "v1.billing.meter.error_report_triggered", + func, + ) + return func + + def on_v1_billing_meter_no_meter_found( + self, + func: "Callable[[V1BillingMeterNoMeterFoundEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V1BillingMeterNoMeterFoundEvent` (`v1.billing.meter.no_meter_found`) event notification. + """ + self._register( + "v1.billing.meter.no_meter_found", + func, + ) + return func + + def on_v2_core_account_closed( + self, + func: "Callable[[V2CoreAccountClosedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountClosedEvent` (`v2.core.account.closed`) event notification. + """ + self._register( + "v2.core.account.closed", + func, + ) + return func + + def on_v2_core_account_created( + self, + func: "Callable[[V2CoreAccountCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountCreatedEvent` (`v2.core.account.created`) event notification. + """ + self._register( + "v2.core.account.created", + func, + ) + return func + + def on_v2_core_account_including_configuration_customer_capability_status_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationCustomerCapabilityStatusUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationCustomerCapabilityStatusUpdatedEvent` (`v2.core.account[configuration.customer].capability_status_updated`) event notification. + """ + self._register( + "v2.core.account[configuration.customer].capability_status_updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_customer_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationCustomerUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationCustomerUpdatedEvent` (`v2.core.account[configuration.customer].updated`) event notification. + """ + self._register( + "v2.core.account[configuration.customer].updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_merchant_capability_status_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationMerchantCapabilityStatusUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationMerchantCapabilityStatusUpdatedEvent` (`v2.core.account[configuration.merchant].capability_status_updated`) event notification. + """ + self._register( + "v2.core.account[configuration.merchant].capability_status_updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_merchant_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationMerchantUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationMerchantUpdatedEvent` (`v2.core.account[configuration.merchant].updated`) event notification. + """ + self._register( + "v2.core.account[configuration.merchant].updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_recipient_capability_status_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationRecipientCapabilityStatusUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationRecipientCapabilityStatusUpdatedEvent` (`v2.core.account[configuration.recipient].capability_status_updated`) event notification. + """ + self._register( + "v2.core.account[configuration.recipient].capability_status_updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_recipient_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationRecipientUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationRecipientUpdatedEvent` (`v2.core.account[configuration.recipient].updated`) event notification. + """ + self._register( + "v2.core.account[configuration.recipient].updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_storer_capability_status_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationStorerCapabilityStatusUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationStorerCapabilityStatusUpdatedEvent` (`v2.core.account[configuration.storer].capability_status_updated`) event notification. + """ + self._register( + "v2.core.account[configuration.storer].capability_status_updated", + func, + ) + return func + + def on_v2_core_account_including_configuration_storer_updated( + self, + func: "Callable[[V2CoreAccountIncludingConfigurationStorerUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingConfigurationStorerUpdatedEvent` (`v2.core.account[configuration.storer].updated`) event notification. + """ + self._register( + "v2.core.account[configuration.storer].updated", + func, + ) + return func + + def on_v2_core_account_including_defaults_updated( + self, + func: "Callable[[V2CoreAccountIncludingDefaultsUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingDefaultsUpdatedEvent` (`v2.core.account[defaults].updated`) event notification. + """ + self._register( + "v2.core.account[defaults].updated", + func, + ) + return func + + def on_v2_core_account_including_identity_updated( + self, + func: "Callable[[V2CoreAccountIncludingIdentityUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingIdentityUpdatedEvent` (`v2.core.account[identity].updated`) event notification. + """ + self._register( + "v2.core.account[identity].updated", + func, + ) + return func + + def on_v2_core_account_including_requirements_updated( + self, + func: "Callable[[V2CoreAccountIncludingRequirementsUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountIncludingRequirementsUpdatedEvent` (`v2.core.account[requirements].updated`) event notification. + """ + self._register( + "v2.core.account[requirements].updated", + func, + ) + return func + + def on_v2_core_account_link_returned( + self, + func: "Callable[[V2CoreAccountLinkReturnedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountLinkReturnedEvent` (`v2.core.account_link.returned`) event notification. + """ + self._register( + "v2.core.account_link.returned", + func, + ) + return func + + def on_v2_core_account_person_created( + self, + func: "Callable[[V2CoreAccountPersonCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountPersonCreatedEvent` (`v2.core.account_person.created`) event notification. + """ + self._register( + "v2.core.account_person.created", + func, + ) + return func + + def on_v2_core_account_person_deleted( + self, + func: "Callable[[V2CoreAccountPersonDeletedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountPersonDeletedEvent` (`v2.core.account_person.deleted`) event notification. + """ + self._register( + "v2.core.account_person.deleted", + func, + ) + return func + + def on_v2_core_account_person_updated( + self, + func: "Callable[[V2CoreAccountPersonUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountPersonUpdatedEvent` (`v2.core.account_person.updated`) event notification. + """ + self._register( + "v2.core.account_person.updated", + func, + ) + return func + + def on_v2_core_account_updated( + self, + func: "Callable[[V2CoreAccountUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreAccountUpdatedEvent` (`v2.core.account.updated`) event notification. + """ + self._register( + "v2.core.account.updated", + func, + ) + return func + + def on_v2_core_event_destination_ping( + self, + func: "Callable[[V2CoreEventDestinationPingEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreEventDestinationPingEvent` (`v2.core.event_destination.ping`) event notification. + """ + self._register( + "v2.core.event_destination.ping", + func, + ) + return func + + def on_v2_core_health_event_generation_failure_resolved( + self, + func: "Callable[[V2CoreHealthEventGenerationFailureResolvedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2CoreHealthEventGenerationFailureResolvedEvent` (`v2.core.health.event_generation_failure.resolved`) event notification. + """ + self._register( + "v2.core.health.event_generation_failure.resolved", + func, + ) + return func + + def on_v2_money_management_adjustment_created( + self, + func: "Callable[[V2MoneyManagementAdjustmentCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementAdjustmentCreatedEvent` (`v2.money_management.adjustment.created`) event notification. + """ + self._register( + "v2.money_management.adjustment.created", + func, + ) + return func + + def on_v2_money_management_financial_account_created( + self, + func: "Callable[[V2MoneyManagementFinancialAccountCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementFinancialAccountCreatedEvent` (`v2.money_management.financial_account.created`) event notification. + """ + self._register( + "v2.money_management.financial_account.created", + func, + ) + return func + + def on_v2_money_management_financial_account_updated( + self, + func: "Callable[[V2MoneyManagementFinancialAccountUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementFinancialAccountUpdatedEvent` (`v2.money_management.financial_account.updated`) event notification. + """ + self._register( + "v2.money_management.financial_account.updated", + func, + ) + return func + + def on_v2_money_management_financial_address_activated( + self, + func: "Callable[[V2MoneyManagementFinancialAddressActivatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementFinancialAddressActivatedEvent` (`v2.money_management.financial_address.activated`) event notification. + """ + self._register( + "v2.money_management.financial_address.activated", + func, + ) + return func + + def on_v2_money_management_financial_address_failed( + self, + func: "Callable[[V2MoneyManagementFinancialAddressFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementFinancialAddressFailedEvent` (`v2.money_management.financial_address.failed`) event notification. + """ + self._register( + "v2.money_management.financial_address.failed", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_available( + self, + func: "Callable[[V2MoneyManagementInboundTransferAvailableEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferAvailableEvent` (`v2.money_management.inbound_transfer.available`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.available", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_bank_debit_failed( + self, + func: "Callable[[V2MoneyManagementInboundTransferBankDebitFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferBankDebitFailedEvent` (`v2.money_management.inbound_transfer.bank_debit_failed`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.bank_debit_failed", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_bank_debit_processing( + self, + func: "Callable[[V2MoneyManagementInboundTransferBankDebitProcessingEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferBankDebitProcessingEvent` (`v2.money_management.inbound_transfer.bank_debit_processing`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.bank_debit_processing", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_bank_debit_queued( + self, + func: "Callable[[V2MoneyManagementInboundTransferBankDebitQueuedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferBankDebitQueuedEvent` (`v2.money_management.inbound_transfer.bank_debit_queued`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.bank_debit_queued", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_bank_debit_returned( + self, + func: "Callable[[V2MoneyManagementInboundTransferBankDebitReturnedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferBankDebitReturnedEvent` (`v2.money_management.inbound_transfer.bank_debit_returned`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.bank_debit_returned", + func, + ) + return func + + def on_v2_money_management_inbound_transfer_bank_debit_succeeded( + self, + func: "Callable[[V2MoneyManagementInboundTransferBankDebitSucceededEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementInboundTransferBankDebitSucceededEvent` (`v2.money_management.inbound_transfer.bank_debit_succeeded`) event notification. + """ + self._register( + "v2.money_management.inbound_transfer.bank_debit_succeeded", + func, + ) + return func + + def on_v2_money_management_outbound_payment_canceled( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentCanceledEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentCanceledEvent` (`v2.money_management.outbound_payment.canceled`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.canceled", + func, + ) + return func + + def on_v2_money_management_outbound_payment_created( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentCreatedEvent` (`v2.money_management.outbound_payment.created`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.created", + func, + ) + return func + + def on_v2_money_management_outbound_payment_failed( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentFailedEvent` (`v2.money_management.outbound_payment.failed`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.failed", + func, + ) + return func + + def on_v2_money_management_outbound_payment_posted( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentPostedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentPostedEvent` (`v2.money_management.outbound_payment.posted`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.posted", + func, + ) + return func + + def on_v2_money_management_outbound_payment_returned( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentReturnedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentReturnedEvent` (`v2.money_management.outbound_payment.returned`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.returned", + func, + ) + return func + + def on_v2_money_management_outbound_payment_updated( + self, + func: "Callable[[V2MoneyManagementOutboundPaymentUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundPaymentUpdatedEvent` (`v2.money_management.outbound_payment.updated`) event notification. + """ + self._register( + "v2.money_management.outbound_payment.updated", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_canceled( + self, + func: "Callable[[V2MoneyManagementOutboundTransferCanceledEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferCanceledEvent` (`v2.money_management.outbound_transfer.canceled`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.canceled", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_created( + self, + func: "Callable[[V2MoneyManagementOutboundTransferCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferCreatedEvent` (`v2.money_management.outbound_transfer.created`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.created", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_failed( + self, + func: "Callable[[V2MoneyManagementOutboundTransferFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferFailedEvent` (`v2.money_management.outbound_transfer.failed`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.failed", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_posted( + self, + func: "Callable[[V2MoneyManagementOutboundTransferPostedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferPostedEvent` (`v2.money_management.outbound_transfer.posted`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.posted", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_returned( + self, + func: "Callable[[V2MoneyManagementOutboundTransferReturnedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferReturnedEvent` (`v2.money_management.outbound_transfer.returned`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.returned", + func, + ) + return func + + def on_v2_money_management_outbound_transfer_updated( + self, + func: "Callable[[V2MoneyManagementOutboundTransferUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementOutboundTransferUpdatedEvent` (`v2.money_management.outbound_transfer.updated`) event notification. + """ + self._register( + "v2.money_management.outbound_transfer.updated", + func, + ) + return func + + def on_v2_money_management_payout_method_updated( + self, + func: "Callable[[V2MoneyManagementPayoutMethodUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementPayoutMethodUpdatedEvent` (`v2.money_management.payout_method.updated`) event notification. + """ + self._register( + "v2.money_management.payout_method.updated", + func, + ) + return func + + def on_v2_money_management_received_credit_available( + self, + func: "Callable[[V2MoneyManagementReceivedCreditAvailableEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedCreditAvailableEvent` (`v2.money_management.received_credit.available`) event notification. + """ + self._register( + "v2.money_management.received_credit.available", + func, + ) + return func + + def on_v2_money_management_received_credit_failed( + self, + func: "Callable[[V2MoneyManagementReceivedCreditFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedCreditFailedEvent` (`v2.money_management.received_credit.failed`) event notification. + """ + self._register( + "v2.money_management.received_credit.failed", + func, + ) + return func + + def on_v2_money_management_received_credit_returned( + self, + func: "Callable[[V2MoneyManagementReceivedCreditReturnedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedCreditReturnedEvent` (`v2.money_management.received_credit.returned`) event notification. + """ + self._register( + "v2.money_management.received_credit.returned", + func, + ) + return func + + def on_v2_money_management_received_credit_succeeded( + self, + func: "Callable[[V2MoneyManagementReceivedCreditSucceededEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedCreditSucceededEvent` (`v2.money_management.received_credit.succeeded`) event notification. + """ + self._register( + "v2.money_management.received_credit.succeeded", + func, + ) + return func + + def on_v2_money_management_received_debit_canceled( + self, + func: "Callable[[V2MoneyManagementReceivedDebitCanceledEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedDebitCanceledEvent` (`v2.money_management.received_debit.canceled`) event notification. + """ + self._register( + "v2.money_management.received_debit.canceled", + func, + ) + return func + + def on_v2_money_management_received_debit_failed( + self, + func: "Callable[[V2MoneyManagementReceivedDebitFailedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedDebitFailedEvent` (`v2.money_management.received_debit.failed`) event notification. + """ + self._register( + "v2.money_management.received_debit.failed", + func, + ) + return func + + def on_v2_money_management_received_debit_pending( + self, + func: "Callable[[V2MoneyManagementReceivedDebitPendingEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedDebitPendingEvent` (`v2.money_management.received_debit.pending`) event notification. + """ + self._register( + "v2.money_management.received_debit.pending", + func, + ) + return func + + def on_v2_money_management_received_debit_succeeded( + self, + func: "Callable[[V2MoneyManagementReceivedDebitSucceededEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedDebitSucceededEvent` (`v2.money_management.received_debit.succeeded`) event notification. + """ + self._register( + "v2.money_management.received_debit.succeeded", + func, + ) + return func + + def on_v2_money_management_received_debit_updated( + self, + func: "Callable[[V2MoneyManagementReceivedDebitUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementReceivedDebitUpdatedEvent` (`v2.money_management.received_debit.updated`) event notification. + """ + self._register( + "v2.money_management.received_debit.updated", + func, + ) + return func + + def on_v2_money_management_transaction_created( + self, + func: "Callable[[V2MoneyManagementTransactionCreatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementTransactionCreatedEvent` (`v2.money_management.transaction.created`) event notification. + """ + self._register( + "v2.money_management.transaction.created", + func, + ) + return func + + def on_v2_money_management_transaction_updated( + self, + func: "Callable[[V2MoneyManagementTransactionUpdatedEventNotification, StripeClient], None]", + ): + """ + Registers a callback for the `V2MoneyManagementTransactionUpdatedEvent` (`v2.money_management.transaction.updated`) event notification. + """ + self._register( + "v2.money_management.transaction.updated", + func, + ) + return func + + # event-notification-registration-methods: The end of the section generated from our OpenAPI spec diff --git a/stripe/_stripe_client.py b/stripe/_stripe_client.py index 7a36fb3f3..2b1e6c6ae 100644 --- a/stripe/_stripe_client.py +++ b/stripe/_stripe_client.py @@ -12,6 +12,10 @@ from stripe._api_mode import ApiMode from stripe._error import AuthenticationError +from stripe._event_notification_handler import ( + StripeEventNotificationHandler, + FallbackCallback, +) from stripe._request_options import extract_options_from_dict from stripe._requestor_options import RequestorOptions, BaseAddresses from stripe._client_options import _ClientOptions @@ -327,6 +331,38 @@ def deserialize( api_mode=api_mode, ) + def with_stripe_context( + self, stripe_context: "Optional[Union[str, StripeContext]]" + ) -> "StripeClient": + """ + Creates a new StripeClient with the same configuration as this client, + but with a different stripe_context. This is useful for handling webhooks + where each event may have its own context. + + The new client reuses the HTTP client from this client to avoid + re-establishing TLS connections. + """ + return StripeClient( + api_key=self._requestor.api_key, # type: ignore + stripe_account=self._requestor._options.stripe_account, + stripe_context=stripe_context, + stripe_version=self._requestor._options.stripe_version, + base_addresses=self._requestor._options.base_addresses, + client_id=self._options.client_id, + max_network_retries=self._requestor._options.max_network_retries, + http_client=self._requestor._client, + ) + + def notification_handler( + self, webhook_secret: str, fallback_callback: FallbackCallback + ) -> StripeEventNotificationHandler: + """ + Returns an StripeEventNotificationHandler instance tied to this client. + """ + return StripeEventNotificationHandler( + self, webhook_secret, fallback_callback + ) + # deprecated v1 services: The beginning of the section generated from our OpenAPI spec @property @deprecated( diff --git a/stripe/events/_event_classes.py b/stripe/events/_event_classes.py index 481aed7bb..b250d1e15 100644 --- a/stripe/events/_event_classes.py +++ b/stripe/events/_event_classes.py @@ -19,9 +19,6 @@ from stripe.events._v2_core_account_created_event import ( V2CoreAccountCreatedEventNotification, ) - from stripe.events._v2_core_account_updated_event import ( - V2CoreAccountUpdatedEventNotification, - ) from stripe.events._v2_core_account_including_configuration_customer_capability_status_updated_event import ( V2CoreAccountIncludingConfigurationCustomerCapabilityStatusUpdatedEventNotification, ) @@ -67,6 +64,9 @@ from stripe.events._v2_core_account_person_updated_event import ( V2CoreAccountPersonUpdatedEventNotification, ) + from stripe.events._v2_core_account_updated_event import ( + V2CoreAccountUpdatedEventNotification, + ) from stripe.events._v2_core_event_destination_ping_event import ( V2CoreEventDestinationPingEventNotification, ) diff --git a/tests/test_event_notification_handler.py b/tests/test_event_notification_handler.py new file mode 100644 index 000000000..568abec09 --- /dev/null +++ b/tests/test_event_notification_handler.py @@ -0,0 +1,555 @@ +import json +import pytest +from typing import Optional +from unittest.mock import Mock + +from stripe import StripeClient +from stripe._event_notification_handler import ( + StripeEventNotificationHandler, + UnhandledNotificationDetails, +) +from stripe._stripe_context import StripeContext +from stripe.events._v1_billing_meter_error_report_triggered_event import ( + V1BillingMeterErrorReportTriggeredEventNotification, +) +from stripe.events._v2_core_account_created_event import ( + V2CoreAccountCreatedEventNotification, +) +from stripe.v2.core._event import EventNotification, UnknownEventNotification +from tests.http_client_mock import HTTPClientMock +from tests.test_webhook import DUMMY_WEBHOOK_SECRET, generate_header + + +class TestEventNotificationHandler: + @pytest.fixture(scope="function") + def stripe_client(self, http_client_mock: HTTPClientMock) -> StripeClient: + return StripeClient( + api_key="sk_test_1234", + stripe_context=StripeContext.parse("original_context_123"), + http_client=http_client_mock.get_mock_http_client(), + ) + + @pytest.fixture(scope="function") + def fallback_callback(self) -> Mock: + """Mock handler for unhandled events""" + return Mock() + + @pytest.fixture(scope="function") + def event_handler( + self, stripe_client: StripeClient, fallback_callback: Mock + ) -> StripeEventNotificationHandler: + return StripeEventNotificationHandler( + client=stripe_client, + webhook_secret=DUMMY_WEBHOOK_SECRET, + fallback_callback=fallback_callback, + ) + + @pytest.fixture(scope="function") + def v1_billing_meter_payload(self) -> str: + """A payload for v1.billing.meter.error_report_triggered event""" + return json.dumps( + { + "id": "evt_123", + "object": "v2.core.event", + "type": "v1.billing.meter.error_report_triggered", + "livemode": False, + "created": "2022-02-15T00:27:45.330Z", + "context": "event_context_456", + "related_object": { + "id": "mtr_123", + "type": "billing.meter", + "url": "/v1/billing/meters/mtr_123", + }, + } + ) + + @pytest.fixture(scope="function") + def v2_account_created_payload(self) -> str: + """A payload for v2.core.account.created event with None context""" + return json.dumps( + { + "id": "evt_789", + "object": "v2.core.event", + "type": "v2.core.account.created", + "livemode": False, + "created": "2022-02-15T00:27:45.330Z", + "context": None, + "related_object": { + "id": "acct_abc", + "type": "account", + "url": "/v2/core/accounts/acct_abc", + }, + } + ) + + @pytest.fixture(scope="function") + def unknown_event_payload(self) -> str: + """A payload for an unknown event type (llama.created)""" + return json.dumps( + { + "id": "evt_unknown", + "object": "v2.core.event", + "type": "llama.created", + "livemode": False, + "created": "2022-02-15T00:27:45.330Z", + "context": "event_context_unknown", + "related_object": { + "id": "llama_123", + "type": "llama", + "url": "/v1/llamas/llama_123", + }, + } + ) + + def test_routes_event_to_registered_handler( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that a registered event type is routed to the correct handler""" + handler = Mock() + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + handler.assert_called_once() + + call_args = handler.call_args[0] + assert isinstance( + call_args[0], V1BillingMeterErrorReportTriggeredEventNotification + ) + + fallback_callback.assert_not_called() + + def test_routes_different_events_to_correct_handlers( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + v2_account_created_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that different event types route to their respective handlers""" + billing_handler = Mock() + account_handler = Mock() + + event_handler.on_v1_billing_meter_error_report_triggered( + billing_handler + ) + event_handler.on_v2_core_account_created(account_handler) + + sig_header1 = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header1) + + sig_header2 = generate_header(payload=v2_account_created_payload) + event_handler.handle(v2_account_created_payload, sig_header2) + + billing_handler.assert_called_once() + account_handler.assert_called_once() + + assert isinstance( + billing_handler.call_args[0][0], + V1BillingMeterErrorReportTriggeredEventNotification, + ) + assert isinstance( + account_handler.call_args[0][0], + V2CoreAccountCreatedEventNotification, + ) + + fallback_callback.assert_not_called() + + def test_handler_receives_correct_runtime_type( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + ) -> None: + """Test that handlers receive the correctly typed event notification""" + received_event: Optional[EventNotification] = None + received_client: Optional[StripeClient] = None + + def handler( + event: V1BillingMeterErrorReportTriggeredEventNotification, + client: StripeClient, + ) -> None: + nonlocal received_event, received_client + received_event = event + received_client = client + + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + assert isinstance( + received_event, V1BillingMeterErrorReportTriggeredEventNotification + ) + assert received_event.type == "v1.billing.meter.error_report_triggered" + assert received_event.id == "evt_123" + assert received_event.related_object.id == "mtr_123" + assert isinstance(received_client, StripeClient) + + def test_cannot_register_handler_after_handling( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + ) -> None: + """Test that registering handlers after handle() raises RuntimeError""" + handler = Mock() + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + with pytest.raises( + RuntimeError, + match="Cannot register new event handlers after .handle\\(\\) has been called", + ): + event_handler.on_v2_core_account_created(Mock()) + + def test_cannot_register_duplicate_handler( + self, event_handler: StripeEventNotificationHandler + ) -> None: + """Test that registering the same event type twice raises ValueError""" + handler1 = Mock() + handler2 = Mock() + + event_handler.on_v1_billing_meter_error_report_triggered(handler1) + + with pytest.raises( + ValueError, + match='Handler for event type "v1.billing.meter.error_report_triggered" already registered', + ): + event_handler.on_v1_billing_meter_error_report_triggered(handler2) + + def test_handler_uses_event_stripe_context( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + stripe_client: StripeClient, + ) -> None: + """Test that the handler receives a client with stripe_context from the event""" + received_context: Optional[StripeContext | str] = None + + def handler( + event: V1BillingMeterErrorReportTriggeredEventNotification, + client: StripeClient, + ) -> None: + nonlocal received_context + received_context = client._requestor._options.stripe_context + + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + assert str(received_context) == "event_context_456" + + def test_stripe_context_restored_after_handler_success( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + stripe_client: StripeClient, + ) -> None: + """Test that the original stripe_context is restored after successful handler execution""" + + def handler( + event: V1BillingMeterErrorReportTriggeredEventNotification, + client: StripeClient, + ) -> None: + assert ( + str(client._requestor._options.stripe_context) + == "event_context_456" + ) + + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + def test_stripe_context_restored_after_handler_error( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + stripe_client: StripeClient, + ) -> None: + """Test that the original stripe_context is restored even when handler raises an exception""" + + def handler( + event: V1BillingMeterErrorReportTriggeredEventNotification, + client: StripeClient, + ) -> None: + assert ( + str(client._requestor._options.stripe_context) + == "event_context_456" + ) + raise RuntimeError("Handler error!") + + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + sig_header = generate_header(payload=v1_billing_meter_payload) + + with pytest.raises(RuntimeError, match="Handler error!"): + event_handler.handle(v1_billing_meter_payload, sig_header) + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + def test_stripe_context_set_to_none_when_event_has_no_context( + self, + event_handler: StripeEventNotificationHandler, + v2_account_created_payload: str, + stripe_client: StripeClient, + ) -> None: + """Test that stripe_context is set to None when event context is None""" + received_context: Optional[StripeContext | str] = None + + def handler( + event: V2CoreAccountCreatedEventNotification, client: StripeClient + ) -> None: + nonlocal received_context + received_context = client._requestor._options.stripe_context + + event_handler.on_v2_core_account_created(handler) + + # Verify we're working with StripeContext instances + assert isinstance( + stripe_client._requestor._options.stripe_context, StripeContext + ) + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + sig_header = generate_header(payload=v2_account_created_payload) + event_handler.handle(v2_account_created_payload, sig_header) + + assert received_context is None + + assert ( + str(stripe_client._requestor._options.stripe_context) + == "original_context_123" + ) + + def test_unknown_event_routes_to_on_unhandled( + self, + event_handler: StripeEventNotificationHandler, + unknown_event_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that events without SDK types route to on_unhandled handler""" + sig_header = generate_header(payload=unknown_event_payload) + + event_handler.handle(unknown_event_payload, sig_header) + + fallback_callback.assert_called_once() + + call_args = fallback_callback.call_args[0] + event_notif = call_args[0] + client = call_args[1] + info = call_args[2] + + assert isinstance(event_notif, UnknownEventNotification) + assert event_notif.type == "llama.created" + assert isinstance(client, StripeClient) + assert isinstance(info, UnhandledNotificationDetails) + assert info.is_known_event_type is False + + def test_known_unregistered_event_routes_to_on_unhandled( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that known event types without a registered handler route to on_unhandled""" + sig_header = generate_header(payload=v1_billing_meter_payload) + + event_handler.handle(v1_billing_meter_payload, sig_header) + + fallback_callback.assert_called_once() + + call_args = fallback_callback.call_args[0] + event_notif = call_args[0] + client = call_args[1] + info = call_args[2] + + assert isinstance( + event_notif, V1BillingMeterErrorReportTriggeredEventNotification + ) + assert event_notif.type == "v1.billing.meter.error_report_triggered" + assert isinstance(client, StripeClient) + assert isinstance(info, UnhandledNotificationDetails) + assert info.is_known_event_type is True + + def test_registered_event_does_not_call_on_unhandled( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that registered events don't trigger on_unhandled""" + handler = Mock() + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + sig_header = generate_header(payload=v1_billing_meter_payload) + event_handler.handle(v1_billing_meter_payload, sig_header) + + handler.assert_called_once() + fallback_callback.assert_not_called() + + def test_handler_client_retains_configuration( + self, + http_client_mock: HTTPClientMock, + fallback_callback: Mock, + v1_billing_meter_payload: str, + ) -> None: + """Test that the client passed to handlers retains all configuration except stripe_context""" + api_key = "sk_test_custom_key" + original_context = "original_context_xyz" + + client = StripeClient( + api_key=api_key, + stripe_context=StripeContext.parse(original_context), + http_client=http_client_mock.get_mock_http_client(), + ) + + notif_handler = StripeEventNotificationHandler( + client=client, + webhook_secret=DUMMY_WEBHOOK_SECRET, + fallback_callback=fallback_callback, + ) + + received_api_key: Optional[str] = None + received_context: Optional[StripeContext | str] = None + + def handler( + event: V1BillingMeterErrorReportTriggeredEventNotification, + client: StripeClient, + ) -> None: + nonlocal received_api_key, received_context + received_api_key = client._requestor.api_key + received_context = client._requestor._options.stripe_context + + notif_handler.on_v1_billing_meter_error_report_triggered(handler) + + sig_header = generate_header(payload=v1_billing_meter_payload) + notif_handler.handle(v1_billing_meter_payload, sig_header) + + assert received_api_key == api_key + assert str(received_context) == "event_context_456" + assert ( + str(client._requestor._options.stripe_context) == original_context + ) + + def test_on_unhandled_receives_correct_info_for_unknown( + self, + event_handler: StripeEventNotificationHandler, + unknown_event_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that on_unhandled receives correct UnhandledNotificationDetails for unknown events""" + sig_header = generate_header(payload=unknown_event_payload) + + event_handler.handle(unknown_event_payload, sig_header) + + fallback_callback.assert_called_once() + info = fallback_callback.call_args[0][2] + + assert isinstance(info, UnhandledNotificationDetails) + assert info.is_known_event_type is False + + def test_on_unhandled_receives_correct_info_for_known_unregistered( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + fallback_callback: Mock, + ) -> None: + """Test that on_unhandled receives correct UnhandledNotificationDetails for known unregistered events""" + sig_header = generate_header(payload=v1_billing_meter_payload) + + event_handler.handle(v1_billing_meter_payload, sig_header) + + fallback_callback.assert_called_once() + info = fallback_callback.call_args[0][2] + + assert isinstance(info, UnhandledNotificationDetails) + assert info.is_known_event_type is True + + def test_validates_webhook_signature( + self, + event_handler: StripeEventNotificationHandler, + v1_billing_meter_payload: str, + ) -> None: + """Test that invalid webhook signatures are rejected""" + from stripe._error import SignatureVerificationError + + with pytest.raises(SignatureVerificationError): + event_handler.handle(v1_billing_meter_payload, "invalid_signature") + + def test_registered_event_types_empty( + self, event_handler: StripeEventNotificationHandler + ) -> None: + """Test that registered_event_types returns empty list when no handlers are registered""" + assert event_handler.registered_event_types == [] + + def test_registered_event_types_single( + self, event_handler: StripeEventNotificationHandler + ) -> None: + """Test that registered_event_types returns a single event type""" + handler = Mock() + event_handler.on_v1_billing_meter_error_report_triggered(handler) + + assert event_handler.registered_event_types == [ + "v1.billing.meter.error_report_triggered" + ] + + def test_registered_event_types_multiple_alphabetized( + self, event_handler: StripeEventNotificationHandler + ) -> None: + """Test that registered_event_types returns multiple event types in alphabetical order""" + handler = Mock() + + # Register in non-alphabetical order + event_handler.on_v2_core_account_updated(handler) + event_handler.on_v1_billing_meter_error_report_triggered(handler) + event_handler.on_v2_core_account_created(handler) + + expected = [ + "v1.billing.meter.error_report_triggered", + "v2.core.account.created", + "v2.core.account.updated", + ] + + assert event_handler.registered_event_types == expected + + def test_can_call_wrapped_functions( + self, event_handler: StripeEventNotificationHandler + ): + @event_handler.on_v1_billing_meter_error_report_triggered # type: ignore + def rand_int(notif, client): + """cool docstring""" + return 4 + + assert rand_int(None, None) == 4 # type: ignore