feat: make CheckoutIntent.ssp_product non-nullable

Author: vshaikismail-sonata

enterprise_access/apps/api/serializers/customer_billing.py

@@ -193,6 +193,12 @@ class CheckoutIntentCreateRequestSerializer(CountryFieldMixin, serializers.Model
     class Meta:
         model = CheckoutIntent
         fields = '__all__'
+        extra_kwargs = {
+            'ssp_product': {
+                'required': False,
+                'allow_null': False,
+            },
+        }
         read_only_fields = [
             field.name for field in CheckoutIntent._meta.get_fields()
             if field.name not in [
@@ -201,6 +207,7 @@ class Meta:
                 'quantity',
                 'country',
                 'terms_metadata',
+                'ssp_product',
             ]
         ]
 
@@ -237,13 +244,18 @@ def create(self, validated_data):
         Creates a new CheckoutIntent.
         """
         try:
+            # Pass through any provided `ssp_product`; model-level creation
+            # resolves a default when omitted for backwards compatibility.
+            ssp_product = validated_data.get('ssp_product')
+
             return CheckoutIntent.create_intent(
                 user=self.context['request'].user,
                 quantity=validated_data['quantity'],
                 slug=validated_data.get('enterprise_slug'),
                 name=validated_data.get('enterprise_name'),
                 country=validated_data.get('country'),
                 terms_metadata=validated_data.get('terms_metadata'),
+                ssp_product=ssp_product,
             )
 
         # Catch exceptions that should return 422:

enterprise_access/apps/api/tests/test_serializers.py

@@ -11,6 +11,10 @@
 from django.test import TestCase
 from freezegun import freeze_time
 
+from enterprise_access.apps.api.serializers.customer_billing import (
+    CheckoutIntentCreateRequestSerializer,
+    RecordConflictError
+)
 from enterprise_access.apps.api.serializers.subsidy_access_policy import (
     SubsidyAccessPolicyAggregatesSerializer,
     SubsidyAccessPolicyCreditsAvailableResponseSerializer,
@@ -24,6 +28,7 @@
     AssignmentConfigurationFactory,
     LearnerContentAssignmentFactory
 )
+from enterprise_access.apps.customer_billing.models import FailedCheckoutIntentConflict, SlugReservationConflict
 from enterprise_access.apps.subsidy_access_policy.tests.factories import (
     AssignedLearnerCreditAccessPolicyFactory,
     PerLearnerEnrollmentCapLearnerCreditAccessPolicyFactory
@@ -574,3 +579,95 @@ def test_missing_uuids_field(self):
         serializer = LearnerCreditRequestBulkCancelSerializer(data=data)
         self.assertFalse(serializer.is_valid())
         self.assertIn('learner_credit_request_uuids', serializer.errors)
+
+
+class TestCheckoutIntentCreateRequestSerializer(TestCase):
+    """Tests for CheckoutIntentCreateRequestSerializer behavior."""
+
+    def setUp(self):
+        self.user = mock.Mock()
+        self.request = mock.Mock(user=self.user)
+
+    def _make_serializer(self):
+        return CheckoutIntentCreateRequestSerializer(context={'request': self.request})
+
+    @mock.patch('enterprise_access.apps.api.serializers.customer_billing.CheckoutIntent.create_intent')
+    def test_create_passes_none_ssp_product_when_omitted(self, mock_create_intent):
+        """Serializer create should pass ssp_product=None when not supplied."""
+        serializer = self._make_serializer()
+        mock_create_intent.return_value = mock.Mock()
+
+        serializer.create({
+            'quantity': 10,
+            'enterprise_slug': 'acme-enterprise',
+            'enterprise_name': 'Acme Enterprise',
+        })
+
+        mock_create_intent.assert_called_once_with(
+            user=self.user,
+            quantity=10,
+            slug='acme-enterprise',
+            name='Acme Enterprise',
+            country=None,
+            terms_metadata=None,
+            ssp_product=None,
+        )
+
+    @mock.patch('enterprise_access.apps.api.serializers.customer_billing.CheckoutIntent.create_intent')
+    def test_create_passes_explicit_ssp_product(self, mock_create_intent):
+        """Serializer create should pass through explicitly provided ssp_product."""
+        serializer = self._make_serializer()
+        mock_create_intent.return_value = mock.Mock()
+        product = mock.Mock()
+
+        serializer.create({
+            'quantity': 10,
+            'enterprise_slug': 'acme-enterprise',
+            'enterprise_name': 'Acme Enterprise',
+            'ssp_product': product,
+        })
+
+        mock_create_intent.assert_called_once()
+        self.assertEqual(mock_create_intent.call_args.kwargs['ssp_product'], product)
+
+    @mock.patch('enterprise_access.apps.api.serializers.customer_billing.CheckoutIntent.create_intent')
+    def test_create_translates_slug_conflict(self, mock_create_intent):
+        """SlugReservationConflict should be translated to RecordConflictError."""
+        serializer = self._make_serializer()
+        mock_create_intent.side_effect = SlugReservationConflict()
+
+        with self.assertRaises(RecordConflictError) as exc:
+            serializer.create({
+                'quantity': 10,
+                'enterprise_slug': 'acme-enterprise',
+                'enterprise_name': 'Acme Enterprise',
+            })
+
+        self.assertIn('already been reserved', str(exc.exception.detail))
+
+    @mock.patch('enterprise_access.apps.api.serializers.customer_billing.CheckoutIntent.create_intent')
+    def test_create_translates_failed_conflict(self, mock_create_intent):
+        """FailedCheckoutIntentConflict should be translated to RecordConflictError."""
+        serializer = self._make_serializer()
+        mock_create_intent.side_effect = FailedCheckoutIntentConflict()
+
+        with self.assertRaises(RecordConflictError) as exc:
+            serializer.create({
+                'quantity': 10,
+                'enterprise_slug': 'acme-enterprise',
+                'enterprise_name': 'Acme Enterprise',
+            })
+
+        self.assertIn('failed CheckoutIntent', str(exc.exception.detail))
+
+    def test_validate_terms_metadata_rejects_list(self):
+        """Serializer validation should reject terms_metadata when it is not an object."""
+        serializer = CheckoutIntentCreateRequestSerializer(data={
+            'quantity': 10,
+            'enterprise_slug': 'acme-enterprise',
+            'enterprise_name': 'Acme Enterprise',
+            'terms_metadata': ['not', 'a', 'dict'],
+        })
+
+        self.assertFalse(serializer.is_valid())
+        self.assertIn('terms_metadata', serializer.errors)

enterprise_access/apps/api/v1/tests/test_checkout_bff_views.py

@@ -26,6 +26,7 @@
 from enterprise_access.apps.core.constants import SYSTEM_ENTERPRISE_LEARNER_ROLE
 from enterprise_access.apps.customer_billing.constants import CheckoutIntentState
 from enterprise_access.apps.customer_billing.models import CheckoutIntent
+from enterprise_access.apps.customer_billing.tests.factories import CheckoutIntentFactory
 from enterprise_access.apps.customer_billing.tests.utils import AttrDict
 from test_utils import APITest
 
@@ -74,14 +75,27 @@ def test_context_endpoint_unauthenticated_access(self):
         # For unauthenticated users, checkout_intent should be None
         self.assertIsNone(response.data['checkout_intent'])
 
-    @mock.patch('enterprise_access.apps.customer_billing.models.CheckoutIntent.objects.filter')
-    def test_context_endpoint_authenticated_access(self, mock_filter):
+    @mock.patch('enterprise_access.apps.bffs.checkout.handlers.get_ssp_product_pricing')
+    @mock.patch('enterprise_access.apps.bffs.checkout.handlers.transform_enterprise_customer_users_data')
+    @mock.patch('enterprise_access.apps.bffs.checkout.handlers.get_and_cache_enterprise_customer_users')
+    @mock.patch('enterprise_access.apps.customer_billing.models.CheckoutIntent.for_user')
+    def test_context_endpoint_authenticated_access(
+        self,
+        mock_for_user,
+        mock_get_customers,
+        mock_transform,
+        mock_get_pricing,
+    ):
         """
         Test that authenticated users can access the context endpoint.
         """
+        mock_get_pricing.return_value = {}
+        mock_get_customers.return_value = {'results': []}
+        mock_transform.return_value = {'all_linked_enterprise_customer_users': []}
+
         # Set up a mock checkout intent for the authenticated user
         mock_intent = CheckoutIntent(**self.mock_checkout_intent_data)
-        mock_filter.return_value.first.return_value = mock_intent
+        mock_for_user.return_value = mock_intent
 
         self.set_jwt_cookie([{
             'system_wide_role': SYSTEM_ENTERPRISE_LEARNER_ROLE,
@@ -556,8 +570,8 @@ def test_success_endpoint_handler_exception(  # pylint: disable=unused-argument
     @mock.patch('enterprise_access.apps.bffs.checkout.handlers.CheckoutIntent.for_user')
     def test_success_endpoint_no_checkout_session(self, mock_for_user):
         """Test the success endpoint when no checkout session id is present."""
-        mock_checkout_intent = CheckoutIntent.objects.create(
-            user_id=self.user.id,
+        mock_checkout_intent = CheckoutIntentFactory(
+            user=self.user,
             state=CheckoutIntentState.CREATED,
             quantity=10,
             enterprise_name='Test Enterprise',
@@ -598,8 +612,8 @@ def test_success_endpoint_full_stripe_data(  # pylint: disable=unused-argument
         mock_get_customer, mock_get_pricing,
     ):
         """Test the success endpoint with full Stripe data integration."""
-        mock_checkout_intent = CheckoutIntent.objects.create(
-            user_id=self.user.id,
+        mock_checkout_intent = CheckoutIntentFactory(
+            user=self.user,
             state=CheckoutIntentState.CREATED,
             quantity=10,
             enterprise_name='Test Enterprise',

enterprise_access/apps/bffs/tests/test_checkout_handlers.py

@@ -311,26 +311,6 @@ def test_load_checkout_intent_for_authenticated_user(self, mock_filter, mock_get
         self.assertEqual(context.checkout_intent, context.checkout_intent or {} | mock_intent_data)
         mock_filter.assert_called_once_with(user=self.user)
 
-    @mock.patch('enterprise_access.apps.bffs.checkout.handlers.get_ssp_product_pricing')
-    @mock.patch('enterprise_access.apps.customer_billing.models.CheckoutIntent.objects.filter')
-    def test_load_checkout_intent_no_intent_exists(self, mock_filter, mock_get_pricing):
-        """
-        Test that load_and_process handles case where authenticated user has no checkout intent.
-        """
-        # Setup
-        mock_get_pricing.return_value = {}
-        mock_filter.return_value.first.return_value = None
-
-        context = self._create_context()
-        handler = CheckoutContextHandler(context)
-
-        # Execute
-        handler.load_and_process()
-
-        # Assert
-        self.assertIsNone(context.checkout_intent)
-        mock_filter.assert_called_once_with(user=self.user)
-
     @mock.patch('enterprise_access.apps.bffs.checkout.handlers.get_ssp_product_pricing')
     @mock.patch('enterprise_access.apps.customer_billing.models.CheckoutIntent.objects.filter')
     def test_load_checkout_intent_for_unauthenticated_user(self, mock_filter, mock_get_pricing):

enterprise_access/apps/customer_billing/api.py

@@ -17,7 +17,8 @@
 from enterprise_access.apps.customer_billing.models import (
     CheckoutIntent,
     FailedCheckoutIntentConflict,
-    SlugReservationConflict
+    SlugReservationConflict,
+    SspProduct
 )
 from enterprise_access.apps.customer_billing.pricing_api import get_ssp_product_pricing
 from enterprise_access.apps.customer_billing.stripe_api import create_subscription_checkout_session
@@ -423,13 +424,53 @@ def create_free_trial_checkout_session(
 
     user = input_data['user']
 
+    # Determine SSP product for this checkout. This is required now that
+    # CheckoutIntent.ssp_product is non-nullable.
+    ssp_product = None
+    try:
+        ssp_pricing = get_ssp_product_pricing()
+        matching_price = None
+        for price_data in ssp_pricing.values():
+            if price_data.get('id') == input_data.get('stripe_price_id'):
+                matching_price = price_data
+                break
+        if matching_price:
+            lookup_key = matching_price.get('lookup_key')
+            if lookup_key:
+                ssp_product = SspProduct.objects.filter(stripe_price_lookup_key=lookup_key).first()
+    except Exception:  # pylint: disable=broad-exception-caught
+        logger.exception(
+            'Failed to determine SSP product mapping for stripe_price_id=%s',
+            input_data.get('stripe_price_id'),
+        )
+        # Preserve backwards compatibility if pricing lookup is temporarily unavailable.
+        ssp_product = None
+
+    if not ssp_product:
+        ssp_product = SspProduct.objects.filter(slug='teams-yearly').first()
+    if not ssp_product:
+        error_code, developer_message = CHECKOUT_SESSION_ERROR_CODES['stripe_price_id']['DOES_NOT_EXIST']
+        raise CreateCheckoutSessionValidationError(
+            validation_errors_by_field={
+                'stripe_price_id': {
+                    'error_code': error_code,
+                    'developer_message': (
+                        f'{developer_message} No SspProduct configured for '
+                        f'stripe_price_id={input_data.get("stripe_price_id")}. '
+                        'Ensure stripe_price_lookup_key maps to an existing SspProduct.'
+                    ),
+                }
+            }
+        )
+
     # Create checkout intent, which reserves the enterprise name & slug.
     try:
         intent = CheckoutIntent.create_intent(
             user=user,
             quantity=input_data.get('quantity'),
             slug=input_data.get('enterprise_slug'),
             name=input_data.get('company_name'),
+            ssp_product=ssp_product,
         )
     except SlugReservationConflict as exc:
         raise CreateCheckoutSessionSlugReservationConflict() from exc

enterprise_access/apps/customer_billing/migrations/0037_make_ssp_product_non_nullable_and_user_fk.py

@@ -0,0 +1,41 @@
+"""
+Make CheckoutIntent.ssp_product non-nullable, change user to ForeignKey, and add unique constraint on (user, ssp_product).
+"""
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+def check_no_nulls(apps, schema_editor):
+    CheckoutIntent = apps.get_model('customer_billing', 'CheckoutIntent')
+    null_count = CheckoutIntent.objects.filter(ssp_product__isnull=True).count()
+    if null_count > 0:
+        raise Exception(
+            f"Cannot proceed: {null_count} CheckoutIntent rows still have NULL ssp_product. "
+            "Run the backfill migration (0036_backfill_checkoutintent_ssp_product_teams_yearly) first."
+        )
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('customer_billing', '0036_backfill_checkoutintent_ssp_product_teams_yearly'),
+    ]
+
+    operations = [
+        migrations.RunPython(check_no_nulls, migrations.RunPython.noop),
+        migrations.AlterField(
+            model_name='checkoutintent',
+            name='user',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL),
+        ),
+        migrations.AlterField(
+            model_name='checkoutintent',
+            name='ssp_product',
+            field=models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='customer_billing.sspproduct', null=False),
+        ),
+        migrations.AddConstraint(
+            model_name='checkoutintent',
+            constraint=models.UniqueConstraint(fields=('user', 'ssp_product'), name='unique_user_ssp_product'),
+        ),
+    ]