import uuid from datetime import timedelta from unittest.mock import patch from django.urls import reverse from django.utils import timezone from oauth2_provider.models import get_access_token_model, get_application_model from rest_framework.test import APITestCase, override_settings, APIClient from apps.promotions.tasks import analyze_event_task from apps.users.models import User from apps.promotions.models import Plan, Promotion, EventSaver, ProcessorTypeChoices, Event, Recipient, \ PaymentStateChoices AccessToken = get_access_token_model() Application = get_application_model() application_uuid_hex = 'b14e8b86-8f4a-44d9-b29d-badceb470053' def mock_submit_deposit_success(wallet_uuid, data): import uuid as sys_uuid class Tmp(): uuid = sys_uuid.uuid4() data = Tmp() return data def mock_verify_deposit_success(wallet_uuid, transaction_uuid): import uuid as sys_uuid class Tmp(): uuid = sys_uuid.uuid4() state = 5 data = Tmp() return data def mock_submit_withdraw_success(data): import uuid as sys_uuid class Tmp(): uuid = sys_uuid.uuid4() data = Tmp() return data def mock_verify_withdraw_success(uuid): import uuid as sys_uuid class Tmp(): uuid = sys_uuid.uuid4() state = 5 data = Tmp() return data class ApplicationApiFlowsTests(APITestCase): client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u' application_uuid = uuid.UUID(application_uuid_hex) application_access_token = '14naVsdKCbKNOhnElPyXcrwSnqqFbm' user_access_token = '24naVsdKCbKNOhnElPyXcrwSnqqFbm' user_1_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb470051') user_2_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb470052') first_ad_view_reward_label = 'first_ad_view_reward_label' referral_reward_label = 'referral_reward_label' settlement_reward_label = 'settlement_reward_label' serial_reward_label = 'serial_reward_label' scopes = [ 'promotions.application.event:submit', 'promotions.user.self-plans:list-retrieve', 'promotions.application.user-promotions:promote', 'promotions.application.user-promotions:list-retrieve', 'promotions.application.user-plan:promote', 'promotions.application.user-plan:list-retrieve' ] def setUp(self): self.submit_deposit_success_patcher = patch('apps.promotions.models.deposit_to_user_wallet_submit', mock_submit_deposit_success) self.verify_deposit_success_patcher = patch('apps.promotions.models.deposit_to_user_wallet_verify', mock_verify_deposit_success) # self.submit_withdraw_success_patcher = patch('apps.promotions.models.withdraw_from_user_wallet_submit', # mock_submit_withdraw_success) # self.verify_withdraw_success_patcher = patch('apps.promotions.models.withdraw_from_user_wallet_verify', # mock_verify_withdraw_success) self.submit_deposit_success_patcher.start() self.verify_deposit_success_patcher.start() # self.submit_withdraw_success_patcher.start() # self.verify_withdraw_success_patcher.start() self.user_1, _ = User.objects.get_or_create(pk=self.user_1_uuid) self.user_2, _ = User.objects.get_or_create(pk=self.user_2_uuid) expire_datetime = timezone.now() + timedelta(seconds=3600) expire_datetime.isoformat() self.application_access_token, _created = AccessToken.objects.update_or_create( token=self.application_access_token, defaults={ "client_id": self.client_id, # "client_owner": owner, "application_id": self.application_uuid, "scope": ' '.join(self.scopes), "expires": expire_datetime.isoformat(), }, ) self.user_access_token, _created = AccessToken.objects.update_or_create( token=self.user_access_token, defaults={ "user": self.user_1, "client_id": self.client_id, # "client_owner": owner, "application_id": self.application_uuid, "scope": ' '.join(self.scopes), "expires": expire_datetime.isoformat(), }, ) self.application, _created = Application.objects.get_or_create( client_id=self.client_id, uuid=self.application_uuid, ) def tearDown(self): super().tearDown() def _create_authorization_header(self, token): return "Bearer {0}".format(token) def test_first_ad_view_reward_success(self): self.first_ad_view_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.first_ad_view_reward_label, event_label=self.first_ad_view_reward_label, save_once=True, ) self.first_ad_view_reward_plan, _created = Plan.objects.get_or_create( title=self.first_ad_view_reward_label, balance=10000, user=self.user_1, application=self.application, event_list=[self.first_ad_view_reward_label] ) self.first_ad_view_reward_recipient, _created = Recipient.objects.get_or_create( label=self.first_ad_view_reward_label, plan=self.first_ad_view_reward_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field="->event:user", base_amount_field="event:base_amount", ) auth = self._create_authorization_header(self.user_access_token.token) event_create_data = { 'label': self.first_ad_view_reward_label, "data": { 'user': self.user_1_uuid, 'base_amount': 800 } } response = self.client.get( reverse('promotions:promotion-list', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) print(response.status_code) # response = self.client.post(reverse('promotions:user-promote', kwargs={"pk":self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') response = self.client.post( reverse('promotions:promotion-create', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) self.first_ad_view_reward_plan.refresh_from_db() self.assertEqual(self.first_ad_view_reward_plan.balance, 10000 - 800) response = self.client.get( reverse('promotions:promotion-list', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) # response = self.client.post(reverse('promotions:promotion-create', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') # print(response.json()) def test_referral_reward_success(self): self.referral_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.referral_reward_label, event_label=self.referral_reward_label, save_once=True, ) self.referral_reward_plan, _created = Plan.objects.get_or_create( title=self.referral_reward_label, balance=10000, user=self.user_1, application=self.application, event_list=[self.referral_reward_label] ) self.referral_reward_recipient, _created = Recipient.objects.get_or_create( label=self.referral_reward_label, plan=self.referral_reward_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field="->event:referral", base_amount_field="1000", ) auth = self._create_authorization_header(self.user_access_token.token) event_create_data = { 'label': self.referral_reward_label, "data": { 'user': self.user_1_uuid, 'referral': self.user_2_uuid } } response = self.client.post( reverse('promotions:promotion-create', kwargs={"plan": self.referral_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) self.referral_reward_plan.refresh_from_db() self.assertEqual(self.referral_reward_plan.balance, 10000 - 1000) self.assertEqual(response.json()['promotions'][0]['user'], str(self.user_2_uuid)) def test_referral_reward_submit_settlement_success(self): self.referral_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.referral_reward_label, event_label=self.referral_reward_label, save_once=True, ) self.settlement_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.settlement_reward_label, event_label=self.settlement_reward_label, save_once=False, ) self.settlement_reward_plan, _created = Plan.objects.get_or_create( title=self.settlement_reward_label, balance=100000, user=self.user_1, application=self.application, event_list=[self.settlement_reward_label] ) self.settlement_referral_reward_recipient, _created = Recipient.objects.get_or_create( label='settlement_referral_reward_label', plan=self.settlement_reward_plan, wallet_uuid=self.user_1_uuid, # recipient_uuid_field=self.user_2_uuid, recipient_uuid_field=f"QS:Event:label={self.referral_reward_label}&data__referral={self.user_2_uuid}->{self.user_2_uuid}", base_amount_field="event:base_amount", ) auth = self._create_authorization_header(self.user_access_token.token) submit_referral_create_data = { 'label': self.referral_reward_label, "data": { 'user': self.user_1_uuid, 'referral': self.user_2_uuid } } submit_settlement_create_data = { 'label': self.settlement_reward_label, "data": { 'user': self.user_1_uuid, 'settlement': self.user_2_uuid, 'base_amount': 7000 } } self.assertEqual(Event.objects.count(), 0) response = self.client.post(reverse('promotions:event-create'), submit_referral_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) self.settlement_reward_plan.refresh_from_db() self.assertEqual(self.settlement_reward_plan.balance, 100000) self.assertEqual(Event.objects.count(), 1) response = self.client.post( reverse('promotions:promotion-create', kwargs={'plan': str(self.settlement_reward_plan.uuid)}), submit_settlement_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) response = self.client.post( reverse('promotions:promotion-create', kwargs={'plan': str(self.settlement_reward_plan.uuid)}), submit_settlement_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) def test_referral_percentage_reward_submit_settlement_success(self): self.referral_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.referral_reward_label, event_label=self.referral_reward_label, save_once=True, ) self.settlement_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.settlement_reward_label, event_label=self.settlement_reward_label, save_once=False, ) self.settlement_reward_plan, _created = Plan.objects.get_or_create( title=self.settlement_reward_label, balance=100000, user=self.user_1, application=self.application, event_list=[self.settlement_reward_label] ) self.settlement_referral_reward_recipient, _created = Recipient.objects.get_or_create( label='settlement_referral_reward_label', plan=self.settlement_reward_plan, wallet_uuid=self.user_1_uuid, # recipient_uuid_field=self.user_2_uuid, recipient_uuid_field=f"QS:Event:label={self.referral_reward_label}&data__referral={self.user_2_uuid}->{self.user_2_uuid}", base_amount_field="event:base_amount", data={"max_promotion_amount": 0, "promotion_percentage": 50} ) auth = self._create_authorization_header(self.user_access_token.token) submit_referral_create_data = { 'label': self.referral_reward_label, "data": { 'user': self.user_1_uuid, 'referral': self.user_2_uuid } } submit_settlement_create_data = { 'label': self.settlement_reward_label, "data": { 'user': self.user_1_uuid, 'settlement': self.user_2_uuid, 'base_amount': 7000 } } self.assertEqual(Event.objects.count(), 0) response = self.client.post(reverse('promotions:event-create'), submit_referral_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) self.settlement_reward_plan.refresh_from_db() self.assertEqual(self.settlement_reward_plan.balance, 100000) self.assertEqual(Event.objects.count(), 1) response = self.client.post( reverse('promotions:promotion-create', kwargs={'plan': str(self.settlement_reward_plan.uuid)}), submit_settlement_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) response = self.client.post( reverse('promotions:promotion-create', kwargs={'plan': str(self.settlement_reward_plan.uuid)}), submit_settlement_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) def test_serial_ad_view_promotion_success(self): serial_id = 'b14e8b86-8f4a-44d9-b29d-badceb470053' ad_1 = 'b14e8b86-8f4a-44d9-b29d-badceb470054' ad_2 = 'b14e8b86-8f4a-44d9-b29d-badceb470055' ad_3 = 'b14e8b86-8f4a-44d9-b29d-badceb470056' self.serial_ad_view_1_event_saver, _created = EventSaver.objects.get_or_create( title='serial_ad_view_1', event_label='serial_ad_view_1', save_once=True, ) self.serial_ad_view_2_event_saver, _created = EventSaver.objects.get_or_create( title='serial_ad_view_2', event_label='serial_ad_view_2', save_once=True, ) self.serial_ad_view_3_event_saver, _created = EventSaver.objects.get_or_create( title='serial_ad_view_3', event_label='serial_ad_view_3', save_once=True, ) self.serial_promotion_plan, _created = Plan.objects.get_or_create( title=self.serial_reward_label, balance=100000, user=self.user_1, application=self.application, event_list=['serial_ad_view_1', 'serial_ad_view_2', 'serial_ad_view_3'] ) self.serial_ad_view_1_recipient, _created = Recipient.objects.get_or_create( label='serial_ad_view_1_event_saver_label', plan=self.serial_promotion_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field=f"event:label=serial_ad_view_1&event:data__serial={serial_id}->event:user", base_amount_field=300, # data={"max_promotion_amount": 0, "promotion_percentage": 50} ) self.serial_ad_view_2_recipient, _created = Recipient.objects.get_or_create( label='serial_ad_view_2_event_saver_label', plan=self.serial_promotion_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field=f"event:label=serial_ad_view_2&event:data__serial={serial_id}->event:user", # recipient_uuid_field=f"QS:Event:label=serial_ad_view_2&data__user={self.user_1_uuid}&data__serial={serial_id}->user", base_amount_field=200, # data={"max_promotion_amount": 0, "promotion_percentage": 50} ) self.serial_ad_view_3_recipient, _created = Recipient.objects.get_or_create( label='serial_ad_view_3_event_saver_label', plan=self.serial_promotion_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field=f"event:label=serial_ad_view_3&event:data__serial={serial_id}->event:user", # recipient_uuid_field=f"QS:Event:label=serial_ad_view_3&data__user={self.user_1_uuid}&data__serial={serial_id}->user", base_amount_field=100, # data={"max_promotion_amount": 0, "promotion_percentage": 50} ) auth = self._create_authorization_header(self.user_access_token.token) serial_ad_1_event_create_data = { 'label': 'serial_ad_view_1', "data": { 'user': self.user_1_uuid, 'serial': serial_id, 'ad': ad_1 } } serial_ad_2_event_create_data = { 'label': 'serial_ad_view_2', "data": { 'user': self.user_1_uuid, 'serial': serial_id, 'ad': ad_2 } } serial_ad_3_event_create_data = { 'label': 'serial_ad_view_3', "data": { 'user': self.user_1_uuid, 'serial': serial_id, 'ad': ad_3 } } self.assertEqual(Event.objects.count(), 0) response = self.client.post( reverse('promotions:promotion-create', kwargs={"plan": str(self.serial_promotion_plan.pk)}), serial_ad_1_event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) response = self.client.post( reverse('promotions:promotion-create', kwargs={"plan": str(self.serial_promotion_plan.pk)}), serial_ad_2_event_create_data, HTTP_AUTHORIZATION=auth, format='json') # response = self.client.post(reverse('promotions:event-create'), serial_ad_2_event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) # event_2 = Event.objects.first() return response = self.client.post(reverse('promotions:event-create'), serial_ad_1_event_create_data, HTTP_AUTHORIZATION=auth, format='json') event_1 = Event.objects.first() response = self.client.post(reverse('promotions:event-create'), serial_ad_2_event_create_data, HTTP_AUTHORIZATION=auth, format='json') event_2 = Event.objects.first() response = self.client.post(reverse('promotions:event-create'), serial_ad_3_event_create_data, HTTP_AUTHORIZATION=auth, format='json') event_3 = Event.objects.first() analyze_event_task(event_1.uuid) analyze_event_task(event_2.uuid) analyze_event_task(event_3.uuid) print(response.json()) self.serial_promotion_plan.refresh_from_db() self.assertEqual(self.serial_promotion_plan.balance, 100000) self.assertEqual(Event.objects.count(), 3) def test_plan_balance_from_another_promotion_success(self): base_plan, _created = Plan.objects.get_or_create( title='base_plan', balance=0, user=self.user_1, application=self.application, event_list=['advertising:manual'] ) base_promotion, _created = Promotion.objects.get_or_create( user_uuid=self.user_1_uuid, application=self.application, plan=base_plan, state=PaymentStateChoices.SUCCESS, base_amount=10000, promotion_amount=10000 ) promotion_event_saver, _created = EventSaver.objects.get_or_create( title=self.first_ad_view_reward_label, event_label=self.first_ad_view_reward_label, save_once=True, ) promotion_plan, _created = Plan.objects.get_or_create( title='promotion_plan', balance=0, balance_holder=base_promotion.pk, user=self.user_1, application=self.application, event_list=[self.first_ad_view_reward_label] ) self.promotion_recipient, _created = Recipient.objects.get_or_create( label='promotion_recipient', plan=promotion_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field=self.user_2_uuid, base_amount_field=1000, data={} ) auth = self._create_authorization_header(self.user_access_token.token) create_data = { 'label': self.first_ad_view_reward_label, "data": { 'user': self.user_1_uuid, } } self.assertEqual(Event.objects.count(), 0) response = self.client.post(reverse('promotions:promotion-create', kwargs={'plan': str(promotion_plan.uuid)}), create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) promotion_plan.refresh_from_db() base_promotion.refresh_from_db() self.assertEqual(base_promotion.base_amount, 10000) self.assertEqual(base_promotion.promotion_amount, 9000) self.assertEqual(Event.objects.count(), 1) response = self.client.post(reverse('promotions:promotion-create', kwargs={'plan': str(promotion_plan.uuid)}), create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) def test_application_promote_user_success(self): self.first_ad_view_reward_event_saver, _created = EventSaver.objects.get_or_create( title=self.first_ad_view_reward_label, event_label=self.first_ad_view_reward_label, save_once=True, ) self.first_ad_view_reward_plan, _created = Plan.objects.get_or_create( title=self.first_ad_view_reward_label, balance=10000, user=self.user_1, application=self.application, event_list=[self.first_ad_view_reward_label] ) self.first_ad_view_reward_recipient, _created = Recipient.objects.get_or_create( label=self.first_ad_view_reward_label, plan=self.first_ad_view_reward_plan, wallet_uuid=self.user_1_uuid, recipient_uuid_field="->event:user", base_amount_field="event:base_amount", ) auth = self._create_authorization_header(self.application_access_token.token) event_create_data = { 'label': self.first_ad_view_reward_label, "data": { 'user': self.user_1_uuid, 'base_amount': 800 } } url = reverse('promotions-application:plan-promote', kwargs={"pk": self.first_ad_view_reward_plan.uuid, "user_uuid": self.user_1_uuid}) # url = reverse('promotions-application:plan-test', kwargs={"pk": self.first_ad_view_reward_plan.uuid, "user_uuid":self.user_1_uuid}) # response = self.client.get(url, event_create_data, HTTP_AUTHORIZATION=auth, # format='json') # print(response.json()) # print(response.status_code) # response = self.client.post(reverse('promotions:user-promote', kwargs={"pk":self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') response = self.client.post(url, event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) response = self.client.post(url, event_create_data, HTTP_AUTHORIZATION=auth, format='json') print(response.json()) self.first_ad_view_reward_plan.refresh_from_db() self.assertEqual(self.first_ad_view_reward_plan.balance, 10000 - 800) # response = self.client.get(reverse('promotions:promotion-list', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, # format='json') # print(response.json()) # response = self.client.post(reverse('promotions:promotion-create', kwargs={"plan": self.first_ad_view_reward_plan.uuid}), event_create_data, HTTP_AUTHORIZATION=auth, format='json') # print(response.json())