paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-10-25 17:14:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

corporate: Remove '#nocoverage' comments.

Browse Source 
This commit removes '#nocoverage' comment as we have added
few E2E tests in 'test_stripe.py'.
This commit is contained in:
Prakhar Pratyush
2023-12-19 16:54:15 +05:30
committed by Tim Abbott
parent f278e21c7a
commit 63bb63ad80
6 changed files with 65 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
corporate/lib/stripe.py
View File

@@ -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

2
corporate/tests/test_stripe.py
View File

@@ -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

12
corporate/views/billing_page.py
View File

@@ -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,

8
corporate/views/portico.py
View File

@@ -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:

2
corporate/views/session.py
View File

@@ -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
)

12
corporate/views/upgrade.py
View File

@@ -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)