diff --git a/corporate/lib/stripe.py b/corporate/lib/stripe.py index 8352e02410..7ffa6624f5 100644 --- a/corporate/lib/stripe.py +++ b/corporate/lib/stripe.py @@ -835,9 +835,7 @@ class BillingSession(ABC): assert plan.end_date is not None return plan.end_date.strftime("%B %d, %Y") - def get_legacy_remote_server_next_plan( - self, customer: Customer - ) -> Optional[CustomerPlan]: # nocoverage + def get_legacy_remote_server_next_plan(self, customer: Customer) -> Optional[CustomerPlan]: legacy_plan = self.get_remote_server_legacy_plan( customer, CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END ) @@ -852,9 +850,7 @@ class BillingSession(ABC): status=CustomerPlan.NEVER_STARTED, ) - def get_legacy_remote_server_next_plan_name( - self, customer: Customer - ) -> Optional[str]: # nocoverage + def get_legacy_remote_server_next_plan_name(self, customer: Customer) -> Optional[str]: next_plan = self.get_legacy_remote_server_next_plan(customer) if next_plan is None: return None @@ -1123,9 +1119,7 @@ class BillingSession(ABC): ) return stripe_payment_intent_id - def ensure_current_plan_is_upgradable( - self, customer: Customer, new_plan_tier: int - ) -> None: # nocoverage + def ensure_current_plan_is_upgradable(self, customer: Customer, new_plan_tier: int) -> None: # Upgrade for customers with an existing plan is only supported for remote realm / server right now. if isinstance(self, RealmBillingSession): ensure_customer_does_not_have_active_plan(customer) @@ -1137,7 +1131,7 @@ class BillingSession(ABC): return type_of_plan_change = self.get_type_of_plan_tier_change(plan.tier, new_plan_tier) - if type_of_plan_change != PlanTierChangeType.UPGRADE: + if type_of_plan_change != PlanTierChangeType.UPGRADE: # nocoverage raise InvalidPlanUpgradeError( f"Cannot upgrade from {plan.name} to {CustomerPlan.name_from_tier(new_plan_tier)}" ) @@ -1174,10 +1168,10 @@ class BillingSession(ABC): self.ensure_current_plan_is_upgradable(customer, plan_tier) billing_cycle_anchor = None - if remote_server_legacy_plan is not None: # nocoverage + if remote_server_legacy_plan is not None: # Legacy servers don't get an additional free trial. free_trial = False - if should_schedule_upgrade_for_legacy_remote_server: # nocoverage + if should_schedule_upgrade_for_legacy_remote_server: assert remote_server_legacy_plan is not None billing_cycle_anchor = remote_server_legacy_plan.end_date @@ -1230,7 +1224,7 @@ class BillingSession(ABC): ) event_time = billing_cycle_anchor - if should_schedule_upgrade_for_legacy_remote_server: # nocoverage + if should_schedule_upgrade_for_legacy_remote_server: # In this code path, we are currently on a legacy plan # and are scheduling an upgrade to a non-legacy plan # that should occur when the legacy plan expires. @@ -1244,7 +1238,7 @@ class BillingSession(ABC): stripe_customer = stripe_get_customer(customer.stripe_customer_id) if not stripe_customer_has_credit_card_as_default_payment_method( stripe_customer - ): + ): # nocoverage raise BillingError( "no payment method", _("Please add a credit card to schedule upgrade."), @@ -1486,7 +1480,7 @@ class BillingSession(ABC): ): assert plan.next_invoice_date is not None next_billing_cycle = plan.next_invoice_date - elif plan.status == CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END: # nocoverage + elif plan.status == CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END: assert plan.end_date is not None next_billing_cycle = plan.end_date else: @@ -1656,7 +1650,7 @@ class BillingSession(ABC): return None, None return None, last_ledger_entry - def get_next_plan(self, plan: CustomerPlan) -> Optional[CustomerPlan]: # nocoverage + def get_next_plan(self, plan: CustomerPlan) -> Optional[CustomerPlan]: if plan.status == CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END: assert plan.end_date is not None return CustomerPlan.objects.filter( @@ -1811,7 +1805,7 @@ class BillingSession(ABC): context = self.get_billing_context_from_plan(customer, plan, last_ledger_entry, now) next_plan = self.get_next_plan(plan) - if next_plan is not None: # nocoverage + if next_plan is not None: next_plan_context = self.get_billing_context_from_plan( customer, next_plan, last_ledger_entry, now ) @@ -2009,7 +2003,7 @@ class BillingSession(ABC): if status is not None: if status == CustomerPlan.ACTIVE: assert plan.status < CustomerPlan.LIVE_STATUS_THRESHOLD - with transaction.atomic(): # nocoverage + with transaction.atomic(): # Switch to a different plan was cancelled. We end the next plan # and set the current one as active. if plan.status == CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END: @@ -2623,7 +2617,7 @@ class BillingSession(ABC): self, renewal_date: datetime, end_date: datetime, - ) -> None: # nocoverage + ) -> None: assert not isinstance(self, RealmBillingSession) # Set stripe_customer_id to None to avoid customer being charged without a payment method. customer = self.update_or_create_customer( @@ -2732,7 +2726,7 @@ class BillingSession(ABC): def get_last_ledger_for_automanaged_plan_if_exists( self, - ) -> Optional[LicenseLedger]: # nocoverage + ) -> Optional[LicenseLedger]: customer = self.get_customer() if customer is None: return None @@ -3153,28 +3147,32 @@ class RemoteRealmBillingSession(BillingSession): ) @override - def get_audit_log_event(self, event_type: AuditLogEventType) -> int: # nocoverage + def get_audit_log_event(self, event_type: AuditLogEventType) -> int: if event_type is AuditLogEventType.STRIPE_CUSTOMER_CREATED: return RemoteRealmAuditLog.STRIPE_CUSTOMER_CREATED elif event_type is AuditLogEventType.STRIPE_CARD_CHANGED: return RemoteRealmAuditLog.STRIPE_CARD_CHANGED elif event_type is AuditLogEventType.CUSTOMER_PLAN_CREATED: return RemoteRealmAuditLog.CUSTOMER_PLAN_CREATED - elif event_type is AuditLogEventType.DISCOUNT_CHANGED: + elif event_type is AuditLogEventType.DISCOUNT_CHANGED: # nocoverage return RemoteRealmAuditLog.REMOTE_SERVER_DISCOUNT_CHANGED elif event_type is AuditLogEventType.SPONSORSHIP_APPROVED: return RemoteRealmAuditLog.REMOTE_SERVER_SPONSORSHIP_APPROVED elif event_type is AuditLogEventType.SPONSORSHIP_PENDING_STATUS_CHANGED: return RemoteRealmAuditLog.REMOTE_SERVER_SPONSORSHIP_PENDING_STATUS_CHANGED elif event_type is AuditLogEventType.BILLING_MODALITY_CHANGED: - return RemoteRealmAuditLog.REMOTE_SERVER_BILLING_MODALITY_CHANGED + return RemoteRealmAuditLog.REMOTE_SERVER_BILLING_MODALITY_CHANGED # nocoverage elif event_type is AuditLogEventType.BILLING_ENTITY_PLAN_TYPE_CHANGED: return RemoteRealmAuditLog.REMOTE_SERVER_PLAN_TYPE_CHANGED - elif event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN: + elif ( + event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN + ): # nocoverage return RemoteRealmAuditLog.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN - elif event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN: + elif ( + event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN + ): # nocoverage return RemoteRealmAuditLog.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN - else: + else: # nocoverage raise BillingSessionAuditLogEventError(event_type) @override @@ -3184,7 +3182,7 @@ class RemoteRealmBillingSession(BillingSession): event_time: datetime, *, extra_data: Optional[Dict[str, Any]] = None, - ) -> None: # nocoverage + ) -> None: # These audit logs don't use all the fields of `RemoteRealmAuditLog`: # # * remote_id is None because this is not synced from a remote table. @@ -3208,7 +3206,7 @@ class RemoteRealmBillingSession(BillingSession): RemoteRealmAuditLog.objects.create(**log_data) @override - def get_data_for_stripe_customer(self) -> StripeCustomerData: # nocoverage + def get_data_for_stripe_customer(self) -> StripeCustomerData: # Support requests do not set any stripe billing information. assert self.support_session is False metadata: Dict[str, Any] = {} @@ -3224,7 +3222,7 @@ class RemoteRealmBillingSession(BillingSession): @override def update_data_for_checkout_session_and_payment_intent( self, metadata: Dict[str, Any] - ) -> Dict[str, Any]: # nocoverage + ) -> Dict[str, Any]: assert self.remote_billing_user is not None updated_metadata = dict( remote_realm_user_id=self.remote_billing_user.id, @@ -3237,7 +3235,7 @@ class RemoteRealmBillingSession(BillingSession): @override def update_or_create_customer( self, stripe_customer_id: Optional[str] = None, *, defaults: Optional[Dict[str, Any]] = None - ) -> Customer: # nocoverage + ) -> Customer: if stripe_customer_id is not None: # Support requests do not set any stripe billing information. assert self.support_session is False @@ -3521,9 +3519,7 @@ class RemoteServerBillingSession(BillingSession): return self.remote_billing_user.email @override - def current_count_for_billed_licenses( - self, event_time: datetime = timezone_now() - ) -> int: # nocoverage + def current_count_for_billed_licenses(self, event_time: datetime = timezone_now()) -> int: if has_stale_audit_log(self.remote_server): raise MissingDataError remote_server_counts = get_remote_server_guest_and_non_guest_count( @@ -3543,7 +3539,7 @@ class RemoteServerBillingSession(BillingSession): ) @override - def get_audit_log_event(self, event_type: AuditLogEventType) -> int: # nocoverage + def get_audit_log_event(self, event_type: AuditLogEventType) -> int: if event_type is AuditLogEventType.STRIPE_CUSTOMER_CREATED: return RemoteZulipServerAuditLog.STRIPE_CUSTOMER_CREATED elif event_type is AuditLogEventType.STRIPE_CARD_CHANGED: @@ -3551,20 +3547,24 @@ class RemoteServerBillingSession(BillingSession): elif event_type is AuditLogEventType.CUSTOMER_PLAN_CREATED: return RemoteZulipServerAuditLog.CUSTOMER_PLAN_CREATED elif event_type is AuditLogEventType.DISCOUNT_CHANGED: - return RemoteZulipServerAuditLog.REMOTE_SERVER_DISCOUNT_CHANGED + return RemoteZulipServerAuditLog.REMOTE_SERVER_DISCOUNT_CHANGED # nocoverage elif event_type is AuditLogEventType.SPONSORSHIP_APPROVED: return RemoteZulipServerAuditLog.REMOTE_SERVER_SPONSORSHIP_APPROVED elif event_type is AuditLogEventType.SPONSORSHIP_PENDING_STATUS_CHANGED: return RemoteZulipServerAuditLog.REMOTE_SERVER_SPONSORSHIP_PENDING_STATUS_CHANGED elif event_type is AuditLogEventType.BILLING_MODALITY_CHANGED: - return RemoteZulipServerAuditLog.REMOTE_SERVER_BILLING_MODALITY_CHANGED + return RemoteZulipServerAuditLog.REMOTE_SERVER_BILLING_MODALITY_CHANGED # nocoverage elif event_type is AuditLogEventType.BILLING_ENTITY_PLAN_TYPE_CHANGED: return RemoteZulipServerAuditLog.REMOTE_SERVER_PLAN_TYPE_CHANGED - elif event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN: + elif ( + event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN + ): # nocoverage return RemoteZulipServerAuditLog.CUSTOMER_SWITCHED_FROM_MONTHLY_TO_ANNUAL_PLAN - elif event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN: + elif ( + event_type is AuditLogEventType.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN + ): # nocoverage return RemoteZulipServerAuditLog.CUSTOMER_SWITCHED_FROM_ANNUAL_TO_MONTHLY_PLAN - else: + else: # nocoverage raise BillingSessionAuditLogEventError(event_type) @override @@ -3574,7 +3574,7 @@ class RemoteServerBillingSession(BillingSession): event_time: datetime, *, extra_data: Optional[Dict[str, Any]] = None, - ) -> None: # nocoverage + ) -> None: audit_log_event = self.get_audit_log_event(event_type) log_data = { "server": self.remote_server, @@ -3592,7 +3592,7 @@ class RemoteServerBillingSession(BillingSession): RemoteZulipServerAuditLog.objects.create(**log_data) @override - def get_data_for_stripe_customer(self) -> StripeCustomerData: # nocoverage + def get_data_for_stripe_customer(self) -> StripeCustomerData: # Support requests do not set any stripe billing information. assert self.support_session is False metadata: Dict[str, Any] = {} @@ -3608,7 +3608,7 @@ class RemoteServerBillingSession(BillingSession): @override def update_data_for_checkout_session_and_payment_intent( self, metadata: Dict[str, Any] - ) -> Dict[str, Any]: # nocoverage + ) -> Dict[str, Any]: assert self.remote_billing_user is not None updated_metadata = dict( remote_server_user_id=self.remote_billing_user.id, @@ -3621,7 +3621,7 @@ class RemoteServerBillingSession(BillingSession): @override def update_or_create_customer( self, stripe_customer_id: Optional[str] = None, *, defaults: Optional[Dict[str, Any]] = None - ) -> Customer: # nocoverage + ) -> Customer: if stripe_customer_id is not None: # Support requests do not set any stripe billing information. assert self.support_session is False @@ -3638,9 +3638,7 @@ class RemoteServerBillingSession(BillingSession): @override @transaction.atomic - def do_change_plan_type( - self, *, tier: Optional[int], is_sponsored: bool = False - ) -> None: # nocoverage + def do_change_plan_type(self, *, tier: Optional[int], is_sponsored: bool = False) -> None: # This function needs to translate between the different # formats of CustomerPlan.tier and RealmZulipServer.plan_type. if is_sponsored: @@ -3833,7 +3831,7 @@ class RemoteServerBillingSession(BillingSession): self.remote_server.save(update_fields=["org_type"]) @override - def sync_license_ledger_if_needed(self) -> None: # nocoverage + def sync_license_ledger_if_needed(self) -> None: last_ledger = self.get_last_ledger_for_automanaged_plan_if_exists() if last_ledger is None: return @@ -3854,7 +3852,7 @@ class RemoteServerBillingSession(BillingSession): end_of_cycle_plan = self.update_license_ledger_for_automanaged_plan( current_plan, audit_log.event_time ) - if end_of_cycle_plan is None: + if end_of_cycle_plan is None: # nocoverage return current_plan = end_of_cycle_plan diff --git a/corporate/tests/test_stripe.py b/corporate/tests/test_stripe.py index c2cd404dae..bc0b0eaf44 100644 --- a/corporate/tests/test_stripe.py +++ b/corporate/tests/test_stripe.py @@ -739,7 +739,7 @@ class StripeTestCase(ZulipTestCase): def client_billing_patch(self, url_suffix: str, info: Mapping[str, Any] = {}) -> Any: url = f"/json{self.billing_session.billing_base_url}" + url_suffix if self.billing_session.billing_base_url: - response = self.client_patch(url, info, subdomain="selfhosting") # nocoverage + response = self.client_patch(url, info, subdomain="selfhosting") else: response = self.client_patch(url, info) return response diff --git a/corporate/views/billing_page.py b/corporate/views/billing_page.py index 48ad663f6e..0a45f49d0b 100644 --- a/corporate/views/billing_page.py +++ b/corporate/views/billing_page.py @@ -158,7 +158,7 @@ def remote_server_billing_page( billing_session: RemoteServerBillingSession, *, success_message: str = "", -) -> HttpResponse: # nocoverage +) -> HttpResponse: context: Dict[str, Any] = { # We wouldn't be here if user didn't have access. "admin_access": billing_session.has_billing_access(), @@ -167,7 +167,9 @@ def remote_server_billing_page( "billing_base_url": billing_session.billing_base_url, } - if billing_session.remote_server.plan_type == RemoteZulipServer.PLAN_TYPE_COMMUNITY: + if ( + billing_session.remote_server.plan_type == RemoteZulipServer.PLAN_TYPE_COMMUNITY + ): # nocoverage return HttpResponseRedirect( reverse( "remote_server_sponsorship_page", @@ -189,7 +191,7 @@ def remote_server_billing_page( ) ) # If the realm is on a paid plan, show the sponsorship pending message - context["sponsorship_pending"] = True + context["sponsorship_pending"] = True # nocoverage if ( customer is None @@ -212,7 +214,7 @@ def remote_server_billing_page( try: main_context = billing_session.get_billing_page_context() - except MissingDataError: + except MissingDataError: # nocoverage return billing_session.missing_data_error_page(request) if main_context: @@ -292,7 +294,7 @@ def update_plan_for_remote_server( "licenses_at_next_renewal", json_validator=check_int, default=None ), schedule: Optional[int] = REQ("schedule", json_validator=check_int, default=None), -) -> HttpResponse: # nocoverage +) -> HttpResponse: update_plan_request = UpdatePlanRequest( status=status, licenses=licenses, diff --git a/corporate/views/portico.py b/corporate/views/portico.py index a59eae53ac..ca947a3a82 100644 --- a/corporate/views/portico.py +++ b/corporate/views/portico.py @@ -135,7 +135,7 @@ def plans_view(request: HttpRequest) -> HttpResponse: @authenticated_remote_realm_management_endpoint def remote_realm_plans_page( request: HttpRequest, billing_session: RemoteRealmBillingSession -) -> HttpResponse: # nocoverage +) -> HttpResponse: customer = billing_session.get_customer() context = PlansPageContext( is_self_hosted_realm=True, @@ -149,7 +149,7 @@ def remote_realm_plans_page( ) context.on_free_tier = customer is None and not context.is_sponsored - if customer is not None: + if customer is not None: # nocoverage context.sponsorship_pending = customer.sponsorship_pending context.customer_plan = get_current_plan_by_customer(customer) if context.customer_plan is None: @@ -192,7 +192,7 @@ def remote_realm_plans_page( @authenticated_remote_server_management_endpoint def remote_server_plans_page( request: HttpRequest, billing_session: RemoteServerBillingSession -) -> HttpResponse: # nocoverage +) -> HttpResponse: customer = billing_session.get_customer() context = PlansPageContext( is_self_hosted_realm=True, @@ -206,7 +206,7 @@ def remote_server_plans_page( ) context.on_free_tier = customer is None and not context.is_sponsored - if customer is not None: + if customer is not None: # nocoverage context.sponsorship_pending = customer.sponsorship_pending context.customer_plan = get_current_plan_by_customer(customer) if context.customer_plan is None: diff --git a/corporate/views/session.py b/corporate/views/session.py index 134abe7cbc..ec98e5caa1 100644 --- a/corporate/views/session.py +++ b/corporate/views/session.py @@ -97,7 +97,7 @@ def start_card_update_stripe_session_for_remote_server_upgrade( *, manual_license_management: Json[bool] = False, tier: Json[int], -) -> HttpResponse: # nocoverage +) -> HttpResponse: session_data = billing_session.create_card_update_session_for_upgrade( manual_license_management, tier ) diff --git a/corporate/views/upgrade.py b/corporate/views/upgrade.py index da0e0b80b5..2f648fc4af 100644 --- a/corporate/views/upgrade.py +++ b/corporate/views/upgrade.py @@ -147,7 +147,7 @@ def remote_server_upgrade( licenses: Optional[int] = REQ(json_validator=check_int, default=None), remote_server_plan_start_date: Optional[str] = REQ(default=None), tier: int = REQ(default=CustomerPlan.TIER_SELF_HOSTED_BUSINESS, json_validator=check_int), -) -> HttpResponse: # nocoverage +) -> HttpResponse: try: upgrade_request = UpgradeRequest( billing_modality=billing_modality, @@ -161,7 +161,7 @@ def remote_server_upgrade( ) data = billing_session.do_upgrade(upgrade_request) return json_success(request, data) - except BillingError as e: + except BillingError as e: # nocoverage billing_logger.warning( "BillingError during upgrade: %s. remote_server=%s (%s), billing_modality=%s, " "schedule=%s, license_management=%s, licenses=%s", @@ -174,7 +174,7 @@ def remote_server_upgrade( licenses, ) raise e - except Exception: + except Exception: # nocoverage billing_logger.exception("Uncaught exception in billing:", stack_info=True) error_message = BillingError.CONTACT_SUPPORT.format(email=settings.ZULIP_ADMINISTRATOR) error_description = "uncaught exception during upgrade" @@ -244,7 +244,7 @@ def remote_server_upgrade_page( manual_license_management: Json[bool] = False, success_message: str = "", tier: str = str(CustomerPlan.TIER_SELF_HOSTED_BUSINESS), -) -> HttpResponse: # nocoverage +) -> HttpResponse: initial_upgrade_request = InitialUpgradeRequest( manual_license_management=manual_license_management, tier=int(tier), @@ -252,10 +252,10 @@ def remote_server_upgrade_page( ) try: redirect_url, context = billing_session.get_initial_upgrade_context(initial_upgrade_request) - except MissingDataError: + except MissingDataError: # nocoverage return billing_session.missing_data_error_page(request) - if redirect_url: + if redirect_url: # nocoverage return HttpResponseRedirect(redirect_url) response = render(request, "corporate/upgrade.html", context=context)