This commit is contained in:
Sayyid Hamid Mahdavi 2025-10-30 15:55:58 +03:30
parent ca904081c3
commit 61cb8aaf5a
6 changed files with 438 additions and 19 deletions

View file

@ -8,7 +8,7 @@ from utils.clients.wallet_client import deposit_to_user_wallet_submit, deposit_t
logger = logging.getLogger("promotion_handler")
class PromotionTypeChoices(models.TextChoices):
class ProcessorTypeChoices(models.TextChoices):
PERCENTAGE = 'percentage', _('percentage')
REFERRAL = 'referral', _('referral')

View file

@ -1,4 +1,4 @@
# Generated by Django 5.1.7 on 2025-10-16 15:00
# Generated by Django 5.1.7 on 2025-10-30 10:50
import django.db.models.deletion
import uuid
@ -54,7 +54,7 @@ class Migration(migrations.Migration):
('event_list', models.JSONField(default=list)),
('title', models.CharField(db_index=True, max_length=255, verbose_name='title')),
('description', models.TextField(blank=True, verbose_name='description')),
('promotion_type', models.CharField(choices=[('percentage', 'percentage'), ('referral', 'referral'), ('others', 'others')], db_index=True, max_length=64, verbose_name='promotion type')),
('processor', models.CharField(choices=[('percentage', 'percentage'), ('referral', 'referral'), ('others', 'others')], db_index=True, max_length=64, verbose_name='promotion type')),
('data', models.JSONField(blank=True, default=dict, null=True, verbose_name='data')),
('balance', models.IntegerField(default=0, verbose_name='balance')),
('application', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='+', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)),
@ -88,13 +88,13 @@ class Migration(migrations.Migration):
('uuid', models.UUIDField(db_index=True, default=uuid.uuid4, editable=False, primary_key=True, serialize=False, unique=True)),
('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
('updated_at', models.DateTimeField(auto_now=True, db_index=True)),
('user_uuid', models.UUIDField()),
('state', models.IntegerField(choices=[(0, 'Created'), (1, 'Assigned'), (2, 'Promoted'), (10, 'Expired')], default=0)),
('base_amount', models.IntegerField(blank=True, null=True)),
('promotion_amount', models.IntegerField(blank=True, null=True)),
('data', models.JSONField(blank=True, null=True)),
('application', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='+', to=settings.OAUTH2_PROVIDER_APPLICATION_MODEL)),
('plan', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='promotions', to='promotions.plan')),
('user', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to=settings.AUTH_USER_MODEL)),
('recipient', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.PROTECT, related_name='promotions', to='promotions.recipient')),
],
options={

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 PromotionTypeChoices, BasePromotionHandler
from apps.promotions.handlers import ProcessorTypeChoices, BasePromotionHandler
from apps.users.models import User
from utils.clients.wallet_client import deposit_to_user_wallet_submit, deposit_to_user_wallet_verify
@ -92,8 +92,8 @@ class Plan(BaseModel):
event_list = models.JSONField(default=list)
title = models.CharField(max_length=255, verbose_name=_('title'), db_index=True)
description = models.TextField(verbose_name=_('description'), blank=True)
promotion_type = models.CharField(max_length=64, verbose_name=_('promotion type'), db_index=True,
choices=PromotionTypeChoices.choices) # TODO: processor type
processor = models.CharField(max_length=64, verbose_name=_('processor'), db_index=True,
choices=ProcessorTypeChoices.choices) # TODO: processor type
data = models.JSONField(default=dict, verbose_name=_('data'), null=True, blank=True)
balance = models.IntegerField(verbose_name=_('balance'), default=0)

View file

@ -1,3 +1,426 @@
from django.test import TestCase
import uuid
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 apps.users.models import User
from apps.promotions.models import Plan, Promotion, EventSaver, ProcessorTypeChoices, Event, Recipient
AccessToken = get_access_token_model()
Application = get_application_model()
application_uuid_hex = 'b14e8b86-8f4a-44d9-b29d-badceb470053'
class ApplicationApiFlowsTests(APITestCase):
client_id = '4INGOCMoulE0fNY1SQlTbPtsWqqxGj2DdqjADq6u'
application_uuid = uuid.UUID(application_uuid_hex)
application_access_token = '14naVsdKCbKNOhnElPyXcrwSnqqFbm'
user_access_token = '24naVsdKCbKNOhnElPyXcrwSnqqFbm'
user_uuid = uuid.UUID('b14e8b86-8f4a-44d9-b29d-badceb470051')
scopes = [
'promotions.application.event:submit',
'promotions.user.self-plans:list-retrieve',
'promotions.application.user-promotions:promote',
'promotions.application.user-promotions:list-retrieve',
]
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)
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,
"client_id": self.client_id,
# "client_owner": owner,
"application_id": self.application_uuid,
"scope": ' '.join(self.scopes),
"expires": expire_datetime.isoformat(),
},
)
self.payer_access_token, _created = AccessToken.objects.update_or_create(
token=self.payer_access_token,
defaults={
"user": self.payer_user,
"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,
)
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)
def tearDown(self):
super().tearDown()
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_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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payee_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 0)
# Deposit process
data = {
'amount': 100,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
json_response = response.json()
kwargs_verify = {
'uuid': json_response['uuid'],
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.get(reverse('wallet:application_deposit_verify_api', kwargs=kwargs_verify), HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['state'], 5)
# Desired results
response = self.client.get(reverse('wallet:application_account_list_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'], 200)
params = {
'owner': self.payee_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 100)
def test_deposit_flow_insufficient_balance(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]
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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payee_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 0)
# Deposit process
data = {
'amount': 500,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet),
data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 422)
@override_settings(ALLOWED_NEGATIVE_BALANCE_APPLICATIONS=[application_uuid_hex])
def test_deposit_flow_skip_insufficient_balance_for_allowed_negative_balance_applications(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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payee_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 0)
# Deposit process
data = {
'amount': 500,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet),
data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
def test_deposit_flow_zero_amount(self):
# TODO: wallet
auth = self._create_authorization_header(self.application_access_token.token)
# Deposit process
data = {
'amount': 0,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 400)
def test_deposit_flow_negative_amount(self):
# TODO: wallet
auth = self._create_authorization_header(self.application_access_token.token)
# Deposit process
data = {
'amount': -100,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 400)
def test_deposit_flow_prevent_double_spending(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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payee_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 0)
# Deposit process
data = {
'amount': 200,
'payee': self.payee_uuid,
'payee_type': TypeChoices.USER,
'payee_wallet': self.wallet_rial_uuid,
}
kwargs_payer_wallet = {
'payer_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
response = self.client.post(reverse('wallet:application_deposit_submit_api', kwargs=kwargs_payer_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 422)
def test_withdraw_flow_success(self):
auth = self._create_authorization_header(self.payer_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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payer_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 300)
# Deposit process
data = {
'amount': 100,
'payer': self.payer_uuid,
'payer_type': TypeChoices.USER,
'payer_wallet': self.wallet_rial_uuid,
}
kwargs_payee_wallet = {
'payee_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_withdraw_submit_api', kwargs=kwargs_payee_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
json_response = response.json()
kwargs_payee_wallet_uuid = {
'uuid': json_response['uuid'],
'payee_wallet': self.wallet_rial_uuid,
}
response = self.client.get(reverse('wallet:application_withdraw_verify_api', kwargs=kwargs_payee_wallet_uuid), HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json()['state'], 5)
# Desired results
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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 400)
params = {
'owner': self.payer_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 200)
def test_withdraw_flow_zero_amount(self):
auth = self._create_authorization_header(self.payer_access_token.token)
kwargs_payee_wallet = {
'payee_wallet': self.wallet_rial_uuid,
}
# Deposit process
data = {
'amount': 0,
'payer': self.payer_uuid,
'payer_type': TypeChoices.USER,
'payer_account': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_withdraw_submit_api', kwargs=kwargs_payee_wallet),
data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 400)
def test_withdraw_flow_negative_amount(self):
auth = self._create_authorization_header(self.payer_access_token.token)
kwargs_payee_wallet = {
'payee_wallet': self.wallet_rial_uuid,
}
# Deposit process
data = {
'amount': -500,
'payer': self.payer_uuid,
'payer_type': TypeChoices.USER,
'payer_account': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_withdraw_submit_api', kwargs=kwargs_payee_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 400)
def test_withdraw_flow_prevent_double_spending(self):
auth = self._create_authorization_header(self.payer_access_token.token)
kwargs_wallet_uuid = {
'wallet_uuid': self.wallet_rial_uuid,
}
# 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)
self.assertGreater(response.json()['count'], 1)
from_wallet = response.json()['results'][0]
self.assertEqual(from_wallet['balance'], 300)
params = {
'owner': self.payer_uuid,
}
response = self.client.get(reverse('wallet:application_user_account_list_api', kwargs=params), query_params={'wallet': ','.join(wallets)}, HTTP_AUTHORIZATION=auth)
self.assertGreater(response.json()['count'], 1)
to_wallet = response.json()['results'][0]
self.assertEqual(to_wallet['balance'], 300)
# Deposit process
data = {
'amount': 100,
'payer': self.payer_uuid,
'payer_type': TypeChoices.USER,
'payer_wallet': self.wallet_rial_uuid,
}
kwargs_payee_wallet = {
'payee_wallet': self.wallet_rial_uuid,
}
response = self.client.post(reverse('wallet:application_withdraw_submit_api', kwargs=kwargs_payee_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
response = self.client.post(reverse('wallet:application_withdraw_submit_api', kwargs=kwargs_payee_wallet), data=data, HTTP_AUTHORIZATION=auth)
self.assertEqual(response.status_code, 201)
# Create your tests here.

View file

@ -7,16 +7,11 @@ from django.urls import NoReverseMatch, path, re_path, include
app_name = 'promotions'
application_router = DefaultRouter()
event_router = DefaultRouter()
application_router.register(r'promotion', views.ApplicationPromotionViewSet, basename='application_promotions')
event_router.register(r'events', views.ApplicationEventViewSet, basename='application_events')
user_router = DefaultRouter()
user_router.register('plan', views.UserPlanViewSet, basename='user')
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')
urlpatterns = [
path('api/v1/application/', include(event_router.urls), name='events'),
path('api/v1/application/<user_uuid>/', include(application_router.urls), name='application'),
path('api/v1/user/', include(user_router.urls), name='user'),
path('api/v1/', include(router.urls), name='events'),
]

View file

@ -7,6 +7,7 @@ logger = logging.getLogger(__name__)
def exception_handler(exc, context):
logger.exception(exc)
# پاسخ پیش‌فرض DRF را دریافت می‌کنیم
response = drf_exception_handler(exc, context)