zycandClaude Fable 5
dcc5d7fe50
- 根因: CreditReservation.task 是 OneToOneField(CASCADE), AITask.project 也是 CASCADE
—— 删项目→级联删任务→ACTIVE 预留行被吞, 账户冻结份额永不归还(实测 4 账户漂移共 ¥42,
且随删项目持续增长)
- 修法: billing/signals.py 挂 AITask pre_delete, 删除前先 release_credit 释放(级联同样
触发, release 自带行锁+幂等); apps.py ready() 接线
- 回归测试 x3: 直删任务/删项目级联/已扣费不重复释放
- qa/repair_reserved_drift.py: 存量漂移回填脚本(dry-run 默认, --apply 落 RELEASE 流水),
已对线上 4 账户执行并复验清零
- qa/function-audit/output: 本轮 18 页模拟点击审计报告(0 真死按钮 0 JS 报错)
验证: 后端测试 213/213 通过, django check 无警告
补齐 aab4d7e: 该提交带走了 signals.py 之外的测试轮改动, 信号挂载点与回归测试在本提交收口
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
204 lines
11 KiB
Python
204 lines
11 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)
|
|
|
|
def test_task_delete_releases_active_reservation(self):
|
|
"""回归:CreditReservation.task 是 CASCADE,直接删任务会把 ACTIVE 预留行吞掉,
|
|
reserved_balance 冻结永不归还(审计 I2 冻结不平根因)。pre_delete 信号必须先释放。"""
|
|
reserve_credit(team=self.team, user=self.user, task=self.task, amount=Decimal("10.0000"))
|
|
self.task.delete()
|
|
self.account.refresh_from_db()
|
|
|
|
self.assertEqual(self.account.reserved_balance, Decimal("0.0000"))
|
|
self.assertEqual(self.account.balance, Decimal("100.0000"))
|
|
self.assertEqual(CreditLedger.objects.filter(ledger_type=CreditLedger.Type.RELEASE).count(), 1)
|
|
self.assertFalse(CreditReservation.objects.exists()) # 预留行随任务级联删除
|
|
|
|
def test_project_delete_cascade_releases_reservation(self):
|
|
"""回归:删项目 → AITask.project CASCADE 级联删任务,同样必须先释放 ACTIVE 预留。"""
|
|
from apps.products.models import Product
|
|
from apps.projects.models import Project
|
|
|
|
product = Product.objects.create(team=self.team, created_by=self.user, title="P 商品")
|
|
project = Project.objects.create(team=self.team, created_by=self.user, product=product, name="P")
|
|
task = AITask.objects.create(
|
|
team=self.team, created_by=self.user, project=project,
|
|
task_type=AITask.Type.SCRIPT_GENERATION, model_config=self.model,
|
|
idempotency_key="billing-test-task-proj", estimated_cost=Decimal("5.0000"),
|
|
)
|
|
reserve_credit(team=self.team, user=self.user, task=task, amount=Decimal("5.0000"))
|
|
project.delete()
|
|
self.account.refresh_from_db()
|
|
|
|
self.assertEqual(self.account.reserved_balance, Decimal("0.0000"))
|
|
self.assertTrue(CreditLedger.objects.filter(ledger_type=CreditLedger.Type.RELEASE, reason="任务删除,释放未结预留").exists())
|
|
|
|
def test_task_delete_after_charge_does_not_double_release(self):
|
|
"""已扣费(CHARGED)的预留再删任务,信号不应重复释放(release_credit 幂等 + 信号只挑 ACTIVE)。"""
|
|
reservation = reserve_credit(team=self.team, user=self.user, task=self.task, amount=Decimal("10.0000"))
|
|
charge_reserved_credit(reservation=reservation, actual_amount=Decimal("10.0000"))
|
|
self.task.delete()
|
|
self.account.refresh_from_db()
|
|
|
|
self.assertEqual(self.account.balance, Decimal("90.0000"))
|
|
self.assertEqual(self.account.reserved_balance, Decimal("0.0000"))
|
|
self.assertEqual(CreditLedger.objects.filter(ledger_type=CreditLedger.Type.RELEASE).count(), 0) # 无多余释放
|
|
|
|
|
|
class MemberLimitTests(TestCase):
|
|
"""成员三档额度(每日/每月/累计总额,0=不限)在 reserve_credit 时逐档管控(PMC#17/#18)。"""
|
|
|
|
def setUp(self):
|
|
self.user = User.objects.create_user(username="lim-owner", password="pass12345")
|
|
self.team = Team.objects.create(name="Limit Team", owner=self.user)
|
|
self.member = TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
|
CreditAccount.objects.create(team=self.team, balance=Decimal("1000.0000"))
|
|
self.provider = ModelProvider.objects.create(name="volcengine", display_name="Volcano")
|
|
self.model = ModelConfig.objects.create(
|
|
provider=self.provider, name="m", display_name="M", capability=ModelConfig.Capability.IMAGE,
|
|
)
|
|
|
|
def _task(self, key):
|
|
return AITask.objects.create(
|
|
team=self.team, created_by=self.user, task_type=AITask.Type.PRODUCT_IMAGE,
|
|
model_config=self.model, idempotency_key=key, estimated_cost=Decimal("10.0000"),
|
|
)
|
|
|
|
def test_daily_limit_blocks_over_spend(self):
|
|
self.member.daily_credit_limit = Decimal("15.0000")
|
|
self.member.save(update_fields=["daily_credit_limit"])
|
|
# 第一笔 10 预留通过,第二笔 10 超每日 15 → 被拦
|
|
reserve_credit(team=self.team, user=self.user, task=self._task("d1"), amount=Decimal("10.0000"))
|
|
with self.assertRaises(ValueError):
|
|
reserve_credit(team=self.team, user=self.user, task=self._task("d2"), amount=Decimal("10.0000"))
|
|
|
|
def test_total_limit_blocks_over_spend(self):
|
|
self.member.total_credit_limit = Decimal("12.0000")
|
|
self.member.save(update_fields=["total_credit_limit"])
|
|
reserve_credit(team=self.team, user=self.user, task=self._task("t1"), amount=Decimal("10.0000"))
|
|
with self.assertRaises(ValueError):
|
|
reserve_credit(team=self.team, user=self.user, task=self._task("t2"), amount=Decimal("10.0000"))
|
|
|
|
def test_zero_limit_means_unlimited(self):
|
|
# 三档都 0 = 不限:多笔大额预留都不被额度档拦(只受账户余额约束)
|
|
for i in range(5):
|
|
reserve_credit(team=self.team, user=self.user, task=self._task(f"z{i}"), amount=Decimal("10.0000"))
|
|
|
|
|
|
class LedgerTypeFilterTests(TestCase):
|
|
"""账单流水按类型过滤(R70):count/结果随 ledger_type 变化,未知类型忽略回全量。"""
|
|
|
|
def setUp(self):
|
|
self.owner = User.objects.create_user(username="lf-owner", password="ownerpass123")
|
|
self.team = Team.objects.create(name="Ledger Filter Team", owner=self.owner)
|
|
TeamMember.objects.create(team=self.team, user=self.owner, role=TeamMember.Role.OWNER, status=TeamMember.Status.ACTIVE)
|
|
CreditAccount.objects.create(team=self.team, balance=Decimal("100.0000"))
|
|
for i, ltype in enumerate([CreditLedger.Type.RECHARGE, CreditLedger.Type.CHARGE, CreditLedger.Type.CHARGE]):
|
|
CreditLedger.objects.create(
|
|
team=self.team, user=self.owner, ledger_type=ltype,
|
|
amount=Decimal("10.0000"), balance_after=Decimal("100.0000"), reason=f"row {i}",
|
|
)
|
|
self.client_api = APIClient()
|
|
self.client_api.force_authenticate(self.owner)
|
|
|
|
def test_filter_by_type_changes_count_and_rows(self):
|
|
response = self.client_api.get("/api/billing/ledgers/?ledger_type=charge")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["count"], 2)
|
|
self.assertTrue(all(row["ledger_type"] == "charge" for row in response.data["results"]))
|
|
|
|
response = self.client_api.get("/api/billing/ledgers/?ledger_type=recharge")
|
|
self.assertEqual(response.data["count"], 1)
|
|
self.assertEqual(response.data["results"][0]["ledger_type"], "recharge")
|
|
|
|
def test_unknown_type_returns_all(self):
|
|
response = self.client_api.get("/api/billing/ledgers/?ledger_type=bogus")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.data["count"], 3)
|
|
|
|
|
|
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")) # 余额未变,越权被拦
|
|
|