Files
MathApp/payment/tests_bazaar_oauth.py
Zareian 8763ffd7bf let the Bazaar verify URL double as the OAuth redirect URI
The Bazaar account owner registered /api/payment/bazaar/verify as the
redirect URI. POST stays the app's purchase check; GET now takes Bazaar's
?code=, exchanges it for a refresh token and stores it in the new
BazaarCredential row, so connecting needs no SSH or .env edit.

The GET is unauthenticated, so it only acts inside a window opened from
the admin (or bazaar_authorize_link): 72 h by default, closed by the first
successful exchange, state checked when Bazaar echoes it. Authorize/token
URL shapes follow the known devapi v2 flow; not checked against Bazaar's
current docs, which did not load.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-26 21:30:44 +03:30

125 lines
5.6 KiB
Python

from datetime import timedelta
from unittest.mock import patch
from urllib.parse import parse_qs, urlsplit
from django.core.cache import cache
from django.test import TestCase, override_settings
from django.utils import timezone
from rest_framework.test import APIClient
from . import bazaar
from .models import BazaarCredential
URL = '/api/payment/bazaar/verify'
@override_settings(
BAZAAR_CLIENT_ID='cid', BAZAAR_CLIENT_SECRET='csecret', BAZAAR_REFRESH_TOKEN='',
BAZAAR_API_BASE='https://pardakht.cafebazaar.ir/devapi/v2',
BAZAAR_REDIRECT_URI='https://api.example/api/payment/bazaar/verify',
)
class BazaarOAuthCallbackTests(TestCase):
"""GETِ verify همان Redirect URIِ بازار است؛ POSTش خریدِ اپ."""
def setUp(self):
cache.clear()
self.client = APIClient()
def _token_response(self, mock_post, status=200, body=None):
mock_post.return_value.status_code = status
mock_post.return_value.json.return_value = body if body is not None else {
'access_token': 'acc', 'refresh_token': 'ref-123', 'expires_in': 3600,
}
def test_authorize_link_points_back_to_verify(self):
q = parse_qs(urlsplit(bazaar.start_authorization()).query)
self.assertEqual(q['redirect_uri'], ['https://api.example/api/payment/bazaar/verify'])
self.assertEqual(q['client_id'], ['cid'])
self.assertEqual(q['state'], [BazaarCredential.objects.get(pk=1).pending_state])
@patch('payment.bazaar.requests.post')
def test_code_is_exchanged_and_stored(self, mock_post):
self._token_response(mock_post)
state = parse_qs(urlsplit(bazaar.start_authorization()).query)['state'][0]
resp = self.client.get(URL, {'code': 'the-code', 'state': state})
self.assertEqual(resp.status_code, 200)
self.assertNotIn('ref-123', resp.content.decode())
sent = mock_post.call_args.kwargs['data']
self.assertEqual(sent['grant_type'], 'authorization_code')
self.assertEqual(sent['code'], 'the-code')
self.assertEqual(sent['redirect_uri'], 'https://api.example/api/payment/bazaar/verify')
self.assertEqual(bazaar.get_refresh_token(), 'ref-123')
cred = BazaarCredential.objects.get(pk=1)
self.assertEqual(cred.pending_state, '')
@patch('payment.bazaar.requests.post')
def test_without_open_window_nothing_is_called(self, mock_post):
resp = self.client.get(URL, {'code': 'attacker-code'})
self.assertEqual(resp.status_code, 400)
mock_post.assert_not_called()
self.assertEqual(bazaar.get_refresh_token(), '')
@patch('payment.bazaar.requests.post')
def test_expired_window_is_refused(self, mock_post):
bazaar.start_authorization()
BazaarCredential.objects.filter(pk=1).update(
pending_since=timezone.now() - timedelta(days=4),
)
self.assertEqual(self.client.get(URL, {'code': 'c'}).status_code, 400)
mock_post.assert_not_called()
@patch('payment.bazaar.requests.post')
def test_wrong_state_is_refused_and_closes_nothing_new(self, mock_post):
bazaar.start_authorization()
self.assertEqual(self.client.get(URL, {'code': 'c', 'state': 'forged'}).status_code, 400)
mock_post.assert_not_called()
@patch('payment.bazaar.requests.post')
def test_window_is_single_use(self, mock_post):
self._token_response(mock_post)
bazaar.start_authorization()
self.assertEqual(self.client.get(URL, {'code': 'c1'}).status_code, 200)
self.assertEqual(self.client.get(URL, {'code': 'c2'}).status_code, 400)
self.assertEqual(mock_post.call_count, 1)
@patch('payment.bazaar.requests.post')
def test_rejected_code_stores_nothing_and_keeps_window_open(self, mock_post):
self._token_response(mock_post, status=400, body={'error': 'invalid_grant'})
bazaar.start_authorization()
self.assertEqual(self.client.get(URL, {'code': 'bad'}).status_code, 400)
self.assertEqual(bazaar.get_refresh_token(), '')
self._token_response(mock_post)
self.assertEqual(self.client.get(URL, {'code': 'good'}).status_code, 200)
self.assertEqual(bazaar.get_refresh_token(), 'ref-123')
@patch('payment.bazaar.requests.post')
def test_two_days_later_the_link_still_works(self, mock_post):
self._token_response(mock_post)
bazaar.start_authorization()
BazaarCredential.objects.filter(pk=1).update(
pending_since=timezone.now() - timedelta(days=2),
)
self.assertEqual(self.client.get(URL, {'code': 'c'}).status_code, 200)
def test_bazaar_error_param_is_reported(self):
self.assertEqual(self.client.get(URL, {'error': 'access_denied'}).status_code, 400)
def test_post_still_requires_auth(self):
resp = self.client.post(URL, {'orderId': '1', 'purchaseToken': 't'}, format='json')
self.assertIn(resp.status_code, (401, 403))
@patch('payment.bazaar.requests.get')
@patch('payment.bazaar.requests.post')
def test_stored_token_is_used_for_purchase_verification(self, mock_post, mock_get):
BazaarCredential.objects.create(pk=1, refresh_token='db-ref')
mock_post.return_value.raise_for_status = lambda: None
mock_post.return_value.json.return_value = {'access_token': 'acc', 'expires_in': 3600}
mock_get.return_value.status_code = 200
mock_get.return_value.json.return_value = {'purchaseState': 0}
self.assertTrue(bazaar.verify_purchase(product_sku='s', purchase_token='t').is_valid)
self.assertEqual(mock_post.call_args.kwargs['data']['refresh_token'], 'db-ref')