billing: Validate license management at end of billing cycle.

For plans where we're enforcing correct manual license management,
before we create the renewal entry for the licenses ledger at the
end of a billing cycle, we should check that the current count of
billed licenses is not higher than what the renewal license count
would be.

Co-authored-by: Aman Agrawal <amanagr@zulip.com>

corporate/lib/stripe.py

@@ -2186,6 +2186,21 @@ class BillingSession(ABC):
 
         return next_billing_cycle
 
+    def validate_plan_license_management(
+        self, plan: CustomerPlan, renewal_license_count: int
+    ) -> None:
+        if plan.automanage_licenses or plan.customer.exempt_from_license_number_check:
+            return
+
+        # TODO: Enforce manual license management for all paid plans.
+        if plan.tier not in [CustomerPlan.TIER_CLOUD_STANDARD, CustomerPlan.TIER_CLOUD_PLUS]:
+            return  # nocoverage
+
+        if self.current_count_for_billed_licenses() > renewal_license_count:
+            raise BillingError(
+                f"Customer has not manually updated plan for current license count: {plan.customer!s}"
+            )
+
     # event_time should roughly be timezone_now(). Not designed to handle
     # event_times in the past or future
     @transaction.atomic
@@ -2209,6 +2224,7 @@ class BillingSession(ABC):
                 return None, None
 
             if plan.status == CustomerPlan.ACTIVE:
+                self.validate_plan_license_management(plan, licenses_at_next_renewal)
                 return None, LicenseLedger.objects.create(
                     plan=plan,
                     is_renewal=True,
@@ -2245,6 +2261,7 @@ class BillingSession(ABC):
                 )
 
             if plan.status == CustomerPlan.SWITCH_PLAN_TIER_AT_PLAN_END:  # nocoverage
+                self.validate_plan_license_management(plan, licenses_at_next_renewal)
                 plan.status = CustomerPlan.ENDED
                 plan.save(update_fields=["status"])
 
@@ -2266,6 +2283,7 @@ class BillingSession(ABC):
                 )
 
             if plan.status == CustomerPlan.SWITCH_TO_ANNUAL_AT_END_OF_CYCLE:
+                self.validate_plan_license_management(plan, licenses_at_next_renewal)
                 if plan.fixed_price is not None:  # nocoverage
                     raise NotImplementedError("Can't switch fixed priced monthly plan to annual.")
 
@@ -2311,6 +2329,7 @@ class BillingSession(ABC):
                 return new_plan, new_plan_ledger_entry
 
             if plan.status == CustomerPlan.SWITCH_TO_MONTHLY_AT_END_OF_CYCLE:
+                self.validate_plan_license_management(plan, licenses_at_next_renewal)
                 if plan.fixed_price is not None:  # nocoverage
                     raise BillingError("Customer is already on monthly fixed plan.")
 

corporate/tests/test_stripe.py

@@ -5908,6 +5908,44 @@ class InvoiceTest(StripeTestCase):
             "Customer has a paid plan without a Stripe customer ID:",
         )
 
+    @mock_stripe()
+    def test_validate_licenses_for_manual_plan_management(self, *mocks: Mock) -> None:
+        user = self.example_user("hamlet")
+        self.login_user(user)
+
+        # Upgrade with one extra license
+        with (
+            time_machine.travel(self.now, tick=False),
+            patch("corporate.lib.stripe.MIN_INVOICED_LICENSES", 3),
+        ):
+            self.upgrade(invoice=True, licenses=self.seat_count + 1)
+
+        # Set renewal licenses to current seat count
+        with (
+            time_machine.travel(self.now, tick=False),
+            patch("corporate.lib.stripe.MIN_INVOICED_LICENSES", 3),
+        ):
+            result = self.client_billing_patch(
+                "/billing/plan",
+                {"licenses_at_next_renewal": self.seat_count},
+            )
+            self.assert_json_success(result)
+
+        # Add an extra user
+        do_create_user(
+            "email-exra-user",
+            "password-extra-user",
+            get_realm("zulip"),
+            "name-extra-user",
+            acting_user=None,
+        )
+        with self.assertRaises(BillingError) as context:
+            invoice_plans_as_needed(self.next_year)
+        self.assertRegex(
+            context.exception.error_description,
+            "Customer has not manually updated plan for current license count:",
+        )
+
     @mock_stripe()
     def test_invoice_plan(self, *mocks: Mock) -> None:
         user = self.example_user("hamlet")