feat: add provisioning OLI endpoint

also supports requests by checkout intent or uuid. response serializer
returns both values
ENT-11122

Author: iloveagent57

enterprise_access/apps/api/serializers/__init__.py

@@ -25,7 +25,12 @@
     CustomerBillingCreateCheckoutSessionValidationFailedResponseSerializer,
     StripeEventSummaryReadOnlySerializer
 )
-from .provisioning import ProvisioningRequestSerializer, ProvisioningResponseSerializer
+from .provisioning import (
+    ProvisioningRequestSerializer,
+    ProvisioningResponseSerializer,
+    SubscriptionPlanOLIUpdateResponseSerializer,
+    SubscriptionPlanOLIUpdateSerializer
+)
 from .subsidy_access_policy import (
     GroupMemberWithAggregatesRequestSerializer,
     GroupMemberWithAggregatesResponseSerializer,

enterprise_access/apps/api/serializers/provisioning.py

@@ -210,3 +210,62 @@ class ProvisioningResponseSerializer(BaseSerializer):
     enterprise_catalog = EnterpriseCatalogResponseSerializer()
     customer_agreement = CustomerAgreementResponseSerializer()
     subscription_plan = SubscriptionPlanResponseSerializer()
+
+
+class SubscriptionPlanOLIUpdateSerializer(BaseSerializer):
+    """
+    Request serializer for updating a SubscriptionPlan's Salesforce OLI.
+    """
+    checkout_intent_id = serializers.IntegerField(
+        help_text='The integer ID of the CheckoutIntent associated with this subscription.',
+        required=False,
+        allow_null=True,
+    )
+    checkout_intent_uuid = serializers.UUIDField(
+        help_text='The UUID of the CheckoutIntent associated with this subscription.',
+        required=False,
+        allow_null=True,
+    )
+    salesforce_opportunity_line_item = serializers.CharField(
+        help_text='The Salesforce Opportunity Line Item ID to associate with the subscription plan.'
+    )
+    is_trial = serializers.BooleanField(
+        required=False,
+        default=False,
+        help_text='Whether this OLI is for the trial plan (True) or paid plan (False).'
+    )
+
+    def validate(self, data):
+        """
+        Validates that exactly one of CheckoutIntent ``id`` and ``uuid`` is provided.
+        """
+        if not (data.get('checkout_intent_id') or data.get('checkout_intent_uuid')):
+            raise serializers.ValidationError('One of CheckoutIntent id or uuid is required')
+        if data.get('checkout_intent_id') and data.get('checkout_intent_uuid'):
+            raise serializers.ValidationError('Only one of CheckoutIntent id or uuid can be provided')
+        return data
+
+
+class SubscriptionPlanOLIUpdateResponseSerializer(BaseSerializer):
+    """
+    Response serializer for SubscriptionPlan OLI update.
+    """
+    success = serializers.BooleanField(
+        help_text='Whether the update was successful',
+    )
+    subscription_plan_uuid = serializers.UUIDField(
+        help_text='The UUID identifier of the future plan (which receives the updated OLI value)',
+    )
+    salesforce_opportunity_line_item = serializers.CharField(
+        help_text='The Salesforce Opportunity Line Item ID to associate with the subscription plan.'
+    )
+    checkout_intent_uuid = serializers.UUIDField(
+        help_text='The UUID of the CheckoutIntent associated with this subscription.',
+        required=False,
+        allow_null=True,
+    )
+    checkout_intent_id = serializers.IntegerField(
+        help_text='The integer ID of the CheckoutIntent associated with this subscription.',
+        required=False,
+        allow_null=True,
+    )

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

@@ -923,3 +923,177 @@ def test_checkout_intent_different_slug_ignored(self, mock_lms_api_client, mock_
         # Verify workflow was still created successfully
         workflow = ProvisionNewCustomerWorkflow.objects.first()
         self.assertIsNotNone(workflow)
+
+
+@ddt.ddt
+class TestSubscriptionPlanOLIUpdateView(APITest):
+    """
+    Tests for the SubscriptionPlan OLI update endpoint.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.checkout_intent = CheckoutIntent.objects.create(
+            user=self.user,
+            enterprise_slug='test-enterprise',
+            enterprise_name='Test Enterprise',
+            quantity=10,
+            state=CheckoutIntentState.FULFILLED,
+            expires_at=timezone.now() + timedelta(hours=1),
+            enterprise_uuid=TEST_ENTERPRISE_UUID,
+        )
+        self.workflow = ProvisionNewCustomerWorkflow.objects.create(
+            input_data={'test': 'data'}
+        )
+        self.checkout_intent.workflow = self.workflow
+        self.checkout_intent.save()
+
+        self.subscription_plan_uuid = uuid.uuid4()
+        # Create a subscription plan step with complete output data
+        GetCreateSubscriptionPlanStep.objects.create(
+            workflow_record_uuid=self.workflow.uuid,
+            input_data={
+                'title': 'Paid Plan',
+                'is_trial': False,
+                'salesforce_opportunity_line_item': 'existing_oli_123',
+                'start_date': '2025-01-15T00:00:00Z',
+                'expiration_date': '2025-12-31T23:59:59Z',
+                'desired_num_licenses': 10,
+                'product_id': 1,
+            },
+            output_data={
+                'uuid': str(self.subscription_plan_uuid),
+                'title': 'Paid Plan',
+                'salesforce_opportunity_line_item': 'existing_oli_123',
+                'created': '2025-01-01T00:00:00Z',
+                'start_date': '2025-01-15T00:00:00Z',
+                'expiration_date': '2025-12-31T23:59:59Z',
+                'is_active': True,
+                'is_current': True,
+                'plan_type': 'Standard',
+                'enterprise_catalog_uuid': str(TEST_CATALOG_UUID),
+                'product': 1,
+            }
+        )
+
+        self.endpoint_url = reverse('api:v1:subscription-plan-oli-update')
+        self.set_jwt_cookie([
+            {
+                'system_wide_role': SYSTEM_ENTERPRISE_PROVISIONING_ADMIN_ROLE,
+                'context': ALL_ACCESS_CONTEXT,
+            },
+        ])
+
+    def tearDown(self):
+        super().tearDown()
+        CheckoutIntent.objects.all().delete()
+        ProvisionNewCustomerWorkflow.objects.all().delete()
+        GetCreateSubscriptionPlanStep.objects.all().delete()
+
+    @mock.patch('enterprise_access.apps.api.v1.views.provisioning.LicenseManagerApiClient')
+    def test_successful_oli_update(self, mock_license_manager_client):
+        """Test successful update of SubscriptionPlan OLI."""
+        mock_client = mock_license_manager_client.return_value
+        mock_client.update_subscription_plan.return_value = {
+            'uuid': str(self.subscription_plan_uuid),
+            'salesforce_opportunity_line_item': 'new_oli_456',
+        }
+
+        request_data = {
+            'checkout_intent_id': self.checkout_intent.id,
+            'salesforce_opportunity_line_item': 'new_oli_456',
+            'is_trial': False,
+        }
+
+        response = self.client.post(self.endpoint_url, data=request_data)
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK, response.json())
+        response_data = response.json()
+        self.assertTrue(response_data['success'])
+        self.assertEqual(str(response_data['subscription_plan_uuid']), str(self.subscription_plan_uuid))
+        self.assertEqual(response_data['salesforce_opportunity_line_item'], 'new_oli_456')
+        self.assertEqual(str(response_data['checkout_intent_uuid']), str(self.checkout_intent.uuid))
+        self.assertEqual(str(response_data['checkout_intent_id']), str(self.checkout_intent.id))
+
+        mock_client.update_subscription_plan.assert_called_once_with(
+            subscription_uuid=str(self.subscription_plan_uuid),
+            salesforce_opportunity_line_item='new_oli_456'
+        )
+
+    def test_checkout_intent_not_found(self):
+        """Test error when CheckoutIntent doesn't exist."""
+        request_data = {
+            'checkout_intent_uuid': str(uuid.uuid4()),  # Non-existent id
+            'salesforce_opportunity_line_item': 'new_oli_456',
+        }
+
+        response = self.client.post(self.endpoint_url, data=request_data)
+
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+        response_data = response.json()
+        if isinstance(response_data, dict):
+            self.assertIn('not found', response_data.get('detail', '').lower())
+        else:
+            self.assertTrue(any('not found' in str(item).lower() for item in response_data))
+
+    def test_no_workflow_associated(self):
+        """Test error when CheckoutIntent has no workflow."""
+        checkout_intent_no_workflow = CheckoutIntent.objects.create(
+            user=UserFactory(),
+            enterprise_slug='test-no-workflow',
+            enterprise_name='Test No Workflow',
+            quantity=5,
+            state=CheckoutIntentState.PAID,
+            expires_at=timezone.now() + timedelta(hours=1),
+        )
+
+        request_data = {
+            'checkout_intent_uuid': str(checkout_intent_no_workflow.uuid),
+            'salesforce_opportunity_line_item': 'new_oli_456',
+        }
+
+        response = self.client.post(self.endpoint_url, data=request_data)
+
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        response_data = response.json()
+        # Handle both dict and list response formats
+        if isinstance(response_data, dict):
+            self.assertIn('no associated workflow', response_data.get('detail', '').lower())
+        else:
+            # If it's a list (field errors), check the first item
+            self.assertTrue(any('no associated workflow' in str(item).lower() for item in response_data))
+
+    @mock.patch('enterprise_access.apps.api.v1.views.provisioning.LicenseManagerApiClient')
+    def test_license_manager_api_error(self, mock_license_manager_client):
+        """Test error handling when License Manager API fails."""
+        mock_client = mock_license_manager_client.return_value
+        mock_client.update_subscription_plan.side_effect = Exception('License Manager API Error')
+
+        request_data = {
+            'checkout_intent_uuid': str(self.checkout_intent.uuid),
+            'salesforce_opportunity_line_item': 'new_oli_456',
+            'is_trial': False,
+        }
+
+        response = self.client.post(self.endpoint_url, data=request_data)
+
+        self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+        response_data = response.json()
+        if isinstance(response_data, dict):
+            self.assertIn('failed to update subscription plan', response_data.get('detail', '').lower())
+        else:
+            self.assertTrue(any('failed to update subscription plan' in str(item).lower() for item in response_data))
+
+    def test_unauthorized_access(self):
+        """Test that unauthorized users cannot access the endpoint."""
+        self.set_jwt_cookie([])  # Remove authorization
+
+        request_data = {
+            'checkout_intent_id': str(self.checkout_intent.id),
+            'salesforce_opportunity_line_item': 'new_oli_456',
+        }
+
+        response = self.client.post(self.endpoint_url, data=request_data)
+
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

enterprise_access/apps/api/v1/urls.py

@@ -62,6 +62,11 @@
         views.ProvisioningCreateView.as_view(),
         name='provisioning-create',
     ),
+    path(
+        'provisioning/subscription-plan-oli-update',
+        views.SubscriptionPlanOLIUpdateView.as_view(),
+        name='subscription-plan-oli-update',
+    ),
 ]
 
 urlpatterns += router.urls

enterprise_access/apps/api/v1/views/__init__.py

@@ -17,7 +17,7 @@
 from .content_assignments.assignments import LearnerContentAssignmentViewSet
 from .content_assignments.assignments_admin import LearnerContentAssignmentAdminViewSet
 from .customer_billing import CheckoutIntentViewSet, CustomerBillingViewSet, StripeEventSummaryViewSet
-from .provisioning import ProvisioningCreateView
+from .provisioning import ProvisioningCreateView, SubscriptionPlanOLIUpdateView
 from .subsidy_access_policy import (
     SubsidyAccessPolicyAllocateViewset,
     SubsidyAccessPolicyGroupViewset,