490 lines
26 KiB
Python
490 lines
26 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.get_or_create(name="volcengine", defaults={"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.get_or_create(name="volcengine", defaults={"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)
|
||
# 积分制:支付 ¥100 × 10 = 到账 1000 积分;响应带换算快照
|
||
self.assertEqual(response.json()["credited_points"], "1000")
|
||
account = CreditAccount.objects.get(team=self.team)
|
||
self.assertEqual(account.balance, Decimal("1000.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")) # 余额未变,越权被拦
|
||
|
||
|
||
|
||
class PricingEngineTests(TestCase):
|
||
"""计价引擎(apps/billing/pricing):flat 回落 / 配音阶梯 / 视频毛利与取整 / 最低 1 积分。"""
|
||
|
||
def setUp(self):
|
||
from apps.billing.pricing import invalidate_billing_config_cache
|
||
|
||
invalidate_billing_config_cache()
|
||
provider, _ = ModelProvider.objects.get_or_create(name="volcengine", defaults={"display_name": "V"})
|
||
self.text_model = ModelConfig.objects.create(
|
||
provider=provider, name="pe-text", display_name="T", capability=ModelConfig.Capability.TEXT,
|
||
unit_price=Decimal("10"), metadata={"pricing": {"mode": "flat", "base_cost_yuan": 0.1}},
|
||
)
|
||
self.audio_model = ModelConfig.objects.create(
|
||
provider=provider, name="pe-audio", display_name="A", capability=ModelConfig.Capability.AUDIO,
|
||
metadata={"pricing": {"mode": "per_chars", "chars_per_unit": 500, "points_per_unit": 10, "min_units": 1, "base_cost_yuan_per_unit": 0.25}},
|
||
)
|
||
|
||
def test_flat_uses_unit_price_as_points(self):
|
||
from apps.billing.pricing import quote_flat
|
||
|
||
quote = quote_flat(self.text_model)
|
||
self.assertEqual(quote.points, Decimal("10"))
|
||
self.assertEqual(quote.base_cost_yuan, Decimal("0.1"))
|
||
# 多单位(单图批量按张)
|
||
self.assertEqual(quote_flat(self.text_model, units=3).points, Decimal("30"))
|
||
|
||
def test_flat_fallback_when_unit_price_unset(self):
|
||
from apps.billing.pricing import FLAT_FALLBACK_POINTS, quote_flat
|
||
|
||
self.text_model.unit_price = Decimal("0")
|
||
quote = quote_flat(self.text_model)
|
||
self.assertEqual(quote.points, FLAT_FALLBACK_POINTS)
|
||
|
||
def test_voiceover_char_tiers(self):
|
||
from apps.billing.pricing import quote_voiceover
|
||
|
||
cases = {0: 1, 1: 1, 499: 1, 500: 1, 501: 2, 1500: 3, 1501: 4}
|
||
for chars, units in cases.items():
|
||
quote = quote_voiceover(self.audio_model, char_count=chars)
|
||
self.assertEqual(quote.points, Decimal(units * 10), f"chars={chars}")
|
||
self.assertEqual(quote.base_cost_yuan, Decimal("0.25") * units, f"chars={chars}")
|
||
|
||
def test_video_margin_and_rounding(self):
|
||
from apps.billing.pricing import get_billing_config, quote_video_from_cost
|
||
|
||
cfg = get_billing_config()
|
||
self.assertEqual(cfg.video_margin_multiplier, Decimal("1.50")) # 首发 ×1.5(商业决策)
|
||
# ¥1.85 成本 × 1.5 × 10 = 27.75 → HALF_UP → 28 积分;base_cost 保留 ¥ 原值
|
||
quote = quote_video_from_cost(Decimal("1.85"))
|
||
self.assertEqual(quote.points, Decimal("28"))
|
||
self.assertEqual(quote.base_cost_yuan, Decimal("1.85"))
|
||
|
||
def test_minimum_one_point(self):
|
||
from apps.billing.pricing import quote_video_from_cost, yuan_to_points
|
||
|
||
self.assertEqual(quote_video_from_cost(Decimal("0.001")).points, Decimal("1"))
|
||
self.assertEqual(yuan_to_points(Decimal("0")), Decimal("0")) # 0 成本不强收
|
||
|
||
def test_video_reserve_buffer(self):
|
||
from apps.billing.pricing import video_reserve_amount
|
||
|
||
# 28 × 1.10 = 30.8 → 31 积分
|
||
self.assertEqual(video_reserve_amount(Decimal("28")), Decimal("31"))
|
||
|
||
|
||
class TeamMonthlyLimitTests(TestCase):
|
||
"""Team.monthly_credit_limit 真管控:None/-1/0 均不管控,仅正数=硬上限(含在途预留)。"""
|
||
|
||
def setUp(self):
|
||
self.user = User.objects.create_user(username="tml-owner", password="p")
|
||
self.team = Team.objects.create(name="TML", owner=self.user)
|
||
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
||
CreditAccount.objects.create(team=self.team, balance=Decimal("1000.0000"))
|
||
provider, _ = ModelProvider.objects.get_or_create(name="volcengine", defaults={"display_name": "V"})
|
||
self.model = ModelConfig.objects.create(
|
||
provider=provider, name="tml-m", display_name="M", capability=ModelConfig.Capability.TEXT,
|
||
)
|
||
|
||
def _task(self, key):
|
||
return AITask.objects.create(
|
||
team=self.team, created_by=self.user, task_type=AITask.Type.SCRIPT_GENERATION,
|
||
model_config=self.model, idempotency_key=f"tml-{key}",
|
||
)
|
||
|
||
def _reserve(self, key, amount):
|
||
return reserve_credit(team=self.team, user=self.user, task=self._task(key), amount=Decimal(str(amount)))
|
||
|
||
def test_none_means_no_enforcement(self):
|
||
self.team.monthly_credit_limit = None
|
||
self.team.save(update_fields=["monthly_credit_limit"])
|
||
self._reserve("a", 500) # 不抛
|
||
|
||
def test_minus_one_means_unlimited(self):
|
||
self.team.monthly_credit_limit = Decimal("-1")
|
||
self.team.save(update_fields=["monthly_credit_limit"])
|
||
self._reserve("b", 500) # 不抛
|
||
|
||
def test_zero_means_no_enforcement(self):
|
||
# 0 必须不管控:前端把 null 映射成 0、成员限额 0=不限,存量 0 行若被解释成「冻结」
|
||
# 会在部署当天静默冻结整个团队(review 确认)。冻结消费用 QuotaPolicy(monthly=0)。
|
||
self.team.monthly_credit_limit = Decimal("0")
|
||
self.team.save(update_fields=["monthly_credit_limit"])
|
||
self._reserve("c", 500) # 不抛
|
||
|
||
def test_positive_limit_counts_active_reservations(self):
|
||
self.team.monthly_credit_limit = Decimal("100")
|
||
self.team.save(update_fields=["monthly_credit_limit"])
|
||
reservation = self._reserve("d", 60) # 在途 60
|
||
with self.assertRaisesMessage(ValueError, "团队本月额度不足"):
|
||
self._reserve("e", 50) # 60+50 > 100
|
||
# 结算 40(在途转已扣 40,当月已用 40):40+50 <= 100 → 放行
|
||
charge_reserved_credit(reservation=reservation, actual_amount=Decimal("40"))
|
||
self._reserve("f", 50)
|
||
|
||
|
||
class RechargePointsTests(TestCase):
|
||
"""充值积分语义:¥ × points_per_yuan + bonus_points;metadata 快照支付金额与汇率。"""
|
||
|
||
def setUp(self):
|
||
self.owner = User.objects.create_user(username="rp-owner", password="p")
|
||
self.team = Team.objects.create(name="RP", owner=self.owner)
|
||
TeamMember.objects.create(team=self.team, user=self.owner, role=TeamMember.Role.OWNER)
|
||
CreditAccount.objects.create(team=self.team, balance=Decimal("0"))
|
||
self.client = APIClient()
|
||
self.client.force_authenticate(self.owner)
|
||
|
||
def test_recharge_converts_yuan_to_points_with_bonus(self):
|
||
resp = self.client.post("/api/billing/recharge/", {"amount": "500", "bonus_points": "300"}, format="json")
|
||
self.assertEqual(resp.status_code, 201)
|
||
body = resp.json()
|
||
self.assertEqual(body["credited_points"], "5300")
|
||
self.assertEqual(body["points_per_yuan"], "10.00")
|
||
account = CreditAccount.objects.get(team=self.team)
|
||
self.assertEqual(account.balance, Decimal("5300"))
|
||
ledger = CreditLedger.objects.get(team=self.team, ledger_type=CreditLedger.Type.RECHARGE)
|
||
# 真实支付金额只能从 metadata 读(老流水不可用 amount÷rate 反推)
|
||
self.assertEqual(ledger.metadata["paid_amount"], "500")
|
||
self.assertEqual(ledger.metadata["bonus_points"], "300")
|
||
self.assertEqual(ledger.metadata["points_per_yuan"], "10.00")
|
||
|
||
def test_legacy_bonus_yuan_converted(self):
|
||
resp = self.client.post("/api/billing/recharge/", {"amount": "100", "bonus": "20"}, format="json")
|
||
self.assertEqual(resp.status_code, 201)
|
||
self.assertEqual(resp.json()["credited_points"], "1200") # 100×10 + 20×10
|
||
|
||
|
||
class BillingConfigEndpointTests(TestCase):
|
||
"""公共 GET /api/billing/config/(鉴权)+ admin PATCH(权限/生效/审计)。"""
|
||
|
||
def setUp(self):
|
||
from apps.billing.pricing import invalidate_billing_config_cache
|
||
|
||
invalidate_billing_config_cache()
|
||
self.user = User.objects.create_user(username="bc-user", password="p")
|
||
self.team = Team.objects.create(name="BC", owner=self.user)
|
||
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
||
self.admin = User.objects.create_user(username="bc-admin", password="p", is_platform_admin=True)
|
||
|
||
def test_public_config_requires_auth(self):
|
||
self.assertEqual(APIClient().get("/api/billing/config/").status_code, 401)
|
||
client = APIClient()
|
||
client.force_authenticate(self.user)
|
||
body = client.get("/api/billing/config/").json()
|
||
self.assertEqual(body["currency_unit"], "points")
|
||
self.assertEqual(body["points_per_yuan"], "10.00")
|
||
self.assertEqual(body["video_margin_multiplier"], "1.50")
|
||
|
||
def test_admin_patch_updates_and_takes_effect(self):
|
||
from apps.billing.pricing import get_billing_config, invalidate_billing_config_cache
|
||
|
||
client = APIClient()
|
||
client.force_authenticate(self.admin)
|
||
resp = client.patch("/api/admin/billing-config/", {"video_margin_multiplier": "2.5"}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
invalidate_billing_config_cache()
|
||
self.assertEqual(get_billing_config().video_margin_multiplier, Decimal("2.5"))
|
||
# 非超管 PATCH 拒绝
|
||
stranger = APIClient()
|
||
stranger.force_authenticate(self.user)
|
||
self.assertEqual(stranger.patch("/api/admin/billing-config/", {"points_per_yuan": "1"}, format="json").status_code, 403)
|
||
|
||
|
||
class TeamPriceMultiplierTests(TestCase):
|
||
"""团队差异化调价:全类型统一系数、两步取整、视频结算用下单快照(中途改价不影响在途)。"""
|
||
|
||
def setUp(self):
|
||
from apps.billing.pricing import invalidate_billing_config_cache
|
||
|
||
invalidate_billing_config_cache()
|
||
self.user = User.objects.create_user(username="tpm-owner", password="p")
|
||
self.team = Team.objects.create(name="TPM", owner=self.user, price_multiplier=Decimal("0.80"))
|
||
TeamMember.objects.create(team=self.team, user=self.user, role=TeamMember.Role.OWNER)
|
||
CreditAccount.objects.create(team=self.team, balance=Decimal("10000"))
|
||
provider, _ = ModelProvider.objects.get_or_create(name="volcengine", defaults={"display_name": "V"})
|
||
self.image_model = ModelConfig.objects.create(
|
||
provider=provider, name="tpm-img", display_name="I", capability=ModelConfig.Capability.IMAGE,
|
||
unit_price=Decimal("20"),
|
||
)
|
||
|
||
def test_flat_applies_multiplier(self):
|
||
from apps.billing.pricing import quote_flat
|
||
|
||
# 20 × 0.8 = 16;不传 team = 标准价(零回归)
|
||
quote = quote_flat(self.image_model, team=self.team)
|
||
self.assertEqual(quote.points, Decimal("16"))
|
||
self.assertEqual(quote_flat(self.image_model).points, Decimal("20"))
|
||
# meta.rate 是汇率快照契约:create_ai_task 落 payload.points_per_yuan_snapshot,毛利报表用它防汇率漂移
|
||
self.assertEqual(Decimal(quote.meta["rate"]), Decimal("10"))
|
||
|
||
def test_two_step_rounding(self):
|
||
from apps.billing.pricing import quote_video_from_cost
|
||
|
||
# 挂牌:¥1.85×1.5×10=27.75→28;再 ×0.8=22.4→22(两步取整,与前端镜像逐字一致)
|
||
quote = quote_video_from_cost(Decimal("1.85"), multiplier=Decimal("0.80"))
|
||
self.assertEqual(quote.points, Decimal("22"))
|
||
self.assertEqual(quote.meta["price_multiplier"], "0.80")
|
||
|
||
def test_minimum_one_point_after_discount(self):
|
||
from apps.billing.pricing import apply_team_price
|
||
|
||
self.assertEqual(apply_team_price(Decimal("1"), Decimal("0.10")), Decimal("1"))
|
||
|
||
def test_invalid_multiplier_falls_back_to_one(self):
|
||
from apps.billing.pricing import team_price_multiplier
|
||
|
||
self.team.price_multiplier = Decimal("0")
|
||
self.assertEqual(team_price_multiplier(self.team), Decimal("1"))
|
||
self.assertEqual(team_price_multiplier(None), Decimal("1"))
|
||
|
||
def test_video_settle_uses_snapshot_not_current(self):
|
||
"""下单时 0.8 → 中途管理员改成 2.0 → 结算仍按 0.8 快照(jimeng 同款纪律)。"""
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
from apps.ai.free_video import finalize_free_video, submit_free_video
|
||
from apps.ai.models import AITask
|
||
|
||
provider = MagicMock()
|
||
provider.create_video_task.return_value = {"id": "ark-tpm", "status": "queued"}
|
||
provider.extract_first_media_url.return_value = "http://x/v.mp4"
|
||
patch("apps.ai.services.build_provider", return_value=provider).start()
|
||
patch("apps.ai.tasks.poll_free_video_task.apply_async").start()
|
||
patch("apps.ai.free_video._store_free_video_media").start()
|
||
self.addCleanup(patch.stopall)
|
||
|
||
task = submit_free_video(team=self.team, user=self.user, params={
|
||
"prompt": "测试", "mode": "universal", "model": "doubao-seedance-2-0-260128",
|
||
"aspect_ratio": "16:9", "resolution": "480p", "duration": 4, "references": [],
|
||
})
|
||
self.assertEqual(task.request_payload["price_multiplier"], "0.80")
|
||
# 挂牌 28 × 0.8 = 22.4 → 22 积分
|
||
self.assertEqual(task.estimated_cost, Decimal("22"))
|
||
# 中途涨价到 2.0(在途任务不受影响)
|
||
self.team.price_multiplier = Decimal("2.00")
|
||
self.team.save(update_fields=["price_multiplier"])
|
||
provider.poll_video_task.return_value = {"status": "succeeded", "usage": {"total_tokens": 30000}}
|
||
task = finalize_free_video(task=task)
|
||
# 真实 30000 tokens:挂牌 1.38×1.5×10=20.7→21;×快照0.8=16.8→17(而非 ×2.0=42)
|
||
self.assertEqual(task.status, AITask.Status.SUCCEEDED)
|
||
self.assertEqual(task.actual_cost, Decimal("17"))
|
||
|
||
def test_admin_pricing_endpoint(self):
|
||
admin = User.objects.create_user(username="tpm-admin", password="p", is_platform_admin=True)
|
||
client = APIClient()
|
||
client.force_authenticate(admin)
|
||
resp = client.post(f"/api/admin/teams/{self.team.id}/pricing/", {"price_multiplier": "0.9"}, format="json")
|
||
self.assertEqual(resp.status_code, 200)
|
||
self.assertEqual(resp.json()["price_multiplier"], "0.90")
|
||
# 三位小数走 HALF_UP 而非银行家舍入:0.125 → 0.13(quantize 默认会给 0.12)
|
||
resp = client.post(f"/api/admin/teams/{self.team.id}/pricing/", {"price_multiplier": "0.125"}, format="json")
|
||
self.assertEqual(resp.json()["price_multiplier"], "0.13")
|
||
# 边界与非法值
|
||
self.assertEqual(client.post(f"/api/admin/teams/{self.team.id}/pricing/", {"price_multiplier": "0.05"}, format="json").status_code, 400)
|
||
self.assertEqual(client.post(f"/api/admin/teams/{self.team.id}/pricing/", {"price_multiplier": "NaN"}, format="json").status_code, 400)
|
||
# 非超管 403
|
||
stranger = APIClient()
|
||
stranger.force_authenticate(self.user)
|
||
self.assertEqual(stranger.post(f"/api/admin/teams/{self.team.id}/pricing/", {"price_multiplier": "0.5"}, format="json").status_code, 403)
|
||
|
||
def test_public_config_returns_team_multiplier(self):
|
||
client = APIClient()
|
||
client.force_authenticate(self.user)
|
||
self.assertEqual(client.get("/api/billing/config/").json()["team_price_multiplier"], "0.80")
|