addwin/promotions
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

delme

Browse source
This commit is contained in:
Sayyid Hamid Mahdavi 2025-11-02 18:08:47 +03:30
parent 61cb8aaf5a
commit c6ad3cb82d
5 changed files with 122 additions and 60 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
apps/promotions/handlers.py
View file

@ -14,6 +14,10 @@ class ProcessorTypeChoices(models.TextChoices):
OTHERS = 'others', _('others')
class PromotionTypeChoices(models.TextChoices):
pass
class BasePromotionHandler:
def __init__(self, plan):
self.plan = plan

5
apps/promotions/models.py
View file

@ -9,7 +9,7 @@ from oauth2_provider.settings import oauth2_settings
from apps.promotions import handlers
import logging
from apps.promotions.handlers import ProcessorTypeChoices, BasePromotionHandler
from apps.promotions.handlers import ProcessorTypeChoices, BasePromotionHandler, PromotionTypeChoices
from apps.users.models import User
from utils.clients.wallet_client import deposit_to_user_wallet_submit, deposit_to_user_wallet_verify
@ -145,7 +145,8 @@ class Recipient(BaseModel):
elif model_name == 'plan':
obj = plan
else:
raise NotImplementedError()
raise NotImplement
edError()
return obj.data.get(key)
except Exception as e:

132
apps/promotions/tests.py
View file

