from unittest.mock import Mock, patch import jwt from django.test import Client, 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.api.services import FeatureAccessService from apps.billing.models import ( BazaarPurchase, BazaarSettings, FeaturePlan, 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) # urlencode مقدار را percent-encode می‌کند — همین درست است، چون # redirect_uri باید به‌شکلِ امنِ query-string فرستاده شود. self.assertIn("redirect_uri=https%3A%2F%2Fexample.com%2Fcallback%2F", 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) class FreeFeatureAccessTests(TestCase): """قوانین انتشار کافه‌بازار: بخشی از امکانات باید بدون خرید در دسترس باشد. این تست‌ها تضمین می‌کنند ابزارهای رایگان همیشه باز بمانند — حتی بعد از پایانِ دوره‌ی آزمایشی.""" def setUp(self): self.user = User.objects.create_user(phone="+989127777777", password="pass") self.client = APIClient() self.client.force_authenticate(self.user) def test_seed_creates_features_and_gem_packs(self): # ریشه‌ی باگی که کافه‌بازار گزارش کرد: نبودِ داده‌ی پایه یعنی هر # ابزار «فعال نیست» نشان داده می‌شود. self.assertGreater(FeaturePlan.objects.count(), 0) self.assertGreater(GemPack.objects.filter(is_active=True).count(), 0) def test_at_least_one_feature_is_free(self): self.assertTrue( FeaturePlan.objects.filter(is_free=True).exists(), "حداقل یک ابزار باید رایگان باشد وگرنه انتشار رد می‌شود.", ) def test_free_feature_is_active_without_trial_or_gems(self): self.user.trial_expires_at = timezone.now() - timezone.timedelta(days=1) self.user.gem_balance = 0 self.user.save() self.assertFalse(self.user.is_in_trial) for feature in FeaturePlan.objects.filter(is_free=True): self.assertTrue( FeatureAccessService.is_active(self.user, feature.key), f"ابزار رایگان {feature.key} باید باز باشد.", ) def test_paid_feature_is_locked_after_trial_ends(self): self.user.trial_expires_at = timezone.now() - timezone.timedelta(days=1) self.user.save() paid = FeaturePlan.objects.filter(is_free=False).first() self.assertIsNotNone(paid) self.assertFalse(FeatureAccessService.is_active(self.user, paid.key)) def test_paid_feature_is_open_during_trial(self): self.user.trial_expires_at = timezone.now() + timezone.timedelta(days=3) self.user.save() paid = FeaturePlan.objects.filter(is_free=False).first() self.assertTrue(FeatureAccessService.is_active(self.user, paid.key)) def test_features_endpoint_reports_free_flag(self): response = self.client.get(reverse("billing:feature-list")) self.assertEqual(response.status_code, 200) self.assertGreater(len(response.data), 0) free = [f for f in response.data if f["is_free"]] self.assertGreater(len(free), 0) for f in free: self.assertTrue(f["is_active"]) def test_free_features_have_no_price_tiers(self): # ابزارِ رایگان نباید در فروشگاه قیمت نشان دهد. for feature in FeaturePlan.objects.filter(is_free=True): self.assertEqual(feature.tiers.count(), 0) def test_expected_free_and_paid_split(self): """ترکیبِ تصمیم‌گرفته‌شده‌ی محصول — اگر کسی سهوا عوضش کند، همین‌جا گیر می‌افتد (چون رایگان‌بودنِ این سه، شرطِ انتشار در کافه‌بازار است).""" free = set(FeaturePlan.objects.filter(is_free=True).values_list("key", flat=True)) paid = set(FeaturePlan.objects.filter(is_free=False).values_list("key", flat=True)) self.assertEqual(free, {"customers", "sms_bank", "invoice"}) self.assertEqual(paid, {"ledger", "inventory", "orders"}) def test_paid_features_have_price_tiers(self): for feature in FeaturePlan.objects.filter(is_free=False, is_purchasable=True): self.assertGreater( feature.tiers.count(), 0, f"{feature.key} قیمت ندارد." ) class GuestFeatureCatalogTests(TestCase): """کاربرِ مهمان (بدون حساب) باید بتواند کاتالوگ ابزارها را ببیند و بفهمد کدام رایگان است — وگرنه پیش از ورود همه‌چیز قفل به‌نظر می‌رسد.""" def setUp(self): self.client = APIClient() # عمدا بدون احراز هویت def test_anonymous_can_list_features(self): response = self.client.get(reverse("billing:feature-list")) self.assertEqual(response.status_code, 200) self.assertGreater(len(response.data), 0) def test_free_features_report_active_for_anonymous(self): response = self.client.get(reverse("billing:feature-list")) by_key = {f["key"]: f for f in response.data} for key in ["customers", "sms_bank", "invoice"]: self.assertTrue(by_key[key]["is_free"], key) self.assertTrue(by_key[key]["is_active"], key) def test_paid_features_are_inactive_for_anonymous(self): response = self.client.get(reverse("billing:feature-list")) by_key = {f["key"]: f for f in response.data} for key in ["ledger", "inventory", "orders"]: self.assertFalse(by_key[key]["is_active"], key) def test_anonymous_gets_no_expiry(self): response = self.client.get(reverse("billing:feature-list")) for f in response.data: self.assertIsNone(f["expires_at"]) def test_wallet_still_requires_auth(self): # کاتالوگ عمومی شد، ولی داده‌ی شخصی نه. response = self.client.get(reverse("billing:wallet")) self.assertIn(response.status_code, (401, 403)) class GemGiftAdminTests(TestCase): """هدیه‌ی جم از پنل ادمین — موجودی و تاریخچه باید با هم جلو بروند.""" def setUp(self): self.staff = User.objects.create_superuser(phone="09120000001", password="x") self.client = Client() self.client.force_login(self.staff) self.url = reverse("admin:billing_gemtransaction_gift") def test_phone_normalization_accepts_the_shapes_admins_actually_paste(self): from apps.billing.forms import normalize_iran_phone for raw in ["09121234567", "+989121234567", "00989121234567", "9121234567", "۰۹۱۲۱۲۳۴۵۶۷", "0912 123 4567"]: self.assertEqual(normalize_iran_phone(raw), "09121234567", raw) def test_gift_adds_gems_and_records_history(self): target = User.objects.create(phone="09121112233") response = self.client.post( self.url, {"phones": "09121112233", "amount": 250, "note": "جبران خطا"}, ) self.assertEqual(response.status_code, 302) target.refresh_from_db() self.assertEqual(target.gem_balance, 250) tx = GemTransaction.objects.get(user=target) self.assertEqual(tx.amount, 250) self.assertEqual(tx.reason, GemTransaction.Reason.ADMIN_ADJUSTMENT) self.assertEqual(tx.note, "جبران خطا") def test_gift_handles_several_numbers_in_one_go(self): a = User.objects.create(phone="09121112233") b = User.objects.create(phone="09124445566") self.client.post( self.url, {"phones": "09121112233\n+989124445566", "amount": 100}, ) a.refresh_from_db() b.refresh_from_db() self.assertEqual((a.gem_balance, b.gem_balance), (100, 100)) def test_unknown_number_is_skipped_but_the_rest_still_get_gems(self): known = User.objects.create(phone="09121112233") self.client.post( self.url, {"phones": "09121112233\n09129998877", "amount": 50}, ) known.refresh_from_db() self.assertEqual(known.gem_balance, 50) self.assertFalse(User.objects.filter(phone="09129998877").exists()) def test_create_missing_makes_the_user_so_the_gift_waits_for_them(self): self.client.post( self.url, {"phones": "09129998877", "amount": 75, "create_missing": "on"}, ) created = User.objects.get(phone="09129998877") self.assertEqual(created.gem_balance, 75) def test_invalid_number_rejects_the_whole_form_rather_than_half_applying(self): User.objects.create(phone="09121112233") response = self.client.post( self.url, {"phones": "09121112233\n12345", "amount": 10} ) self.assertEqual(response.status_code, 200) self.assertEqual(GemTransaction.objects.count(), 0)