From 7a2a0665b80a3d80d9472e52df73a08663daa1bf Mon Sep 17 00:00:00 2001 From: Max Date: Fri, 4 Jul 2025 21:43:35 +0300 Subject: [PATCH] test: Add comprehensive tests for user promo operations This commit introduces a suite of tests for the user-facing promo code functionality, ensuring robustness and correctness. The new tests cover the following areas: - Promo Code Activation: - Verification of activation logic, including success and failure scenarios for common and unique promos. - Edge cases such as inactive promos, targeting mismatches, and exhausted activation limits (max_count). - Checks to prevent users who are blocked by the anti-fraud service from activating promos. - Promo Activation History: - Tests for the user's promo activation history endpoint. - Validation of the returned data and its order. - Pagination functionality (limit and offset). - Anti-Fraud Service Integration: - Unit tests for the AntiFraudService to verify its caching behavior and resilience to external service failures. - Integration tests to confirm that the promo activation process correctly respects the anti-fraud service's verdicts, including delayed updates from the cache. --- promo_code/user/tests/user/base.py | 23 +- .../tests/user/operations/test_activate.py | 401 ++++++++++++++++++ .../test_activate_validate_cache.py | 148 +++++++ .../tests/user/operations/test_history.py | 233 ++++++++++ .../user/tests/user/test_antifraud_service.py | 131 ++++++ 5 files changed, 934 insertions(+), 2 deletions(-) create mode 100644 promo_code/user/tests/user/operations/test_activate.py create mode 100644 promo_code/user/tests/user/operations/test_activate_validate_cache.py create mode 100644 promo_code/user/tests/user/operations/test_history.py create mode 100644 promo_code/user/tests/user/test_antifraud_service.py diff --git a/promo_code/user/tests/user/base.py b/promo_code/user/tests/user/base.py index a8131b5..62c9335 100644 --- a/promo_code/user/tests/user/base.py +++ b/promo_code/user/tests/user/base.py @@ -1,4 +1,7 @@ +import django.conf +import django.core.cache import django.urls +import django_redis import rest_framework.test import rest_framework_simplejwt.token_blacklist.models as tb_models @@ -15,13 +18,21 @@ def setUpTestData(cls): cls.user_signin_url = django.urls.reverse('api-user:user-sign-in') cls.user_profile_url = django.urls.reverse('api-user:user-profile') cls.user_feed_url = django.urls.reverse('api-user:user-feed') - + cls.user_promo_history_url = django.urls.reverse( + 'api-user:user-promo-history', + ) cls.company_signin_url = django.urls.reverse( 'api-business:company-sign-in', ) cls.promo_list_create_url = django.urls.reverse( 'api-business:promo-list-create', ) + cls.antifraud_update_user_verdict_url = ( + django.conf.settings.ANTIFRAUD_UPDATE_USER_VERDICT_URL + ) + cls.antifraud_set_delay_url = ( + django.conf.settings.ANTIFRAUD_SET_DELAY_URL + ) company1_data = { 'name': 'Digital Marketing Solutions Inc.', @@ -68,11 +79,12 @@ def tearDown(self): business.models.Company.objects.all().delete() business.models.Promo.objects.all().delete() business.models.PromoCode.objects.all().delete() + user.models.PromoActivationHistory.objects.all().delete() user.models.PromoComment.objects.all().delete() user.models.PromoLike.objects.all().delete() user.models.User.objects.all().delete() tb_models.BlacklistedToken.objects.all().delete() - tb_models.OutstandingToken.objects.all().delete() + django_redis.get_redis_connection('default').flushall() super().tearDown() @classmethod @@ -89,6 +101,13 @@ def get_user_promo_detail_url(cls, promo_id): kwargs={'id': promo_id}, ) + @classmethod + def get_user_promo_activate_url(cls, promo_id): + return django.urls.reverse( + 'api-user:user-promo-activate', + kwargs={'id': promo_id}, + ) + @classmethod def get_user_promo_like_url(cls, promo_id): return django.urls.reverse( diff --git a/promo_code/user/tests/user/operations/test_activate.py b/promo_code/user/tests/user/operations/test_activate.py new file mode 100644 index 0000000..e28c313 --- /dev/null +++ b/promo_code/user/tests/user/operations/test_activate.py @@ -0,0 +1,401 @@ +import datetime + +import requests +import rest_framework.status + +import user.tests.user.base + + +class PromoActivationTests(user.tests.user.base.BaseUserTestCase): + def setUp(self): + super().setUp() + + user1_data = { + 'name': 'Steve', + 'surname': 'Wozniak', + 'email': 'creator2@apple.com', + 'password': 'WhoLiveSInCalifornia2000!', + 'other': {'age': 60, 'country': 'gb'}, + } + response = self.client.post( + self.user_signup_url, + user1_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user1_token = response.data['access'] + + user2_data = { + 'name': 'Mike', + 'surname': 'Bloomberg', + 'email': 'mike2@bloomberg.com', + 'password': 'WhoLiveSInCalifornia2000!', + 'other': {'age': 15, 'country': 'us'}, + } + response = self.client.post( + self.user_signup_url, + user2_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user2_token = response.data['access'] + + user3_data = { + 'name': 'Yefim', + 'surname': 'Dinitz', + 'email': 'algo3@prog.ru', + 'password': 'HardPASSword1!', + 'other': {'age': 40, 'country': 'kz'}, + } + response = self.client.post( + self.user_signup_url, + user3_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user3_token = response.data['access'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company1_token, + ) + promo1_data = { + 'description': 'Active COMMON promo for all', + 'target': {}, + 'max_count': 4, + 'active_from': '2025-01-10', + 'mode': 'COMMON', + 'promo_common': 'sale-10', + } + response = self.client.post( + self.promo_list_create_url, + promo1_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo1_id = response.data['id'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company2_token, + ) + promo2_data = { + 'description': 'Inactive COMMON promo for kz, 28..', + 'target': {'country': 'kz', 'age_from': 28}, + 'max_count': 10, + 'active_from': ( + datetime.date.today() + datetime.timedelta(days=10) + ).strftime( + '%Y-%m-%d', + ), + 'mode': 'COMMON', + 'promo_common': 'sale-20', + } + response = self.client.post( + self.promo_list_create_url, + promo2_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo2_id = response.data['id'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company2_token, + ) + promo3_data = { + 'description': 'Active UNIQUE promo for gb, 45..', + 'target': {'country': 'gb', 'age_from': 45}, + 'active_until': '2035-02-10', + 'mode': 'UNIQUE', + 'max_count': 1, + 'promo_unique': ['uniq1', 'uniq2'], + } + response = self.client.post( + self.promo_list_create_url, + promo3_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo3_id = response.data['id'] + + def test_inactive_promo_activation_denied(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user3_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo2_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_targeting_mismatch_promo_activation_denied(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company1_token, + ) + promo_target_data = { + 'description': 'Active COMMON promo for us, 20..', + 'target': {'country': 'us', 'age_from': 20}, + 'max_count': 10, + 'active_from': '2020-01-01', + 'mode': 'COMMON', + 'promo_common': 'target-sale', + } + response = self.client.post( + self.promo_list_create_url, + promo_target_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + promo_target_id = response.data['id'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user3_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(promo_target_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_common_promo_activation_by_user1(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data['promo'], 'sale-10') + + def test_common_promo_activation_by_user3(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user3_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data['promo'], 'sale-10') + + def test_common_promo_multiple_activation_by_same_user(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data['promo'], 'sale-10') + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data['promo'], 'sale-10') + + def test_blocked_user_promo_activation_denied(self): + self.client.credentials() + response = requests.post( + self.antifraud_update_user_verdict_url, + json={'user_email': 'mike2@bloomberg.com', 'ok': False}, + ) + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user2_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertNotIn('promo', response) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_common_promo_max_count_exhausted(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + for _ in range(2): + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user3_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_patch_max_count_less_than_used_count_denied(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company1_token, + ) + patch_data = {'max_count': 0} + response = self.client.patch( + self.get_business_promo_detail_url(self.promo1_id), + patch_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_400_BAD_REQUEST, + ) + + def test_unique_promo_activation_by_user1_first_attempt(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertIn( + response.data['promo'], + ['uniq1', 'uniq2'], + ) + + def test_unique_promo_activation_by_user1_denied_by_max_count( + self, + ): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + response = self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_get_promo_detail_for_activated_promo_by_user1(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + response = self.client.get( + self.get_user_promo_detail_url(self.promo1_id), + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data['promo_id'], self.promo1_id) + self.assertTrue(response.data['active']) + self.assertTrue(response.data['is_activated_by_user']) diff --git a/promo_code/user/tests/user/operations/test_activate_validate_cache.py b/promo_code/user/tests/user/operations/test_activate_validate_cache.py new file mode 100644 index 0000000..f150f89 --- /dev/null +++ b/promo_code/user/tests/user/operations/test_activate_validate_cache.py @@ -0,0 +1,148 @@ +import time + +import requests +import rest_framework.status + +import user.tests.user.base + + +class TestPromoCodeActivationValidateCache( + user.tests.user.base.BaseUserTestCase, +): + def setUp(self): + super().setUp() + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company1_token, + ) + promo_data = { + 'description': 'Active COMMON promo for all', + 'target': {}, + 'max_count': 100, + 'active_from': '2024-01-01', + 'mode': 'COMMON', + 'promo_common': 'sale-for-all', + } + response = self.client.post( + self.promo_list_create_url, + promo_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo1_id = response.data['id'] + + user4_data = { + 'name': 'Mason', + 'surname': 'Jones', + 'email': 'dontstopthemusic@gmail.com', + 'password': 'PasswordForMason123!', + 'other': {'age': 25, 'country': 'ca'}, + } + response = self.client.post( + self.user_signup_url, + user4_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user4_token = response.data['access'] + self.user4_email = user4_data['email'] + + user5_data = { + 'name': 'Chloe', + 'surname': 'Taylor', + 'email': 'slowdown@antifraud.ru', + 'password': 'PasswordForChloe456!', + 'other': {'age': 32, 'country': 'au'}, + } + response = self.client.post( + self.user_signup_url, + user5_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user5_token = response.data['access'] + + def test_promo_activation_and_antifraud_block_with_delay(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user4_token, + ) + response = requests.post( + self.antifraud_update_user_verdict_url, + json={'user_email': self.user4_email, 'ok': True}, + ) + + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + antifraud_verdict_data = {'user_email': self.user4_email, 'ok': False} + + antifraud_response = requests.post( + self.antifraud_update_user_verdict_url, + json=antifraud_verdict_data, + timeout=5, + ) + self.assertEqual( + antifraud_response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + time.sleep(4) + + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_403_FORBIDDEN, + ) + + def test_promo_activation_with_cached_antifraud_verdict(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user5_token, + ) + + start_time_0 = time.time() + response = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + ) + end_time_0 = time.time() + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + start_time_1 = time.time() + response_1 = self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + ) + end_time_1 = time.time() + + self.assertEqual( + response_1.status_code, + rest_framework.status.HTTP_200_OK, + ) + + self.assertTrue((end_time_1 - start_time_1) < 1.0) + self.assertTrue(end_time_0 - start_time_0 > end_time_1 - start_time_1) diff --git a/promo_code/user/tests/user/operations/test_history.py b/promo_code/user/tests/user/operations/test_history.py new file mode 100644 index 0000000..14ce985 --- /dev/null +++ b/promo_code/user/tests/user/operations/test_history.py @@ -0,0 +1,233 @@ +import rest_framework.status + +import user.models +import user.tests.user.base + + +class PromoHistoryTests(user.tests.user.base.BaseUserTestCase): + def setUp(self): + super().setUp() + + user1_data = { + 'name': 'Steve', + 'surname': 'Wozniak', + 'email': 'creator2@apple.com', + 'password': 'WhoLiveSInCalifornia2000!', + 'other': {'age': 60, 'country': 'gb'}, + } + response = self.client.post( + self.user_signup_url, + user1_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.user1_token = response.data['access'] + self.user1 = user.models.User.objects.get(email=user1_data['email']) + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company1_token, + ) + promo1_data = { + 'description': 'Common promo for all', + 'target': {}, + 'max_count': 4, + 'active_from': '2025-01-10', + 'mode': 'COMMON', + 'promo_common': 'sale-10', + } + response = self.client.post( + self.promo_list_create_url, + promo1_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo1_id = response.data['id'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.company2_token, + ) + promo3_data = { + 'description': 'Unique promo for gb, 45..', + 'target': {'country': 'gb', 'age_from': 45}, + 'active_until': '2035-02-10', + 'mode': 'UNIQUE', + 'max_count': 1, + 'promo_unique': ['unique-code-a', 'unique-code-b'], + } + response = self.client.post( + self.promo_list_create_url, + promo3_data, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_201_CREATED, + ) + self.promo3_id = response.data['id'] + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + + def test_get_promo_history_for_user1(self): + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + response = self.client.get(self.user_promo_history_url, format='json') + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + user.models.PromoActivationHistory.objects.filter( + user=self.user1, + ).delete() + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + + response = self.client.get(self.user_promo_history_url, format='json') + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + + expected_promo_ids_in_history = [ + self.promo3_id, + self.promo3_id, + self.promo1_id, + self.promo1_id, + ] + + self.assertEqual( + len(response.data), + len(expected_promo_ids_in_history), + ) + for i, item in enumerate(response.data): + self.assertEqual( + item['promo_id'], + expected_promo_ids_in_history[i], + ) + self.assertFalse( + item['active'], + ) + self.assertTrue(item['is_activated_by_user']) + + def test_get_promo_history_with_pagination_offset_limit(self): + user.models.PromoActivationHistory.objects.filter( + user=self.user1, + ).delete() + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + + expected_promo_ids = [self.promo1_id, self.promo1_id] + + response = self.client.get( + self.user_promo_history_url, + {'offset': 2, 'limit': 3}, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(len(response.data), 2) + + self.assertEqual(response.data[0]['promo_id'], expected_promo_ids[0]) + self.assertEqual(response.data[1]['promo_id'], expected_promo_ids[1]) + + self.assertFalse(response.data[0]['active']) + self.assertTrue(response.data[0]['is_activated_by_user']) + self.assertFalse(response.data[1]['active']) + self.assertTrue(response.data[1]['is_activated_by_user']) + + self.assertEqual(response.headers['X-Total-Count'], '4') + + def test_get_promo_history_with_zero_limit(self): + user.models.PromoActivationHistory.objects.filter( + user=self.user1, + ).delete() + + self.client.credentials( + HTTP_AUTHORIZATION='Bearer ' + self.user1_token, + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo1_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + self.client.post( + self.get_user_promo_activate_url(self.promo3_id), + format='json', + ) + + response = self.client.get( + self.user_promo_history_url, + {'limit': 0}, + format='json', + ) + self.assertEqual( + response.status_code, + rest_framework.status.HTTP_200_OK, + ) + self.assertEqual(response.data, []) + self.assertEqual(response.headers['X-Total-Count'], '4') diff --git a/promo_code/user/tests/user/test_antifraud_service.py b/promo_code/user/tests/user/test_antifraud_service.py new file mode 100644 index 0000000..57c04e2 --- /dev/null +++ b/promo_code/user/tests/user/test_antifraud_service.py @@ -0,0 +1,131 @@ +import datetime +import unittest.mock + +import django.test +import requests.exceptions + +import user.antifraud_service + + +class AntiFraudServiceTests(django.test.SimpleTestCase): + def setUp(self): + self.service = user.antifraud_service.AntiFraudService() + self.user_email = 'test@example.com' + self.promo_id = '1bfd61b1-52ff-4c0f-ba8b-434ad3d0f812' + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_get_verdict_from_cache(self, mock_cache, mock_post): + mock_cache.get.return_value = {'ok': True, 'reason': 'From Cache'} + + result = self.service.get_verdict(self.user_email, self.promo_id) + + mock_cache.get.assert_called_once_with( + f'antifraud_verdict_{self.user_email}', + ) + mock_post.assert_not_called() + self.assertEqual(result, {'ok': True, 'reason': 'From Cache'}) + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_fetch_from_service_and_set_cache(self, mock_cache, mock_post): + mock_cache.get.return_value = None + + now = datetime.datetime.now(datetime.timezone.utc) + api_data = { + 'ok': True, + 'cache_until': (now + datetime.timedelta(seconds=60)).isoformat(), + } + mock_response = unittest.mock.MagicMock( + status_code=200, + json=unittest.mock.MagicMock(return_value=api_data), + ) + mock_post.return_value = mock_response + + result = self.service.get_verdict(self.user_email, self.promo_id) + + mock_cache.get.assert_called_once_with( + f'antifraud_verdict_{self.user_email}', + ) + mock_post.assert_called_once() + mock_cache.set.assert_called_once() + key, val = mock_cache.set.call_args[0][:2] + timeout = mock_cache.set.call_args[1]['timeout'] + self.assertEqual(key, f'antifraud_verdict_{self.user_email}') + self.assertEqual(val, api_data) + self.assertAlmostEqual(timeout, 60, delta=1) + self.assertEqual(result, api_data) + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_cache_is_not_set_if_verdict_is_not_ok( + self, + mock_cache, + mock_post, + ): + mock_cache.get.return_value = None + api_data = {'ok': False, 'reason': 'Blocked'} + mock_response = unittest.mock.MagicMock( + status_code=200, + json=unittest.mock.MagicMock(return_value=api_data), + ) + mock_post.return_value = mock_response + + result = self.service.get_verdict(self.user_email, self.promo_id) + + mock_post.assert_called_once() + mock_cache.set.assert_not_called() + self.assertEqual(result, api_data) + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_handles_antifraud_service_unavailable( + self, + mock_cache, + mock_post, + ): + mock_cache.get.return_value = None + mock_post.side_effect = requests.exceptions.RequestException( + 'Connection timed out', + ) + + result = self.service.get_verdict(self.user_email, self.promo_id) + + self.assertEqual(mock_post.call_count, 2) + self.assertEqual( + result, + {'ok': False, 'error': 'Anti-fraud service unavailable'}, + ) + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_does_not_set_cache_with_invalid_date(self, mock_cache, mock_post): + mock_cache.get.return_value = None + api_data = {'ok': True, 'cache_until': 'invalid-date-format'} + mock_response = unittest.mock.MagicMock( + json=unittest.mock.MagicMock(return_value=api_data), + ) + mock_post.return_value = mock_response + + self.service.get_verdict(self.user_email, self.promo_id) + + mock_cache.set.assert_not_called() + + @unittest.mock.patch('user.antifraud_service.requests.post') + @unittest.mock.patch('user.antifraud_service.django.core.cache.cache') + def test_handles_api_http_error(self, mock_cache, mock_post): + mock_cache.get.return_value = None + mock_response = unittest.mock.MagicMock(status_code=500) + + mock_response.raise_for_status.side_effect = ( + requests.exceptions.HTTPError('Server Error') + ) + mock_post.return_value = mock_response + + result = self.service.get_verdict(self.user_email, self.promo_id) + + self.assertEqual(mock_post.call_count, 2) + self.assertEqual( + result, + {'ok': False, 'error': 'Anti-fraud service unavailable'}, + )