corporate: Move functions for realm installation data to new file.

Moves two functions in corporate/lib/stripe.py that are used to
get data for the main installation activity analytics page to a
separate file: corporate/lib/analytics.py.

Also, updates these functions for the possibility of realm being
None for a Customer object.

analytics/views/installation_activity.py

@@ -25,9 +25,9 @@ from zerver.lib.request import has_request_variables
 from zerver.models import Realm, get_org_type_display_name
 
 if settings.BILLING_ENABLED:
-    from corporate.lib.stripe import (
+    from corporate.lib.analytics import (
         estimate_annual_recurring_revenue_by_realm,
-        get_realms_to_default_discount_dict,
+        get_realms_with_default_discount_dict,
     )
 
 
@@ -198,7 +198,7 @@ def realm_summary_table() -> str:
     total_arr = 0
     if settings.BILLING_ENABLED:
         estimated_arrs = estimate_annual_recurring_revenue_by_realm()
-        realms_to_default_discount = get_realms_to_default_discount_dict()
+        realms_with_default_discount = get_realms_with_default_discount_dict()
 
         for row in rows:
             row["plan_type_string"] = get_plan_name(row["plan_type"])
@@ -209,14 +209,14 @@ def realm_summary_table() -> str:
                 row["arr"] = estimated_arrs[string_id]
 
             if row["plan_type"] in [Realm.PLAN_TYPE_STANDARD, Realm.PLAN_TYPE_PLUS]:
-                row["effective_rate"] = 100 - int(realms_to_default_discount.get(string_id, 0))
+                row["effective_rate"] = 100 - int(realms_with_default_discount.get(string_id, 0))
             elif row["plan_type"] == Realm.PLAN_TYPE_STANDARD_FREE:
                 row["effective_rate"] = 0
             elif (
                 row["plan_type"] == Realm.PLAN_TYPE_LIMITED
-                and string_id in realms_to_default_discount
+                and string_id in realms_with_default_discount
             ):
-                row["effective_rate"] = 100 - int(realms_to_default_discount[string_id])
+                row["effective_rate"] = 100 - int(realms_with_default_discount[string_id])
             else:
                 row["effective_rate"] = ""
 

corporate/lib/analytics.py

@@ -0,0 +1,39 @@
+from decimal import Decimal
+from typing import Any, Dict
+
+from django.utils.timezone import now as timezone_now
+
+from corporate.lib.stripe import renewal_amount
+from corporate.models import Customer, CustomerPlan
+from zerver.lib.utils import assert_is_not_none
+
+
+def get_realms_with_default_discount_dict() -> Dict[str, Decimal]:
+    realms_with_default_discount: Dict[str, Any] = {}
+    customers = (
+        Customer.objects.exclude(default_discount=None)
+        .exclude(default_discount=0)
+        .exclude(realm=None)
+    )
+    for customer in customers:
+        assert customer.realm is not None
+        realms_with_default_discount[customer.realm.string_id] = assert_is_not_none(
+            customer.default_discount
+        )
+    return realms_with_default_discount
+
+
+def estimate_annual_recurring_revenue_by_realm() -> Dict[str, int]:  # nocoverage
+    annual_revenue = {}
+    for plan in CustomerPlan.objects.filter(status=CustomerPlan.ACTIVE).select_related(
+        "customer__realm"
+    ):
+        if plan.customer.realm is not None:
+            # TODO: figure out what to do for plans that don't automatically
+            # renew, but which probably will renew
+            renewal_cents = renewal_amount(plan, timezone_now())
+            if plan.billing_schedule == CustomerPlan.MONTHLY:
+                renewal_cents *= 12
+            # TODO: Decimal stuff
+            annual_revenue[plan.customer.realm.string_id] = int(renewal_cents / 100)
+    return annual_revenue

corporate/lib/stripe.py

@@ -5,7 +5,7 @@ import secrets
 from datetime import datetime, timedelta
 from decimal import Decimal
 from functools import wraps
-from typing import Any, Callable, Dict, Generator, Optional, Tuple, TypeVar, Union
+from typing import Callable, Dict, Generator, Optional, Tuple, TypeVar, Union
 
 import stripe
 from django.conf import settings
@@ -1057,33 +1057,6 @@ def process_downgrade(plan: CustomerPlan) -> None:
     plan.save(update_fields=["status"])
 
 
-def estimate_annual_recurring_revenue_by_realm() -> Dict[str, int]:  # nocoverage
-    annual_revenue = {}
-    for plan in CustomerPlan.objects.filter(status=CustomerPlan.ACTIVE).select_related(
-        "customer__realm"
-    ):
-        # TODO: figure out what to do for plans that don't automatically
-        # renew, but which probably will renew
-        renewal_cents = renewal_amount(plan, timezone_now())
-        if plan.billing_schedule == CustomerPlan.MONTHLY:
-            renewal_cents *= 12
-        # TODO: Decimal stuff
-        assert plan.customer.realm is not None
-        annual_revenue[plan.customer.realm.string_id] = int(renewal_cents / 100)
-    return annual_revenue
-
-
-def get_realms_to_default_discount_dict() -> Dict[str, Decimal]:
-    realms_to_default_discount: Dict[str, Any] = {}
-    customers = Customer.objects.exclude(default_discount=None).exclude(default_discount=0)
-    for customer in customers:
-        assert customer.realm is not None
-        realms_to_default_discount[customer.realm.string_id] = assert_is_not_none(
-            customer.default_discount
-        )
-    return realms_to_default_discount
-
-
 # During realm deactivation we instantly downgrade the plan to Limited.
 # Extra users added in the final month are not charged. Also used
 # for the cancellation of Free Trial.

corporate/tests/test_stripe.py

@@ -35,6 +35,7 @@ from django.utils.crypto import get_random_string
 from django.utils.timezone import now as timezone_now
 from typing_extensions import ParamSpec, override
 
+from corporate.lib.analytics import get_realms_with_default_discount_dict
 from corporate.lib.stripe import (
     DEFAULT_INVOICE_DAYS_UNTIL_DUE,
     MAX_INVOICED_LICENSES,
@@ -58,7 +59,6 @@ from corporate.lib.stripe import (
     get_latest_seat_count,
     get_plan_renewal_or_end_date,
     get_price_per_license,
-    get_realms_to_default_discount_dict,
     invoice_plan,
     invoice_plans_as_needed,
     is_free_trial_offer_enabled,
@@ -4537,24 +4537,6 @@ class BillingHelpersTest(ZulipTestCase):
         )
         self.assertEqual(get_current_plan_by_realm(realm), plan)
 
-    def test_get_realms_to_default_discount_dict(self) -> None:
-        Customer.objects.create(realm=get_realm("zulip"), stripe_customer_id="cus_1")
-        lear_customer = Customer.objects.create(realm=get_realm("lear"), stripe_customer_id="cus_2")
-        lear_customer.default_discount = Decimal(30)
-        lear_customer.save(update_fields=["default_discount"])
-        zephyr_customer = Customer.objects.create(
-            realm=get_realm("zephyr"), stripe_customer_id="cus_3"
-        )
-        zephyr_customer.default_discount = Decimal(0)
-        zephyr_customer.save(update_fields=["default_discount"])
-
-        self.assertEqual(
-            get_realms_to_default_discount_dict(),
-            {
-                "lear": Decimal("30.0000"),
-            },
-        )
-
     def test_is_realm_on_free_trial(self) -> None:
         realm = get_realm("zulip")
         self.assertFalse(is_realm_on_free_trial(realm))
@@ -4636,6 +4618,37 @@ class BillingHelpersTest(ZulipTestCase):
             )
 
 
+class AnalyticsHelpersTest(ZulipTestCase):
+    def test_get_realms_to_default_discount_dict(self) -> None:
+        Customer.objects.create(realm=get_realm("zulip"), stripe_customer_id="cus_1")
+        lear_customer = Customer.objects.create(realm=get_realm("lear"), stripe_customer_id="cus_2")
+        lear_customer.default_discount = Decimal(30)
+        lear_customer.save(update_fields=["default_discount"])
+        zephyr_customer = Customer.objects.create(
+            realm=get_realm("zephyr"), stripe_customer_id="cus_3"
+        )
+        zephyr_customer.default_discount = Decimal(0)
+        zephyr_customer.save(update_fields=["default_discount"])
+        remote_server = RemoteZulipServer.objects.create(
+            uuid=str(uuid.uuid4()),
+            api_key="magic_secret_api_key",
+            hostname="demo.example.com",
+            contact_email="email@example.com",
+        )
+        remote_customer = Customer.objects.create(
+            remote_server=remote_server, stripe_customer_id="cus_4"
+        )
+        remote_customer.default_discount = Decimal(50)
+        remote_customer.save(update_fields=["default_discount"])
+
+        self.assertEqual(
+            get_realms_with_default_discount_dict(),
+            {
+                "lear": Decimal("30.0000"),
+            },
+        )
+
+
 class LicenseLedgerTest(StripeTestCase):
     def test_add_plan_renewal_if_needed(self) -> None:
         with patch("corporate.lib.stripe.timezone_now", return_value=self.now):