from unittest.mock import Mock, patch import jwt from django.test import TestCase from django.urls import reverse from django.utils import timezone from rest_framework.test import APIClient from apps.account.models import User from apps.billing.models import BazaarPurchase, BazaarSettings, GemPack, GemTransaction from apps.billing.providers.bazaar import ( BazaarAuthError, build_authorize_url, exchange_code_for_tokens, verify_purchase, ) from apps.billing.providers.dynamic_discount import ( DynamicDiscountNotConfigured, generate_dynamic_discount_token, ) def _make_response(status_ok=True, json_body=None, status_code=200): response = Mock() response.ok = status_ok response.status_code = status_code response.json.return_value = json_body or {} response.text = str(json_body) response.raise_for_status = Mock() if not status_ok: response.raise_for_status.side_effect = Exception("http error") return response class BazaarStaticTokenVerifyTests(TestCase): """روش جدید — توکن ثابتِ پیشخان، بدون هیچ منطق تمدید.""" def setUp(self): self.settings_obj = BazaarSettings.get_solo() self.settings_obj.access_token = "static-pishkhan-token" self.settings_obj.package_name = "ir.niqow.kangelo" self.settings_obj.save() @patch("apps.billing.providers.bazaar.requests.get") def test_sends_static_token_via_pishkhan_header_not_query_param(self, mock_get): mock_get.return_value = _make_response(json_body={"purchaseState": 0}) result = verify_purchase("gem_pack_100", "purchase-token-abc") self.assertTrue(result) _, kwargs = mock_get.call_args self.assertEqual(kwargs["headers"], {"CAFEBAZAAR-PISHKHAN-API-SECRET": "static-pishkhan-token"}) self.assertNotIn("params", kwargs) @patch("apps.billing.providers.bazaar.requests.get") def test_refund_purchase_state_is_rejected(self, mock_get): mock_get.return_value = _make_response(json_body={"purchaseState": 1}) self.assertFalse(verify_purchase("gem_pack_100", "tok")) @patch("apps.billing.providers.bazaar.requests.get") def test_network_error_fails_closed(self, mock_get): import requests mock_get.side_effect = requests.RequestException("boom") self.assertFalse(verify_purchase("gem_pack_100", "tok")) def test_missing_settings_fails_closed_without_network_call(self): BazaarSettings.objects.all().delete() with patch("apps.billing.providers.bazaar.requests.get") as mock_get: self.assertFalse(verify_purchase("gem_pack_100", "tok")) mock_get.assert_not_called() class BazaarOAuthTests(TestCase): """روش قدیمی — OAuth با client_id/client_secret و تمدید خودکار.""" def setUp(self): self.settings_obj = BazaarSettings.get_solo() self.settings_obj.package_name = "ir.niqow.kangelo" self.settings_obj.client_id = "client-123" self.settings_obj.client_secret = "secret-456" self.settings_obj.save() def test_build_authorize_url_requires_client_id(self): self.settings_obj.client_id = "" self.settings_obj.save() with self.assertRaises(BazaarAuthError): build_authorize_url("https://example.com/callback/") def test_build_authorize_url_shape(self): url = build_authorize_url("https://example.com/callback/") self.assertIn("client_id=client-123", url) self.assertIn("redirect_uri=https://example.com/callback/", url) self.assertIn("response_type=code", url) @patch("apps.billing.providers.bazaar.requests.post") def test_exchange_code_saves_tokens_and_expiry(self, mock_post): mock_post.return_value = _make_response( json_body={"access_token": "fresh-token", "refresh_token": "refresh-token", "expires_in": 3600} ) exchange_code_for_tokens("auth-code", "https://example.com/callback/") self.settings_obj.refresh_from_db() self.assertEqual(self.settings_obj.access_token, "fresh-token") self.assertEqual(self.settings_obj.refresh_token, "refresh-token") self.assertIsNotNone(self.settings_obj.access_token_expires_at) self.assertGreater(self.settings_obj.access_token_expires_at, timezone.now()) @patch("apps.billing.providers.bazaar.requests.post") def test_exchange_code_failure_raises_and_does_not_save(self, mock_post): mock_post.return_value = _make_response(status_ok=False, status_code=400) with self.assertRaises(BazaarAuthError): exchange_code_for_tokens("bad-code", "https://example.com/callback/") self.settings_obj.refresh_from_db() self.assertEqual(self.settings_obj.access_token, "") @patch("apps.billing.providers.bazaar.requests.get") @patch("apps.billing.providers.bazaar.requests.post") def test_verify_purchase_refreshes_expired_token_then_uses_query_param(self, mock_post, mock_get): self.settings_obj.access_token = "stale-token" self.settings_obj.refresh_token = "refresh-token" self.settings_obj.access_token_expires_at = timezone.now() - timezone.timedelta(seconds=5) self.settings_obj.save() mock_post.return_value = _make_response( json_body={"access_token": "brand-new-token", "expires_in": 3600} ) mock_get.return_value = _make_response(json_body={"purchaseState": 0}) result = verify_purchase("sku", "tok") self.assertTrue(result) mock_post.assert_called_once() _, get_kwargs = mock_get.call_args self.assertEqual(get_kwargs["params"], {"access_token": "brand-new-token"}) self.settings_obj.refresh_from_db() self.assertEqual(self.settings_obj.access_token, "brand-new-token") @patch("apps.billing.providers.bazaar.requests.get") def test_verify_purchase_does_not_refresh_when_token_still_fresh(self, mock_get): self.settings_obj.access_token = "still-good" self.settings_obj.refresh_token = "refresh-token" self.settings_obj.access_token_expires_at = timezone.now() + timezone.timedelta(hours=1) self.settings_obj.save() mock_get.return_value = _make_response(json_body={"purchaseState": 0}) with patch("apps.billing.providers.bazaar.requests.post") as mock_post: verify_purchase("sku", "tok") mock_post.assert_not_called() @patch("apps.billing.providers.bazaar.requests.get") def test_verify_purchase_fails_closed_when_refresh_credentials_missing(self, mock_get): self.settings_obj.access_token = "stale" self.settings_obj.refresh_token = "" # نداریم؛ تمدید ممکن نیست self.settings_obj.access_token_expires_at = timezone.now() - timezone.timedelta(seconds=5) self.settings_obj.save() self.assertFalse(verify_purchase("sku", "tok")) mock_get.assert_not_called() class DynamicDiscountTokenTests(TestCase): def setUp(self): self.settings_obj = BazaarSettings.get_solo() self.settings_obj.package_name = "ir.niqow.kangelo" self.settings_obj.dynamic_discount_secret = "top-secret-signing-key" self.settings_obj.save() def test_raises_when_not_configured(self): self.settings_obj.dynamic_discount_secret = "" self.settings_obj.save() with self.assertRaises(DynamicDiscountNotConfigured): generate_dynamic_discount_token(sku="gem_pack_100", price_rial=50000) def test_token_payload_matches_bazaar_spec(self): token = generate_dynamic_discount_token(sku="gem_pack_100", price_rial=50000, account_id="acc-1") payload = jwt.decode(token, "top-secret-signing-key", algorithms=["HS256"]) self.assertEqual(payload["price"], 50000) self.assertEqual(payload["package_name"], "ir.niqow.kangelo") self.assertEqual(payload["sku"], "gem_pack_100") self.assertEqual(payload["account_id"], "acc-1") self.assertIn("nonce", payload) self.assertIn("exp", payload) def test_two_tokens_for_same_input_have_different_nonce(self): token_a = generate_dynamic_discount_token(sku="sku", price_rial=1000) token_b = generate_dynamic_discount_token(sku="sku", price_rial=1000) self.assertNotEqual(token_a, token_b) class BazaarSettingsAdminOAuthViewTests(TestCase): """کانکت/کال‌بک OAuth از پنل ادمین — فقط staff، و کد از querystring هم اگر دستی چسبانده شود (redirect_uri متفاوت) کار می‌کند.""" def setUp(self): self.staff = User.objects.create_user(phone="+989120000000", password="pass") self.staff.is_staff = True self.staff.is_superuser = True self.staff.save() self.client.force_login(self.staff) self.settings_obj = BazaarSettings.get_solo() self.settings_obj.client_id = "client-123" self.settings_obj.client_secret = "secret-456" self.settings_obj.save() def test_connect_view_redirects_to_cafebazaar_when_configured(self): response = self.client.get(reverse("admin:billing_bazaarsettings_oauth_connect")) self.assertEqual(response.status_code, 302) self.assertIn("pardakht.cafebazaar.ir/devapi/v2/auth/authorize", response.url) def test_connect_view_errors_gracefully_when_client_id_missing(self): self.settings_obj.client_id = "" self.settings_obj.save() response = self.client.get(reverse("admin:billing_bazaarsettings_oauth_connect")) self.assertEqual(response.status_code, 302) self.assertIn("/admin/", response.url) def test_callback_without_code_shows_error_not_crash(self): response = self.client.get(reverse("admin:billing_bazaarsettings_oauth_callback")) self.assertEqual(response.status_code, 302) @patch("apps.billing.admin.exchange_code_for_tokens") def test_callback_with_code_exchanges_and_redirects_back(self, mock_exchange): response = self.client.get(reverse("admin:billing_bazaarsettings_oauth_callback"), {"code": "abc"}) self.assertEqual(response.status_code, 302) mock_exchange.assert_called_once() self.assertEqual(mock_exchange.call_args[0][0], "abc") def test_anonymous_cannot_reach_oauth_views(self): self.client.logout() response = self.client.get(reverse("admin:billing_bazaarsettings_oauth_connect")) self.assertNotEqual(response.status_code, 200) class PurchaseFlowIntegrationTests(TestCase): """مسیر کامل و واقعی: ساخت خرید → verify → جم اضافه شود، دوبار-verify دوبار جم ندهد، خرید نامعتبر جم ندهد.""" def setUp(self): self.user = User.objects.create_user(phone="+989121234567", password="pass") self.client = APIClient() self.client.force_authenticate(self.user) self.gem_pack = GemPack.objects.create( bazaar_sku="gem_pack_100", display_name="۱۰۰ جم", gem_amount=100 ) settings_obj = BazaarSettings.get_solo() settings_obj.access_token = "static-token" settings_obj.package_name = "ir.niqow.kangelo" settings_obj.save() def _create_purchase(self, token="purchase-token-1"): response = self.client.post( reverse("billing:purchase-create"), {"gem_pack_id": self.gem_pack.id, "purchase_token": token}, ) self.assertEqual(response.status_code, 201) return response.data["id"] @patch("apps.billing.api.views.verify_purchase", return_value=True) def test_verified_purchase_grants_gems_exactly_once(self, mock_verify): purchase_id = self._create_purchase() response = self.client.post(reverse("billing:purchase-verify", args=[purchase_id])) self.assertEqual(response.status_code, 200) self.assertEqual(response.data["status"], BazaarPurchase.Status.VERIFIED) self.user.refresh_from_db() self.assertEqual(self.user.gem_balance, 100) self.assertEqual( GemTransaction.objects.filter(user=self.user, reason=GemTransaction.Reason.PURCHASE).count(), 1 ) @patch("apps.billing.api.views.verify_purchase", return_value=True) def test_double_verify_does_not_double_credit(self, mock_verify): purchase_id = self._create_purchase() self.client.post(reverse("billing:purchase-verify", args=[purchase_id])) self.client.post(reverse("billing:purchase-verify", args=[purchase_id])) self.user.refresh_from_db() self.assertEqual(self.user.gem_balance, 100) self.assertEqual(GemTransaction.objects.filter(user=self.user).count(), 1) @patch("apps.billing.api.views.verify_purchase", return_value=False) def test_invalid_purchase_grants_no_gems(self, mock_verify): purchase_id = self._create_purchase() response = self.client.post(reverse("billing:purchase-verify", args=[purchase_id])) self.assertEqual(response.status_code, 400) self.user.refresh_from_db() self.assertEqual(self.user.gem_balance, 0) purchase = BazaarPurchase.objects.get(id=purchase_id) self.assertEqual(purchase.status, BazaarPurchase.Status.FAILED) def test_reusing_same_purchase_token_is_idempotent_at_create(self): first_id = self._create_purchase(token="same-token") second_id = self._create_purchase(token="same-token") self.assertEqual(first_id, second_id) self.assertEqual(BazaarPurchase.objects.filter(purchase_token="same-token").count(), 1)