65 lines
3.4 KiB
Python
65 lines
3.4 KiB
Python
from unittest import mock
|
|
|
|
from django.test import TestCase
|
|
|
|
from payment import events
|
|
from payment.models import Order, PaymentEvent
|
|
from payment.services import cancel_pending_order, create_order
|
|
|
|
|
|
class PaymentEventTests(TestCase):
|
|
def test_create_order_records_event_once_per_order(self):
|
|
order, _ = create_order(user_id=7, amount=1000, idempotency_key='k-1')
|
|
create_order(user_id=7, amount=1000, idempotency_key='k-1') # همان سفارش، رویدادِ دوم نه
|
|
rows = PaymentEvent.objects.filter(order_id=order.id, event='order_created')
|
|
self.assertEqual(rows.count(), 1)
|
|
self.assertEqual(rows.get().user_id, 7)
|
|
|
|
def test_cancel_records_event(self):
|
|
order, _ = create_order(user_id=7, amount=1000)
|
|
cancel_pending_order(order, staff='tester')
|
|
self.assertTrue(PaymentEvent.objects.filter(order_id=order.id, event='admin_cancel').exists())
|
|
|
|
def test_failure_to_record_does_not_raise(self):
|
|
with mock.patch.object(PaymentEvent.objects, 'create', side_effect=RuntimeError('db')):
|
|
with self.assertLogs('payment.events', level='ERROR'):
|
|
events.record('x', user_id=1)
|
|
self.assertEqual(Order.objects.count(), 0)
|
|
|
|
|
|
class PaymentResultEventTests(TestCase):
|
|
def test_result_is_recorded_without_bazaar_token(self):
|
|
from payment.services import apply_order_payment_result
|
|
order, _ = create_order(user_id=7, amount=1000, provider='bazaar', bazaar_product_sku='10342')
|
|
apply_order_payment_result(order=order, gateway='bazaar', gateway_ref_id='SECRET-TOKEN',
|
|
new_status='paid', event='bazaar_verify')
|
|
row = PaymentEvent.objects.get(order_id=order.id, event='payment_result')
|
|
self.assertEqual(row.detail['new_status'], 'paid')
|
|
self.assertFalse(row.detail['ignored'])
|
|
self.assertNotIn('SECRET-TOKEN', str(row.detail))
|
|
|
|
def test_duplicate_result_is_recorded_as_ignored(self):
|
|
from payment.services import apply_order_payment_result
|
|
order, _ = create_order(user_id=7, amount=1000)
|
|
for _ in range(2):
|
|
apply_order_payment_result(order=order, gateway='zarinpal', gateway_ref_id='R1',
|
|
new_status='paid', event='zarinpal_callback')
|
|
flags = list(PaymentEvent.objects.filter(order_id=order.id, event='payment_result')
|
|
.order_by('id').values_list('detail__ignored', flat=True))
|
|
self.assertEqual(flags, [False, True])
|
|
|
|
def test_zarinpal_callback_records_arrival_and_result(self):
|
|
from payment.zarinpal import ZarinpalVerifyResult
|
|
order, _ = create_order(user_id=7, amount=1000)
|
|
Order.objects.filter(pk=order.pk).update(gateway_authority='A123')
|
|
with mock.patch('payment.zarinpal.verify_payment',
|
|
return_value=ZarinpalVerifyResult(is_valid=True, ref_id='999')):
|
|
self.client.get('/api/payment/zarinpal/callback', {'Authority': 'A123', 'Status': 'OK'})
|
|
got = list(PaymentEvent.objects.filter(order_id=order.id).order_by('id').values_list('event', flat=True))
|
|
self.assertEqual(got, ['order_created', 'callback_received', 'payment_result'])
|
|
|
|
def test_unknown_authority_is_recorded_without_order(self):
|
|
self.client.get('/api/payment/zarinpal/callback', {'Authority': 'NOPE', 'Status': 'OK'})
|
|
row = PaymentEvent.objects.get(event='callback_unknown_authority')
|
|
self.assertIsNone(row.order_id)
|