zycandClaude Fable 5
216a711291
后端: - 新增 celery_health 生成前置闸——无 worker 在线时图片/视频生成入口 一律 503,防"提交到 ARK 后无人轮询、结果悬空+额度冻结"的数据丢失 - 拼接导出:帧率跟随源众数、字幕逐句重映射、原声/BGM 混音修复 - 资金核算审计脚本 + genesis 账目回填 + 滞留预留清理命令 - 接入 yunqi provider 与豆包 TTS 模型(catalog/migrations/bootstrap 命令) 前端:全站页面更新(pipeline/library/products/projects/team/account 等), 新增共享 pager 分页组件 QA:刷新 function-audit 全量输出,新增 full-qa 报告 文档:BP 产品介绍资料、design/CLAUDE.md Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
90 lines
4.4 KiB
Python
90 lines
4.4 KiB
Python
from decimal import Decimal
|
|
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIClient
|
|
|
|
from apps.accounts.models import Team, TeamMember, User
|
|
from apps.ai.models import AITask, ModelConfig, ModelProvider
|
|
from apps.billing.models import CreditAccount, CreditLedger, CreditReservation
|
|
from apps.billing.services.ledger import charge_reserved_credit, release_credit, reserve_credit
|
|
|
|
|
|
class CreditLedgerTests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="owner", password="pass")
|
|
self.team = Team.objects.create(name="Billing Team", owner=self.user)
|
|
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
|
self.account = CreditAccount.objects.create(team=self.team, balance=Decimal("100.0000"))
|
|
self.provider = ModelProvider.objects.create(name="volcengine", display_name="Volcano")
|
|
self.model = ModelConfig.objects.create(
|
|
provider=self.provider,
|
|
name="doubao-seed-2-0-pro-260215",
|
|
display_name="Doubao",
|
|
capability=ModelConfig.Capability.TEXT,
|
|
)
|
|
self.task = AITask.objects.create(
|
|
team=self.team,
|
|
created_by=self.user,
|
|
task_type=AITask.Type.SCRIPT_GENERATION,
|
|
model_config=self.model,
|
|
idempotency_key="billing-test-task",
|
|
estimated_cost=Decimal("10.0000"),
|
|
)
|
|
|
|
def test_reserve_and_charge_credit(self):
|
|
reservation = reserve_credit(team=self.team, user=self.user, task=self.task, amount=Decimal("10.0000"))
|
|
self.account.refresh_from_db()
|
|
|
|
self.assertEqual(reservation.status, CreditReservation.Status.ACTIVE)
|
|
self.assertEqual(self.account.balance, Decimal("100.0000"))
|
|
self.assertEqual(self.account.reserved_balance, Decimal("10.0000"))
|
|
|
|
charge_reserved_credit(reservation=reservation, actual_amount=Decimal("8.0000"))
|
|
self.account.refresh_from_db()
|
|
reservation.refresh_from_db()
|
|
|
|
self.assertEqual(reservation.status, CreditReservation.Status.CHARGED)
|
|
self.assertEqual(self.account.balance, Decimal("92.0000"))
|
|
self.assertEqual(self.account.reserved_balance, Decimal("0.0000"))
|
|
self.assertEqual(CreditLedger.objects.filter(team=self.team).count(), 3)
|
|
|
|
def test_release_reserved_credit(self):
|
|
reservation = reserve_credit(team=self.team, user=self.user, task=self.task, amount=Decimal("10.0000"))
|
|
release_credit(reservation=reservation, reason="model failed")
|
|
self.account.refresh_from_db()
|
|
reservation.refresh_from_db()
|
|
|
|
self.assertEqual(reservation.status, CreditReservation.Status.RELEASED)
|
|
self.assertEqual(self.account.balance, Decimal("100.0000"))
|
|
self.assertEqual(self.account.reserved_balance, Decimal("0.0000"))
|
|
self.assertEqual(CreditLedger.objects.filter(ledger_type=CreditLedger.Type.RELEASE).count(), 1)
|
|
|
|
|
|
class RechargePermissionTests(TestCase):
|
|
"""充值是团队资金操作:仅 owner/admin 可发起,普通成员/访客被 403 拦截。"""
|
|
|
|
def setUp(self):
|
|
self.owner = User.objects.create_user(username="r-owner", password="ownerpass123")
|
|
self.member = User.objects.create_user(username="r-member", password="memberpass123")
|
|
self.team = Team.objects.create(name="Recharge Team", owner=self.owner)
|
|
TeamMember.objects.create(team=self.team, user=self.owner, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE)
|
|
TeamMember.objects.create(team=self.team, user=self.member, role=TeamMember.Role.MEMBER, status=TeamMember.Status.ACTIVE)
|
|
CreditAccount.objects.create(team=self.team, balance=Decimal("0.0000"))
|
|
|
|
def test_owner_can_recharge(self):
|
|
client = APIClient()
|
|
client.force_authenticate(self.owner)
|
|
response = client.post("/api/billing/recharge/", {"amount": "100"}, format="json")
|
|
self.assertEqual(response.status_code, 201)
|
|
account = CreditAccount.objects.get(team=self.team)
|
|
self.assertEqual(account.balance, Decimal("100.0000"))
|
|
|
|
def test_member_cannot_recharge(self):
|
|
client = APIClient()
|
|
client.force_authenticate(self.member)
|
|
response = client.post("/api/billing/recharge/", {"amount": "100"}, format="json")
|
|
self.assertEqual(response.status_code, 403)
|
|
account = CreditAccount.objects.get(team=self.team)
|
|
self.assertEqual(account.balance, Decimal("0.0000")) # 余额未变,越权被拦
|
|
|