SONIC-704: Add CT discount availed check for outline tab

commerce_coordinator/apps/commercetools/clients.py

@@ -603,3 +603,36 @@ def retire_customer_anonymize_fields(self, customer_id: str, customer_version: i
                          f"with ID: {customer_id}, after LMS retirement with "
                          f"error correlation id {err.correlation_id} and error/s: {err.errors}")
             raise err
+
+    def is_first_time_discount_eligible(self, email: str) -> bool:
+        """
+        Check if a user is eligible for a first time discount
+        Args:
+            email (str): Email of the user
+        Returns (bool): True if the user is eligible for a first time discount
+        """
+        try:
+            discounts = self.base_client.discount_codes.query(
+                where="code in :discountCodes",
+                predicate_var={'discountCodes': settings.COMMERCETOOLS_FIRST_TIME_DISCOUNTS}
+            )
+            discount_ids = [discount.id for discount in discounts.results]
+
+            discounted_orders = self.base_client.orders.query(
+                where=[
+                    "customerEmail=:email",
+                    "orderState=:orderState",
+                    "discountCodes(discountCode(id in :discountIds))"
+                ],
+                predicate_var={'email': email, 'discountIds': discount_ids, 'orderState': 'Complete'}
+            )
+
+            if discounted_orders.total > 0:
+                return False
+
+            return True
+        except CommercetoolsError as err:  # pragma no cover
+            # Logs & ignores version conflict errors due to duplicate Commercetools messages
+            handle_commercetools_error(err, f"Unable to check if user {email} is eligible for a "
+                                            f"first time discount", True)
+            return True

commerce_coordinator/apps/commercetools/pipeline.py

