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")