Merge pull request #644 from openedx/sameeramin/ENT-8798

feat: create CRUD endpoint for `CustomerAgreement`

license_manager/apps/api/serializers.py

@@ -167,6 +167,34 @@ def get_subscriptions(self, obj):
         return serializer.data
 
 
+class CustomerAgreementCreateRequestSerializer(MinimalCustomerAgreementSerializer):
+    """
+    Serializer for creating a new `CustomerAgreement` instance.
+    """
+
+    class Meta:
+        model = CustomerAgreement
+        fields = [
+            'enterprise_customer_uuid',
+            'default_enterprise_catalog_uuid',
+            'disable_expiration_notifications',
+            'disable_onboarding_notifications',
+        ]
+
+
+class CustomerAgreementUpdateRequestSerializer(MinimalCustomerAgreementSerializer):
+    """
+    Serializer for partially updating an existing `CustomerAgreement` instance.
+    """
+
+    class Meta:
+        model = CustomerAgreement
+        fields = [
+            'default_enterprise_catalog_uuid',
+            'disable_onboarding_notifications',
+        ]
+
+
 class LicenseSerializer(serializers.ModelSerializer):
     """
     Serializer for the `License` model.

license_manager/apps/api/v1/tests/test_views.py

@@ -43,6 +43,7 @@
 from license_manager.apps.subscriptions import constants
 from license_manager.apps.subscriptions.exceptions import LicenseRevocationError
 from license_manager.apps.subscriptions.models import (
+    CustomerAgreement,
     License,
     SubscriptionLicenseSource,
     SubscriptionsFeatureRole,
@@ -183,6 +184,42 @@ def _customer_agreement_list_request(api_client, user, enterprise_customer_uuid)
     return api_client.get(url)
 
 
+def _customer_agreement_create_request(api_client, user, enterprise_customer_uuid):
+    """
+    Helper method that requests to create a new CustomerAgreement for a specific enterprise customer uuid.
+    """
+    if user:
+        api_client.force_authenticate(user=user)
+
+    url = reverse('api:v1:customer-agreement-list')
+    data = {
+        'enterprise_customer_uuid': str(enterprise_customer_uuid),
+        'default_enterprise_catalog_uuid': str(uuid4()),
+        'disable_expiration_notifications': False,
+        'disable_onboarding_notifications': True,
+    }
+
+    return api_client.post(url, data=data)
+
+
+def _customer_agreement_partial_update_request(api_client, user, customer_agreement_uuid):
+    """
+    Helper method that requests to partially update a CustomerAgreement.
+    """
+    if user:
+        api_client.force_authenticate(user=user)
+
+    url = reverse('api:v1:customer-agreement-detail', kwargs={
+        'customer_agreement_uuid': customer_agreement_uuid,
+    })
+
+    data = {
+        'disable_onboarding_notifications': False,
+    }
+
+    return api_client.patch(url, data=data)
+
+
 def _subscriptions_list_request(api_client, user, enterprise_customer_uuid=None, current_plan=None):
     """
     Helper method that requests a list of subscriptions entities for a given enterprise_customer_uuid.
@@ -366,6 +403,45 @@ def test_customer_agreement_unauthenticated_user_401(api_client):
     assert status.HTTP_401_UNAUTHORIZED == response.status_code
 
 
+@pytest.mark.django_db
+def test_customer_agreement_create_unauthenticated_user_401(api_client):
+    """
+    Verify that unauthenticated users receive a 401 from the customer agreement create endpoint.
+    """
+    response = _customer_agreement_create_request(
+        api_client=api_client,
+        user=None,
+        enterprise_customer_uuid=uuid4(),
+    )
+    assert status.HTTP_401_UNAUTHORIZED == response.status_code
+
+
+@pytest.mark.django_db
+def test_customer_agreement_partial_update_unauthenticated_user_401(api_client):
+    """
+    Verify that unauthenticated users receive a 401 from the customer agreement partial update endpoint.
+    """
+    response = _customer_agreement_partial_update_request(
+        api_client=api_client,
+        user=None,
+        customer_agreement_uuid=uuid4(),
+    )
+    assert status.HTTP_401_UNAUTHORIZED == response.status_code
+
+
+@pytest.mark.django_db
+def test_customer_agreement_create_non_staff_user_403(api_client, staff_user):
+    """
+    Verify that non-staff users without JWT roles receive a 403 from the customer agreement create endpoint.
+    """
+    response = _customer_agreement_create_request(
+        api_client=api_client,
+        user=staff_user,
+        enterprise_customer_uuid=uuid4(),
+    )
+    assert status.HTTP_403_FORBIDDEN == response.status_code
+
+
 @pytest.mark.django_db
 def test_customer_agreement_non_staff_user_403(api_client, non_staff_user):
     """
@@ -375,6 +451,15 @@ def test_customer_agreement_non_staff_user_403(api_client, non_staff_user):
     assert status.HTTP_403_FORBIDDEN == response.status_code
 
 
+@pytest.mark.django_db
+def test_customer_agreement_partial_update_non_staff_user_403(api_client, non_staff_user):
+    """
+    Verify that non-staff users without JWT roles receive a 403 from the customer agreement partial update endpoint.
+    """
+    response = _customer_agreement_partial_update_request(api_client, non_staff_user, uuid4())
+    assert status.HTTP_403_FORBIDDEN == response.status_code
+
+
 @pytest.mark.django_db
 def test_customer_agreement_detail_non_staff_user_200(api_client, non_staff_user, user_role, boolean_toggle):
     """
