"""斗鱼鱼翅充值渠道分流测试。""" import asyncio import os import unittest from unittest.mock import AsyncMock, Mock, patch os.environ.setdefault("DATABASE_URL", "sqlite://") os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=") from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from web.backend.database import Base from web.backend.models import Account, AuditLog, DouyuTask, LoginTask, User from web.backend.routers.douyu import get_recharge_channel, supplier_recharge_callback from web.backend.services.douyu_runner import DouyuBatchRunner from web.backend.services.douyu_service import ensure_douyu_config from web.backend.services.audit_service import record_audit from core.douyu import FishFinRechargeClient, FishFinRechargeConfig class DouyuGoldRechargeChannelTests(unittest.TestCase): def setUp(self): self.engine = create_engine("sqlite://") Base.metadata.create_all(self.engine) self.session = sessionmaker(bind=self.engine)() user = User(username="operator", password_hash="hash", role="super_admin") self.session.add(user) self.session.commit() self.account = Account( username="douyu-user", password="password", email="mail@example.com", email_password="mail-password", uid="10001", nickname="罗炅729", ) self.session.add(self.account) self.session.commit() self.task = DouyuTask( batch_id="batch", account_id=self.account.id, task_type="create_gold_qr", handbook_scope="elite", status="running", created_by=user.id, ) self.session.add(self.task) self.session.commit() self.runner = DouyuBatchRunner(self.session, "batch", "create_gold_qr") os.environ["FISH_FIN_RECHARGE_BASE_URL"] = "https://supplier.example" os.environ["FISH_FIN_RECHARGE_APP_ID"] = "test-app" os.environ["FISH_FIN_RECHARGE_APP_SECRET"] = "test-secret" def tearDown(self): self.session.close() Base.metadata.drop_all(self.engine) self.engine.dispose() @patch("web.backend.services.douyu_runner.FishFinRechargeClient") def test_supplier_channel_creates_order_using_nickname_and_finishes_on_success(self, client_class): supplier = Mock() supplier.create_order.return_value = { "code": 200, "result": {"order_status": 2, "order_id": "supplier-001"}, "sign": "not-stored", } client_class.return_value = supplier config = { "gold_recharge_channel": "supplier_api", "gold_api_product_id": "gold-product", "gold_api_account_template_name": "斗鱼账号", } self.task.result = {"payload": {"amount": 10}} self.session.commit() self.runner._execute_create_gold_qr( self.session, self.task, self.account, "acf_uid=10001", config, ) expected_order_id = self.runner._supplier_out_order_id(self.task) supplier.create_order.assert_called_once_with( buy_num=10, pay_amount=unittest.mock.ANY, out_order_id=expected_order_id, product_id="gold-product", recharge_arg=[{"templateName": "斗鱼账号", "templateVal": "罗炅729"}], order_type=0, notify_url=unittest.mock.ANY, ) self.assertEqual(str(supplier.create_order.call_args.kwargs["pay_amount"]), "10") self.session.refresh(self.task) self.assertEqual(self.task.status, "success") self.assertEqual(self.task.result["recharge_channel"], "supplier_api") self.assertEqual(self.task.result["supplier_order_status"], 2) self.assertEqual(self.task.result["buy_num"], 10) self.assertEqual(self.task.result["out_order_id"], expected_order_id) self.assertEqual(self.task.result["recharge_account"], "罗炅729") self.assertEqual(self.task.result["douyu_uid"], "10001") self.assertEqual(self.task.result["order_id"], "supplier-001") self.assertEqual(self.task.result["pay_amount"], "10") self.assertNotIn("pay_url", self.task.result) self.assertNotIn("sign", self.task.result["supplier_order"]) @patch("web.backend.services.douyu_runner.FishFinRechargeClient") def test_supplier_channel_rejects_account_without_nickname(self, client_class): self.account.nickname = "" self.session.commit() config = { "gold_recharge_channel": "supplier_api", "gold_api_product_id": "gold-product", "gold_api_account_template_name": "斗鱼昵称", } self.runner._execute_create_gold_qr( self.session, self.task, self.account, "acf_uid=10001", config, ) client_class.assert_not_called() self.session.refresh(self.task) self.assertEqual(self.task.status, "failed") self.assertIn("斗鱼昵称", self.task.message) def test_supplier_out_order_id_reuses_persisted_value(self): self.task.supplier_out_order_id = "DYGFHISTORICAL-001" self.assertEqual( self.runner._supplier_out_order_id(self.task), "DYGFHISTORICAL-001", ) def test_recharge_channel_can_be_read_without_full_config(self): config = ensure_douyu_config(self.session) config.gold_recharge_channel = "supplier_api" self.session.commit() channel = get_recharge_channel( db=self.session, current=self.session.get(User, self.task.created_by), ) self.assertEqual(channel, {"gold_recharge_channel": "supplier_api"}) def test_recharge_audit_detail_includes_safe_account_identity(self): record_audit( self.session, self.session.get(User, self.task.created_by), action="recharge:douyu:create", detail={ "recharge_accounts": [{ "username": self.account.username, "douyu_uid": self.account.uid, "douyu_nickname": self.account.nickname, }], }, ) self.session.commit() audit = self.session.query(AuditLog).one() self.assertIn("douyu_uid", audit.detail) self.assertIn("10001", audit.detail) self.assertNotIn("password", audit.detail) @patch("web.backend.services.douyu_runner.FishFinRechargeClient") def test_supplier_channel_marks_non_200_create_response_as_failed(self, client_class): supplier = Mock() supplier.create_order.return_value = {"code": 400, "msg": "商品已下架"} client_class.return_value = supplier config = { "gold_recharge_channel": "supplier_api", "gold_api_product_id": "gold-product", "gold_api_account_template_name": "斗鱼UID", } self.runner._execute_create_gold_qr( self.session, self.task, self.account, "acf_uid=10001", config, ) self.session.refresh(self.task) self.assertEqual(self.task.status, "failed") self.assertEqual(self.task.message, "商品已下架") @patch("web.backend.services.douyu_runner.check_douyu_cookie") @patch("web.backend.services.douyu_runner.SessionLocal") def test_invalid_cookie_skips_activity_and_marks_relogin_required(self, session_local, check_cookie): login_task = LoginTask( batch_id="login-batch", account_id=self.account.id, created_by=self.task.created_by, status="success", cookie="acf_uid=10001", ) self.task.status = "planned" self.session.add(login_task) self.session.commit() task_id = self.task.id login_task_id = login_task.id session_local.return_value = self.session check_cookie.return_value = { "valid": False, "message": "鱼丸接口: 未登录；等级接口: 未登录", "fish_ball": None, "nickname": None, "level": None, "checked_at": None, } self.runner._execute_create_gold_qr = Mock() self.runner._execute_one(task_id, {}, 1) self.runner._execute_create_gold_qr.assert_not_called() verify_session = sessionmaker(bind=self.engine)() try: task = verify_session.get(DouyuTask, task_id) checked_login = verify_session.get(LoginTask, login_task_id) self.assertEqual(task.status, "failed") self.assertIn("Cookie 已失效，请重新登录", task.message) self.assertEqual(checked_login.ck_check_status, "invalid") self.assertEqual(checked_login.ck_check_result["message"], "鱼丸接口: 未登录；等级接口: 未登录") finally: verify_session.close() def test_supplier_callback_verifies_signature_and_updates_terminal_task(self): self.task.supplier_out_order_id = f"DYGFBATCHT{self.task.id}" self.task.result = {"recharge_channel": "supplier_api", "out_order_id": self.task.supplier_out_order_id} self.session.commit() client = FishFinRechargeClient(FishFinRechargeConfig( base_url="https://supplier.example", app_id="test-app", app_secret="test-secret", )) payload = { "app_id": "test-app", "timestamp": 1786629000, "out_order_id": self.task.supplier_out_order_id, "order_id": "supplier-001", "order_type": 0, "order_status": 2, "create_time": "2026/8/13 21:30", } payload["sign"] = client.sign(payload, "POST") request = Mock() request.json = AsyncMock(return_value=payload) response = asyncio.run(supplier_recharge_callback(request, self.session)) self.assertEqual(response["code"], 200) self.assertEqual(response["message"], "success") self.assertTrue(client.verify_response_sign(response, "POST")) self.session.refresh(self.task) self.assertEqual(self.task.status, "success") self.assertEqual(self.task.message, "供应商直充成功（异步通知）") self.session.refresh(self.account) self.assertEqual(self.account.bind_status, "gold_recharged") self.assertEqual(self.task.result["order_id"], "supplier-001") self.assertEqual(self.task.result["supplier_order_status"], 2) audit = self.session.query(AuditLog).filter(AuditLog.action == "recharge:douyu:callback").one() self.assertEqual(audit.username, "supplier") self.assertTrue(audit.success) self.assertNotIn("sign", audit.detail) # 同一终态回调可重复投递，不会改写已结束任务。 asyncio.run(supplier_recharge_callback(request, self.session)) self.session.refresh(self.task) self.assertEqual(self.task.status, "success") if __name__ == "__main__": unittest.main()