@ -4,7 +4,7 @@ from datetime import timedelta
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
from rest_framework.test import APITestCase, override_settings, APIClient
from apps.users.models import User
from apps.promotions.models import Plan, Promotion, EventSaver, ProcessorTypeChoices, Event, Recipient
@ -21,6 +21,9 @@ class ApplicationApiFlowsTests(APITestCase):
application_access_token = '14naVsdKCbKNOhnElPyXcrwSnqqFbm'
user_access_token = '24naVsdKCbKNOhnElPyXcrwSnqqFbm'
user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb470051')
first_ad_view_reward_label = 'first_ad_view_reward_label'
referral_reward_label = 'referral_reward_label'
settlement_reward_label = 'settlement_reward_label'
scopes = [
'promotions.application.event:submit',
@ -30,18 +33,16 @@ class ApplicationApiFlowsTests(APITestCase):
]
def setUp(self):
self.application_user, _ = User.objects.get_or_create(pk=self.application_uuid)
self.payer_user, _ = User.objects.get_or_create(pk=self.payer_uuid)
self.application_account, _ = Account.objects.get_or_create(owner_uuid=self.application_uuid, owner_type=TypeChoices.APPLICATION, balance=300)
self.payee_account, _ = Account.objects.get_or_create(owner_uuid=self.payee_uuid, owner_type=TypeChoices.USER)
self.payer_account, _ = Account.objects.get_or_create(owner_uuid=self.payer_uuid, owner_type=TypeChoices.USER, balance=300)
self.user, _ = User.objects.get_or_create(pk=self.user_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={
"user": self.application_user,
"user": self.user,
"client_id": self.client_id,
# "client_owner": owner,
"application_id": self.application_uuid,
@ -50,10 +51,10 @@ class ApplicationApiFlowsTests(APITestCase):
},
)
self.payer_access_token, _created = AccessToken.objects.update_or_create(
token=self.payer_access_token,
self.user_access_token, _created = AccessToken.objects.update_or_create(
token=self.user_access_token,
defaults={
"user": self.payer_user,
"user": self.user,
"client_id": self.client_id,
# "client_owner": owner,
"application_id": self.application_uuid,
@ -66,14 +67,44 @@ class ApplicationApiFlowsTests(APITestCase):
client_id=self.client_id,
uuid=self.application_uuid,
)
self.wallet_rial, _ = Wallet.objects.get_or_create(application=self.application, uuid=self.wallet_rial_uuid)
self.wallet_reward, _ = Wallet.objects.get_or_create(application=self.application, uuid=self.wallet_reward_uuid)
self.application_account1, _ = Account.objects.get_or_create(owner_uuid=self.application_uuid, owner_type=TypeChoices.APPLICATION, balance=300, wallet=self.wallet_rial)
self.application_account2, _ = Account.objects.get_or_create(owner_uuid=self.application_uuid, owner_type=TypeChoices.APPLICATION, balance=300, wallet=self.wallet_reward)
self.payee_account1, _ = Account.objects.get_or_create(owner_uuid=self.payee_uuid, owner_type=TypeChoices.USER, wallet=self.wallet_rial)
self.payee_account2, _ = Account.objects.get_or_create(owner_uuid=self.payee_uuid, owner_type=TypeChoices.USER, wallet=self.wallet_reward)
self.payer_account1, _ = Account.objects.get_or_create(owner_uuid=self.payer_uuid, owner_type=TypeChoices.USER, balance=300, wallet=self.wallet_rial)
self.payer_account2, _ = Account.objects.get_or_create(owner_uuid=self.payer_uuid, owner_type=TypeChoices.USER, balance=300, wallet=self.wallet_reward)
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,
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_uuid,
recipient_uuid_field="",
base_amount_field="",
)
self.referral_reward_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,
application = self.application,
event_list=[self.referral_reward_label]
)
def tearDown(self):
super().tearDown()
@ -81,43 +112,36 @@ class ApplicationApiFlowsTests(APITestCase):
def _create_authorization_header(self, token):
return "Bearer {0}".format(token)
def test_authentication_allow(self):
auth = self._create_authorization_header(self.application_access_token.token)
wallets = [self.wallet_rial_uuid.hex, self.wallet_reward_uuid.hex]
response = self.client.get(reverse('wallet:user_wallet_account_api'), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
def test_first_ad_view_reward_success(self):
auth = self._create_authorization_header(self.user_access_token.token)
event_create_data = {
'label': self.first_ad_view_reward_label,
"data": {
'user': self.user_uuid,
'ad': self.user_uuid
}
}
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')
print(response.json())
# event = Event.objects.get(user=self.user, label=self.first_ad_view_reward_label)
# event.analyze()
# print(event)
return
def test_authentication_expired_token(self):
access_token = '34naVsdKCbKNOhnElPyXcrwSnqqFbm'
expired_access_token, _created = AccessToken.objects.update_or_create(
token=access_token,
defaults={
"user": self.application_user,
"client_id": self.client_id,
# "client_owner": owner,
"application_id": self.application_uuid,
"scope": ' '.join(self.scopes),
"expires": (timezone.now() - timedelta(seconds=3600)).isoformat(),
},
)
auth = self._create_authorization_header(expired_access_token.token)
response = self.client.get(reverse('wallet:user_wallet_account_api'), HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 401)
# def test_authentication_disallow(self):
# auth = self._create_authorization_header('fake_token')
# response = self.client.get(reverse('wallet:user_wallet_balance_api'), HTTP_AUTHORIZATION=auth)
# self.assertEqual(response.status_code, 401)
def test_deposit_flow_success(self):
auth = self._create_authorization_header(self.application_access_token.token)
# Checking wallets initials values(application balance = 300 and payee balance = zero)
wallets = [self.wallet_rial_uuid.hex, self.wallet_reward_uuid.hex]
response = self.client.get(reverse('wallet:application_account_list_api'), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
def test_referral_reward_success(self):
auth = self._create_authorization_header(self.user_access_token.token)
event_create_data = {
'label': self.first_ad_view_reward_label,
"data": {
'user': self.user_uuid,
'referral': self.user_uuid
}
}
response = self.client.post(reverse('promotions:application-events-list'), event_create_data, HTTP_AUTHORIZATION=auth, format='json')
event = Event.objects.get(user=self.user, label=self.first_ad_view_reward_label)
event.analyze()
print(event)
return
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)

6
apps/promotions/urls.py
View file

@ -9,8 +9,10 @@ app_name = 'promotions'
router = DefaultRouter()
router.register(r'application', views.ApplicationPromotionViewSet, basename='application_promotions')
router.register(r'events', views.ApplicationEventViewSet, basename='application_events')
router.register(r'plans', views.UserPlanViewSet, basename='user')
router.register(r'events', views.ApplicationViewSet, basename='application-events')
router.register(r'plans', views.UserViewSet, basename='user')
# urlpatterns =
urlpatterns = [
path('api/v1/', include(router.urls), name='events'),

35
apps/promotions/views.py
View file

@ -63,8 +63,12 @@ class ApplicationPromotionViewSet(
promotion = serializer.save(promotion_amount=promotion_amount, user=user, application=application)
promotion.promote(**promotion_args)
@action(detail=False, methods=['post'], serializer_class=EventSerializer)
def submit_event(self, request):
pass
class ApplicationEventViewSet(
class ApplicationViewSet(
mixins.CreateModelMixin,
GenericViewSet
):
@ -90,15 +94,17 @@ class ApplicationEventViewSet(
raise exceptions.ValidationError(str(e))
analyze_event_task.delay(event.uuid)
serializer.instance = event
class UserPlanViewSet(mixins.RetrieveModelMixin,
class UserViewSet(mixins.RetrieveModelMixin,
mixins.ListModelMixin,
GenericViewSet):
serializer_class = PlanSerializer
permission_classes = [IsAuthenticatedOrTokenMatchesOASRequirements]
required_alternate_scopes = {
"GET": [["promotions.user.self-plans:list-retrieve"]],
"POST": [["promotions.user.self-plans:list-retrieve"]],
}
def get_queryset(self):
@ -119,3 +125,28 @@ class UserPlanViewSet(mixins.RetrieveModelMixin,
plan.calculate_promotion(**serializer.validated_data)
serializer = self.get_serializer(plans, many=True)
return self.get_paginated_response(serializer.data)
@action(detail=True, methods=['POST'], serializer_class=EventSerializer)
def promote(self, request, pk=None):
serializer = self.get_serializer(data=self.request.data)
serializer.is_valid(raise_exception=True)
if hasattr(self.request, 'user'):
user = self.request.user
else:
user = None
application = get_application(self.request)
# EventSaver.objects.first(event_label=serializer.validated_data['event_label'], valid_application__contains=application.uuid)
event_saver: EventSaver = EventSaver.objects.filter(event_label=serializer.validated_data['label']).first()
try:
event = event_saver.save_event(user=user, application=application, **serializer.validated_data)
except Exception as e:
raise exceptions.ValidationError(str(e))
plan: Plan = self.get_object()
plan.process_event(event)
return