@@ -435,6 +520,33 @@ def test_customer_agreement_list_superuser_200(api_client, superuser):
     _assert_customer_agreement_response_correct(response.data['results'][0], customer_agreement)
 
 
+@pytest.mark.django_db
+def test_customer_agreement_create_superuser_201(api_client, superuser):
+    """
+    Verify that superusers can create a new customer agreement.
+    """
+    enterprise_customer_uuid = uuid4()
+    response = _customer_agreement_create_request(api_client, superuser, enterprise_customer_uuid)
+    customer_agreement = CustomerAgreementFactory.create(enterprise_customer_uuid=enterprise_customer_uuid)
+
+    assert status.HTTP_201_CREATED == response.status_code
+    _assert_customer_agreement_response_correct(response.data, customer_agreement)
+    assert CustomerAgreement.objects.filter(uuid=response.data['uuid']).exists()
+
+
+@pytest.mark.django_db
+def test_customer_agreement_partial_update_superuser_200(api_client, superuser):
+    """
+    Verify that superusers can partially update a customer agreement.
+    """
+    enterprise_customer_uuid = uuid4()
+    customer_agreement = CustomerAgreementFactory.create(enterprise_customer_uuid=enterprise_customer_uuid)
+    response = _customer_agreement_partial_update_request(api_client, superuser, customer_agreement.uuid)
+
+    assert status.HTTP_200_OK == response.status_code
+    _assert_customer_agreement_response_correct(response.data, customer_agreement)
+
+
 @pytest.mark.django_db
 def test_customer_agreement_list_non_staff_user_200(api_client, non_staff_user, user_role, boolean_toggle):
     """

license_manager/apps/api/v1/views.py

@@ -17,7 +17,7 @@
 from edx_rest_framework_extensions.auth.jwt.authentication import (
     JwtAuthentication,
 )
-from rest_framework import filters, permissions, status, viewsets
+from rest_framework import filters, mixins, permissions, status, viewsets
 from rest_framework.authentication import SessionAuthentication
 from rest_framework.decorators import action
 from rest_framework.exceptions import ParseError, ValidationError
@@ -81,14 +81,41 @@
 ESTIMATED_COUNT_PAGINATOR_THRESHOLD = 10000
 
 
+@extend_schema_view(
+    list=extend_schema(
+        summary='List all CustomerAgreements',
+        description='List all CustomerAgreements with associated `SubscriptionPlans`',
+    ),
+    create=extend_schema(
+        summary='Create a new CustomerAgreement',
+        description='Create a new CustomerAgreement for a given `enterprise_customer_uuid`',
+        request=serializers.CustomerAgreementCreateRequestSerializer,
+    ),
+    retrieve=extend_schema(
+        summary='Retrieve a CustomerAgreement',
+        description='Retrieve a CustomerAgreement by its UUID',
+    ),
+    partial_update=extend_schema(
+        summary='Partial update a CustomerAgreement',
+        description='Partial update a CustomerAgreement by its UUID',
+        request=serializers.CustomerAgreementUpdateRequestSerializer,
+    ),
+    auto_apply=extend_schema(
+        summary='Auto-apply a license',
+        description='Auto-apply licenses for the given `user_email` and `lms_user_id`.',
+    ),
+)
 class CustomerAgreementViewSet(
     PermissionRequiredForListingMixin,
     UserDetailsFromJwtMixin,
-    viewsets.ReadOnlyModelViewSet,
+    viewsets.GenericViewSet,
+    mixins.ListModelMixin,
+    mixins.RetrieveModelMixin,
+    mixins.CreateModelMixin,
 ):
     """ Viewset for read operations on CustomerAgreements. """
 
-    authentication_classes = [JwtAuthentication]
+    authentication_classes = [JwtAuthentication, SessionAuthentication]
     permission_classes = [permissions.IsAuthenticated]
 
     lookup_field = 'uuid'
@@ -101,6 +128,27 @@ class CustomerAgreementViewSet(
     allowed_roles = [constants.SUBSCRIPTIONS_ADMIN_ROLE, constants.SUBSCRIPTIONS_LEARNER_ROLE]
     role_assignment_class = SubscriptionsRoleAssignment
 
+    def partial_update(self, request, *args, **kwargs):
+        """
+        Partial update a CustomerAgreement against given UUID.
+        """
+
+        if 'enterprise_customer_uuid' in request.data:
+            return Response(
+                'enterprise_customer_uuid cannot be updated',
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+
+        customer_agreement = self.get_object()
+        serializer = serializers.CustomerAgreementSerializer(
+            customer_agreement,
+            data=request.data,
+            partial=True,
+        )
+        serializer.is_valid(raise_exception=True)
+        serializer.save()
+        return Response(data=serializer.data, status=status.HTTP_200_OK)
+
     @property
     def requested_enterprise_uuid(self):
         return utils.get_requested_enterprise_uuid(self.request)