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') @override_settings( BAZAAR_PISHKHAN_API_SECRET='pishkhan-secret', BAZAAR_REFRESH_TOKEN='', BAZAAR_PACKAGE_NAME='ir.example.app', BAZAAR_API_BASE='https://pardakht.cafebazaar.ir/devapi/v2', ) class BazaarValidateAgainstDocsTests(TestCase): """آدرس‌ها و معنای پاسخ‌ها طبقِ سندِ بازار (خوانده‌شده ۲۰۲۶-۰۹-۲۶).""" def setUp(self): cache.clear() def _resp(self, mock_get, status, body): mock_get.return_value.status_code = status mock_get.return_value.json.return_value = body @patch('payment.bazaar.requests.post') @patch('payment.bazaar.requests.get') def test_inapp_url_and_new_method_header(self, mock_get, mock_post): self._resp(mock_get, 200, {'purchaseState': 0, 'consumptionState': 1}) self.assertTrue(bazaar.verify_purchase(product_sku='monthly', purchase_token='tok').is_valid) self.assertEqual( mock_get.call_args.args[0], 'https://pardakht.cafebazaar.ir/devapi/v2/api/validate/ir.example.app/inapp/monthly/purchases/tok/', ) self.assertEqual(mock_get.call_args.kwargs['headers'], {'CAFEBAZAAR-PISHKHAN-API-SECRET': 'pishkhan-secret'}) mock_post.assert_not_called() # روشِ جدید OAuth نمی‌خواهد @patch('payment.bazaar.requests.get') def test_refunded_is_invalid(self, mock_get): self._resp(mock_get, 200, {'purchaseState': 1}) self.assertFalse(bazaar.verify_purchase(product_sku='s', purchase_token='t').is_valid) @patch('payment.bazaar.requests.get') def test_not_found_is_invalid(self, mock_get): self._resp(mock_get, 404, {'error': 'not_found'}) self.assertFalse(bazaar.verify_purchase(product_sku='s', purchase_token='forged').is_valid) @patch('payment.bazaar.requests.get') def test_our_credential_error_is_503_not_invalid(self, mock_get): # کلیدِ غلط یا سقفِ روزانه نباید خریدِ واقعیِ کاربر را failed کند. for status in (401, 403, 500): self._resp(mock_get, status, {'error': 'invalid_credentials'}) with self.assertRaises(bazaar.BazaarAPIError): bazaar.verify_purchase(product_sku='s', purchase_token='t') @patch('payment.bazaar.requests.get') def test_client_token_cannot_escape_the_path(self, mock_get): self._resp(mock_get, 404, {'error': 'not_found'}) bazaar.verify_purchase(product_sku='s', purchase_token='../../applications/x') self.assertIn('/purchases/..%2F..%2Fapplications%2Fx/', mock_get.call_args.args[0]) @patch('payment.bazaar.requests.get') def test_subscription_url_and_expiry(self, mock_get): now_ms = timezone.now().timestamp() * 1000 self._resp(mock_get, 200, {'validUntilTimestampMsec': now_ms + 86_400_000}) self.assertTrue(bazaar.verify_purchase(product_sku='sub', purchase_token='t', is_subscription=True).is_valid) self.assertEqual( mock_get.call_args.args[0], 'https://pardakht.cafebazaar.ir/devapi/v2/api/applications/ir.example.app/subscriptions/sub/purchases/t', ) self._resp(mock_get, 200, {'validUntilTimestampMsec': now_ms - 1000}) self.assertFalse(bazaar.verify_purchase(product_sku='sub', purchase_token='t', is_subscription=True).is_valid)