@@ -409,3 +409,33 @@ def run_filter(
         except HTTPError as err:  # pragma no cover
             log.exception(f"[{tag}] HTTP Error: {err}")
             return PipelineCommand.CONTINUE.value
+
+
+class CheckCommercetoolsDiscountEligibility(PipelineStep):
+    """
+    Checks if a user is eligible for a first time discount in Commercetools.
+    """
+    def run_filter(self, email):  # pylint: disable=arguments-differ
+        """
+        Execute a filter with the signature specified.
+        Arguments:
+            email: Email of the user
+            kwargs: The keyword arguments passed through from the filter
+        Returns:
+            is_eligible (bool): True if the user is eligible for a first time discount
+        """
+        tag = type(self).__name__
+
+        try:
+            ct_api_client = CommercetoolsAPIClient()
+            is_eligible = ct_api_client.is_first_time_discount_eligible(email)
+
+            return {
+                'is_eligible': is_eligible
+            }
+        except CommercetoolsError as err:  # pragma no cover
+            log.exception(f"[{tag}] Commercetools Error: {err}, {err.errors}")
+            return PipelineCommand.CONTINUE.value
+        except HTTPError as err:  # pragma no cover
+            log.exception(f"[{tag}] HTTP Error: {err}")
+            return PipelineCommand.CONTINUE.value

commerce_coordinator/apps/commercetools/tests/test_clients.py

@@ -803,6 +803,99 @@ def test_update_customer_with_anonymized_fields_exception(self):
 
                 log_mock.assert_called_once_with(expected_message)
 
+    def test_is_first_time_discount_eligible_success(self):
+        base_url = self.client_set.get_base_url_from_client()
+        email = '[email protected]'
+
+        mock_discount_codes = {
+            "results": [
+                {"id": "discount-id-1"},
+                {"id": "discount-id-2"}
+            ]
+        }
+
+        mock_orders = {
+            "total": 0
+        }
+
+        with requests_mock.Mocker(real_http=True, case_sensitive=False) as mocker:
+            mocker.get(
+                f"{base_url}discount-codes",
+                json=mock_discount_codes,
+                status_code=200
+            )
+
+            mocker.get(
+                f"{base_url}orders",
+                json=mock_orders,
+                status_code=200
+            )
+
+            result = self.client_set.client.is_first_time_discount_eligible(email)
+            self.assertTrue(result)
+
+    def test_is_first_time_discount_eligible_not_eligible(self):
+        base_url = self.client_set.get_base_url_from_client()
+        email = '[email protected]'
+
+        mock_discount_codes = {
+            "results": [
+                {"id": "discount-id-1"},
+                {"id": "discount-id-2"}
+            ]
+        }
+
+        mock_orders = {
+            "total": 1
+        }
+
+        with requests_mock.Mocker(real_http=True, case_sensitive=False) as mocker:
+            mocker.get(
+                f"{base_url}discount-codes",
+                json=mock_discount_codes,
+                status_code=200
+            )
+
+            mocker.get(
+                f"{base_url}orders",
+                json=mock_orders,
+                status_code=200
+            )
+
+            result = self.client_set.client.is_first_time_discount_eligible(email)
+            self.assertFalse(result)
+
+    def test_is_first_time_discount_eligible_invalid_email(self):
+        invalid_email = "[email protected]"
+        base_url = self.client_set.get_base_url_from_client()
+
+        mock_discount_codes = {
+            "results": [
+                {"id": "discount-id-1"},
+                {"id": "discount-id-2"}
+            ]
+        }
+
+        mock_orders = {
+            "total": 0
+        }
+
+        with requests_mock.Mocker(real_http=True, case_sensitive=False) as mocker:
+            mocker.get(
+                f"{base_url}discount-codes",
+                json=mock_discount_codes,
+                status_code=200
+            )
+
+            mocker.get(
+                f"{base_url}orders",
+                json=mock_orders,
+                status_code=200
+            )
+
+            result = self.client_set.client.is_first_time_discount_eligible(invalid_email)
+            self.assertTrue(result)
+
 
 class PaginatedResultsTest(TestCase):
     """Tests for the simple logic in our Paginated Results Class"""

commerce_coordinator/apps/commercetools/tests/test_pipeline.py

@@ -11,6 +11,7 @@
 from commerce_coordinator.apps.commercetools.constants import COMMERCETOOLS_ORDER_MANAGEMENT_SYSTEM
 from commerce_coordinator.apps.commercetools.pipeline import (
     AnonymizeRetiredUser,
+    CheckCommercetoolsDiscountEligibility,
     CreateReturnForCommercetoolsOrder,
     CreateReturnPaymentTransaction,
     GetCommercetoolsOrders,
@@ -264,3 +265,33 @@ def test_pipeline(self, mock_anonymize_fields, mock_customer_by_lms_id, mock_ano
         ret = pipe.run_filter(lms_user_id=self.mock_lms_user_id)
         result_data = ret['returned_customer']
         self.assertEqual(result_data, self.update_customer_response)
+
+
+class CommercetoolsDiscountEligibilityPipelineTests(TestCase):
+    """Commercetools pipeline testcase for checking discount eligibility in CT"""
+    def setUp(self) -> None:
+        self.mock_email = "[email protected]"
+        self.mock_eligible_result = True
+        self.mock_ineligible_result = False
+
+    @patch(
+        'commerce_coordinator.apps.commercetools.clients.CommercetoolsAPIClient'
+        '.is_first_time_discount_eligible'
+    )
+    def test_pipeline_eligible(self, mock_is_eligible):
+        pipe = CheckCommercetoolsDiscountEligibility("test_pipe", None)
+        mock_is_eligible.return_value = self.mock_eligible_result
+        ret = pipe.run_filter(self.mock_email)
+        result_data = ret['is_eligible']
+        self.assertEqual(result_data, self.mock_eligible_result)
+
+    @patch(
+        'commerce_coordinator.apps.commercetools.clients.CommercetoolsAPIClient'
+        '.is_first_time_discount_eligible'
+    )
+    def test_pipeline_ineligible(self, mock_is_eligible):
+        pipe = CheckCommercetoolsDiscountEligibility("test_pipe", None)
+        mock_is_eligible.return_value = self.mock_ineligible_result
+        ret = pipe.run_filter(self.mock_email)
+        result_data = ret['is_eligible']
+        self.assertEqual(result_data, self.mock_ineligible_result)

commerce_coordinator/apps/lms/filters.py

@@ -64,3 +64,22 @@
         """
 
         return super().run_pipeline(lms_user_id=lms_user_id)
+
+
+class CheckFirstTimeDiscountEligibility(OpenEdxPublicFilter):
+    """
+    Filter to check if a user is eligible for a first time discount
+    """
+    # See pipeline step configuration OPEN_EDX_FILTERS_CONFIG dict in `settings/base.py`
+    filter_type = "org.edx.coordinator.lms.check.first.time.discount.eligibility.v1"
+
+    @classmethod
+    def run_filter(cls, email):
+        """
+        Call the PipelineStep(s) defined for this filter.
+        Arguments:
+            email: Email of the user
+        Returns:
+            is_eligible (bool): True if the user is eligible for a first time discount
+        """
+        return super().run_pipeline(email=email)

commerce_coordinator/apps/lms/tests/test_views.py

@@ -382,3 +382,72 @@ def test_post_with_unexpected_exception_fails(self, mock_filter):
         response = self.client.post(self.url, self.valid_payload, format='json')
 
         self.assertEqual(response.status_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+@ddt.ddt
+class FirstTimeDiscountEligibleViewTests(APITestCase):
+    """
+    Tests for the FirstTimeDiscountEligibleView to check if a user is eligible for a first-time discount.
+    """
+
+    test_user_username = 'test'
+    test_user_email = '[email protected]'
+    test_user_password = 'secret'
+
+    url = reverse('lms:first_time_discount_eligible')
+
+    def setUp(self):
+        super().setUp()
+        self.user = User.objects.create_user(
+            self.test_user_username,
+            self.test_user_email,
+            self.test_user_password,
+            is_staff=True,
+        )
+
+    def tearDown(self):
+        super().tearDown()
+        self.client.logout()
+
+    def authenticate_user(self):
+        self.client.login(username=self.test_user_username, password=self.test_user_password)
+        self.client.force_authenticate(user=self.user)
+
+    @patch('commerce_coordinator.apps.lms.views.CheckFirstTimeDiscountEligibility.run_filter')
+    def test_get_with_valid_email_eligibility_true(self, mock_filter):
+        """
+        Test case where the email is eligible for a first-time discount.
+        """
+        self.authenticate_user()
+        mock_filter.return_value = {'is_eligible': True}
+
+        response = self.client.get(self.url, {'email': self.test_user_email})
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data, {"is_eligible": True})
+        mock_filter.assert_called_once_with(email=self.test_user_email)
+
+    @patch('commerce_coordinator.apps.lms.views.CheckFirstTimeDiscountEligibility.run_filter')
+    def test_get_with_valid_email_eligibility_false(self, mock_filter):
+        """
+        Test case where the email is not eligible for a first-time discount.
+        """
+        self.authenticate_user()
+        mock_filter.return_value = {'is_eligible': False}
+
+        response = self.client.get(self.url, {'email': self.test_user_email})
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(response.data, {"is_eligible": False})
+        mock_filter.assert_called_once_with(email=self.test_user_email)
+
+    def test_get_with_missing_email_fails(self):
+        """
+        Test case where the email is not provided in the request query params.
+        """
+        self.authenticate_user()
+
+        response = self.client.get(self.url)
+
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertEqual(response.data, {'detail': 'Could not detect user email.'})

commerce_coordinator/apps/lms/urls.py

@@ -5,6 +5,7 @@
 from django.urls import path
 
 from commerce_coordinator.apps.lms.views import (
+    FirstTimeDiscountEligibleView,
     OrderDetailsRedirectView,
     PaymentPageRedirectView,
     RefundView,
@@ -16,5 +17,6 @@
     path('payment_page_redirect/', PaymentPageRedirectView.as_view(), name='payment_page_redirect'),
     path('order_details_page_redirect/', OrderDetailsRedirectView.as_view(), name='order_details_page_redirect'),
     path('refund/', RefundView.as_view(), name='refund'),
-    path('user_retirement/', RetirementView.as_view(), name='user_retirement')
+    path('user_retirement/', RetirementView.as_view(), name='user_retirement'),
+    path('first-time-discount-eligible/', FirstTimeDiscountEligibleView.as_view(), name='first_time_discount_eligible'),
 ]

commerce_coordinator/apps/lms/views.py

@@ -8,7 +8,7 @@
 from django.http import HttpResponseBadRequest, HttpResponseRedirect
 from edx_rest_framework_extensions.permissions import LoginRedirectIfUnauthenticated
 from openedx_filters.exceptions import OpenEdxFilterException
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.permissions import IsAdminUser
 from rest_framework.response import Response
 from rest_framework.status import HTTP_200_OK, HTTP_303_SEE_OTHER, HTTP_400_BAD_REQUEST, HTTP_500_INTERNAL_SERVER_ERROR
@@ -17,6 +17,7 @@
 
 from commerce_coordinator.apps.core.constants import HttpHeadersNames, MediaTypes
 from commerce_coordinator.apps.lms.filters import (
+    CheckFirstTimeDiscountEligibility,
     OrderRefundRequested,
     PaymentPageRedirectRequested,
     UserRetirementRequested
@@ -334,3 +335,24 @@ def post(self, request) -> Response:
             logger.exception(f"[RefundView] Exception raised in {self.post.__name__} with error {repr(e)}")
             return Response('Exception occurred while retiring Commercetools customer',
                             status=HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+class FirstTimeDiscountEligibleView(APIView):
+    """View to check if a user is eligible for a first time discount"""
+    permission_classes = [LoginRedirectIfUnauthenticated]
+    throttle_classes = [UserRateThrottle]
+
+    def get(self, request):
+        """Return True if user is eligible for a first time discount."""
+        email = request.query_params.get('email')
+
+        if not email:  # pragma: no cover
+            raise PermissionDenied(detail="Could not detect user email.")
+
+        result = CheckFirstTimeDiscountEligibility.run_filter(email=email)
+
+        output = {
+            "is_eligible": result.get('is_eligible', True)
+        }
+
+        return Response(output)

commerce_coordinator/settings/base.py

@@ -399,6 +399,12 @@ def root(*path_fragments):
         "pipeline": [
             'commerce_coordinator.apps.commercetools.pipeline.AnonymizeRetiredUser',
         ]
+    },
+    'org.edx.coordinator.lms.check.first.time.discount.eligibility.v1': {
+        'fail_silently': False,
+        'pipeline': [
+            'commerce_coordinator.apps.commercetools.pipeline.CheckCommercetoolsDiscountEligibility',
+        ]
     }
 }
 
@@ -454,6 +460,8 @@ def root(*path_fragments):
 # Checkout view urls
 COMMERCETOOLS_FRONTEND_URL = 'http://localhost:3000/SET-ME'
 
+COMMERCETOOLS_FIRST_TIME_DISCOUNTS = ('EDXWELCOME', 'NEW2EDX')
+
 COMMERCETOOLS_MERCHANT_CENTER_ORDERS_PAGE_URL = \
     f'https://mc.{_COMMERCETOOLS_CONFIG_GEO}.commercetools.com/{COMMERCETOOLS_CONFIG["projectKey"]}/orders'