fix: stabilize production payment flow

apps/billing/tests.py

@@ -1,8 +1,12 @@
 import threading
 import time
 import uuid
+import sys
+import tempfile
 from datetime import timedelta
 from decimal import Decimal
+from types import SimpleNamespace
+from unittest.mock import patch
 
 from cryptography.fernet import Fernet
 from django.contrib import admin
@@ -28,6 +32,7 @@ from apps.billing.pricing import (
     get_pricing_rule,
     quote_recharge_points,
 )
+from apps.billing import payment_gateways
 from apps.billing.payment_gateways import PaymentOrderCode
 from apps.billing.services import (
     InsufficientPointsError,
@@ -497,6 +502,97 @@ class BillingServiceTests(TestCase):
         self.assertEqual(order.status, RechargeOrder.Status.FAILED)
         self.assertEqual(order.code_url, "")
 
+    def test_wechat_sdk_native_order_passes_explicit_pay_type(self):
+        order = self.create_recharge_order(amount="1.00", points_granted=10)
+
+        class FakeWeChatPayType:
+            NATIVE = object()
+
+        class FakeWeChatPay:
+            pay_kwargs = None
+
+            def __init__(self, **_kwargs):
+                pass
+
+            def pay(self, **kwargs):
+                FakeWeChatPay.pay_kwargs = kwargs
+                return (
+                    200,
+                    '{"code_url":"weixin://wxpay/bizpayurl?pr=test-ticket"}',
+                )
+
+        fake_module = SimpleNamespace(
+            WeChatPay=FakeWeChatPay,
+            WeChatPayType=FakeWeChatPayType,
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            private_key_path = f"{tmpdir}/apiclient_key.pem"
+            with open(private_key_path, "w", encoding="utf-8") as private_key:
+                private_key.write("test-private-key")
+
+            with (
+                patch.dict(sys.modules, {"wechatpayv3": fake_module}),
+                override_settings(
+                    WECHAT_PAY_APPID="wx-test-appid",
+                    WECHAT_PAY_MCHID="1900000001",
+                    WECHAT_PAY_API_V3_KEY="a" * 32,
+                    WECHAT_PAY_CERT_SERIAL_NO="ABC123",
+                    WECHAT_PAY_PRIVATE_KEY_PATH=private_key_path,
+                    WECHAT_PAY_NOTIFY_URL="https://cm.example.test/api/v1/recharge/callback/wechat",
+                ),
+            ):
+                payment_order = payment_gateways._create_wechat_payment_order_with_sdk(order)
+
+        self.assertEqual(
+            payment_order.code_url,
+            "weixin://wxpay/bizpayurl?pr=test-ticket",
+        )
+        self.assertIs(FakeWeChatPay.pay_kwargs["pay_type"], FakeWeChatPayType.NATIVE)
+
+    def test_wechat_sdk_query_parses_tuple_json_response(self):
+        order = self.create_recharge_order(amount="1.00", points_granted=10)
+
+        class FakeWeChatPay:
+            def __init__(self, **_kwargs):
+                pass
+
+            def query(self, **_kwargs):
+                return (
+                    200,
+                    (
+                        '{"trade_state":"SUCCESS",'
+                        f'"out_trade_no":"{order.order_no}",'
+                        '"amount":{"total":100},'
+                        '"transaction_id":"wx-transaction-001",'
+                        '"success_time":"2026-07-04T17:40:00+08:00"}'
+                    ),
+                )
+
+        fake_module = SimpleNamespace(WeChatPay=FakeWeChatPay)
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            private_key_path = f"{tmpdir}/apiclient_key.pem"
+            with open(private_key_path, "w", encoding="utf-8") as private_key:
+                private_key.write("test-private-key")
+
+            with (
+                patch.dict(sys.modules, {"wechatpayv3": fake_module}),
+                override_settings(
+                    WECHAT_PAY_APPID="wx-test-appid",
+                    WECHAT_PAY_MCHID="1900000001",
+                    WECHAT_PAY_API_V3_KEY="a" * 32,
+                    WECHAT_PAY_CERT_SERIAL_NO="ABC123",
+                    WECHAT_PAY_PRIVATE_KEY_PATH=private_key_path,
+                    WECHAT_PAY_NOTIFY_URL="https://cm.example.test/api/v1/recharge/callback/wechat",
+                ),
+            ):
+                payment = payment_gateways._query_wechat_payment_order_with_sdk(order)
+
+        self.assertEqual(payment.order_no, order.order_no)
+        self.assertEqual(payment.amount, Decimal("1.00"))
+        self.assertEqual(payment.transaction_id, "wx-transaction-001")
+
     def test_precharge_call_debits_wallet_and_writes_pending_call_and_consume_ledger(self):
         charge = precharge_call(
             user=self.user,