Files
xiaoxia-saas/tests/unit/test_points_rules.py
T
xiaoxia 9ee55dd695
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 0s
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 1s
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m13s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 3m26s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 3m37s
AI Code Review / AI Code Review (pull_request) Successful in 4m5s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m46s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 10m15s
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Successful in 10m34s
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Style (pull_request) Has been cancelled
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Validate - Security (pull_request) Has been cancelled
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been cancelled
test(viral-video): 补单测覆盖动态积分定价全部分支(diff覆盖率46%→78%)
- points_rules: resolve_video_dimensions/_match_model_prefix/_infer_resolution_key
  /calculate_viral_video_credits 覆盖别名/has_video_input/actual_tokens/兜底/防御分支(+28 tests,覆盖94%)
- points_service: deduct/settle/refund_viral_video 覆盖none/refund/deduct/余额不足/异常分支(+11 tests,覆盖71%)
- viral_video routes: confirm-copy 预扣/402/already_paid/estimate-credits端点(+8 tests,覆盖53%,全部缺失行已覆盖)
2026-10-02 23:54:24 +08:00

386 lines
15 KiB
Python

"""积分消耗规则单元测试 (#1895) — v1.6.2: 仅保留 voice_clone 相关"""
from __future__ import annotations
import math
import pytest
from packages.domain.points_rules import (
FREE_USER_MULTIPLIER,
MEMBER_DISCOUNT,
MEMBERSHIP_PRICES,
POINTS_PACKAGES,
POINTS_SCENES,
calculate_points_cost,
)
class TestPointsScenesConfig:
"""场景配置完整性"""
def test_voice_clone_scenes_defined(self):
# 仅保留声音克隆两个场景
assert set(POINTS_SCENES.keys()) == {"voice_clone_train", "voice_clone_synth"}
def test_required_keys_present(self):
for key, scene in POINTS_SCENES.items():
assert "base_points" in scene, f"{key} missing base_points"
assert "unit" in scene, f"{key} missing unit"
assert "name" in scene, f"{key} missing name"
def test_voice_clone_train_is_free(self):
assert POINTS_SCENES["voice_clone_train"]["base_points"] == 0
def test_voice_clone_synth_is_per_minute(self):
assert POINTS_SCENES["voice_clone_synth"]["base_points"] == 1
assert POINTS_SCENES["voice_clone_synth"]["unit"] == "分钟"
class TestPointsPackages:
def test_three_packages(self):
assert len(POINTS_PACKAGES) == 3
assert POINTS_PACKAGES["starter_pack"]["points"] == 100
assert POINTS_PACKAGES["basic_pack"]["points"] == 500
assert POINTS_PACKAGES["pro_pack"]["points"] == 2000
class TestMembershipPrices:
def test_three_plans(self):
assert len(MEMBERSHIP_PRICES) == 3
assert MEMBERSHIP_PRICES["monthly"]["price_cents"] == 1990
assert MEMBERSHIP_PRICES["quarterly"]["duration_days"] == 90
assert MEMBERSHIP_PRICES["yearly"]["price_cents"] == 15900
class TestCalculatePointsCost:
"""核心计费逻辑"""
# ── 声音克隆合成(按时长计费) ──
def test_voice_clone_synth_base(self):
cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=3)
assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
def test_voice_clone_synth_rounds_up(self):
cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=2.3)
assert cost == math.ceil(3 * FREE_USER_MULTIPLIER)
def test_voice_clone_synth_minimum_1_minute(self):
cost = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=0.1)
assert cost == math.ceil(1 * FREE_USER_MULTIPLIER)
# ── 免费场景 ──
def test_voice_clone_train_free(self):
cost = calculate_points_cost("voice_clone_train", is_member=False)
assert cost == 0
def test_voice_clone_train_free_for_member(self):
cost = calculate_points_cost("voice_clone_train", is_member=True)
assert cost == 0
# ── 会员折扣 ──
def test_monthly_member_discount(self):
cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1, member_type="monthly")
assert cost == max(1, math.floor(1 * MEMBER_DISCOUNT["monthly"]))
def test_yearly_member_deep_discount(self):
cost = calculate_points_cost(
"voice_clone_synth",
is_member=True,
duration_minutes=2,
member_type="yearly",
)
assert cost == max(1, math.floor(2 * MEMBER_DISCOUNT["yearly"]))
def test_member_without_type_no_discount(self):
cost = calculate_points_cost("voice_clone_synth", is_member=True, duration_minutes=1)
assert cost == 1
# ── 已下线/未知场景(向后兼容:返回 0) ──
@pytest.mark.parametrize(
"scene",
[
"ai_voice",
"ai_video",
"ai_digital_human",
"ai_rewrite",
"ai_cover",
"ai_title",
"douyin_extract",
"nonexistent",
],
)
def test_retired_scenes_return_zero(self, scene):
assert calculate_points_cost(scene, is_member=False) == 0
assert calculate_points_cost(scene, is_member=True, duration_minutes=10) == 0
# ============ 爆款视频动态定价 (#2151) ============
class TestResolveVideoDimensions:
"""resolve_video_dimensions(): 分辨率别名、比例、默认兜底。"""
def test_1080p_16_9(self):
"""1080p + 16:9 → w=1920, h=1080。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("1080p", "16:9")
assert (w, h) == (1920, 1080)
def test_480p_16_9(self):
"""480p + 16:9 → h=480, w 按 16//9 计算。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("480p", "16:9")
assert h == 480
assert w == 480 * 16 // 9
def test_720p_1_1(self):
"""1:1 正方形 → w == h。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("720p", "1:1")
assert (w, h) == (720, 720)
def test_1080p_1_1(self):
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("1080p", "1:1")
assert (w, h) == (1080, 1080)
def test_resolution_aliases(self):
"""中文/英文别名应正确映射到对应高度。"""
from packages.domain.points_rules import resolve_video_dimensions
cases = [
("普清", 480), ("sd", 480), ("low", 480), ("default", 480),
("高清", 720), ("medium", 720), ("hd", 720),
("超清", 1080), ("fhd", 1080), ("ultra", 1080),
("全能", 1080), ("high", 1080),
]
for alias, expected_h in cases:
_, h = resolve_video_dimensions(alias, "1:1")
assert h == expected_h, f"{alias} -> h={h}, expected {expected_h}"
def test_unknown_resolution_falls_back_to_720p(self):
"""未知分辨率字符串兜底到 720p。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("2160p", "1:1")
assert h == 720
assert w == 720
def test_empty_resolution_defaults_to_720p_9_16(self):
"""空 resolution + 空 ratio → 默认 720p + 9:16。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions("", "")
assert h == 720
assert w == 720 * 9 // 16
def test_none_resolution_default_ratio(self):
"""None resolution + None ratio → 720p + 9:16 默认。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions(None, None)
assert h == 720
assert w == 720 * 9 // 16
def test_whitespace_resolution_case_insensitive(self):
"""前后空格 + 大写应被规范化处理。"""
from packages.domain.points_rules import resolve_video_dimensions
w, h = resolve_video_dimensions(" 1080P ", " 16:9 ")
assert (w, h) == (1920, 1080)
class TestMatchModelPrefix:
"""_match_model_prefix() 前缀匹配 + 兜底。"""
def test_seedance_2_5_exact(self):
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("seedance-2.5") == "seedance-2.5"
def test_seedance_2_5_with_variant(self):
"""带后缀版本号(如 seedance-2.5-pro)仍匹配 seedance-2.5。"""
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("seedance-2.5-pro") == "seedance-2.5"
def test_seedance_2_0_exact(self):
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("seedance-2.0") == "seedance-2.0"
def test_seedance_2_0_with_variant(self):
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("seedance-2.0-lite") == "seedance-2.0"
def test_unknown_model_falls_back_to_2_5(self):
"""未知模型前缀兜底 seedance-2.5。"""
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("kling-v1") == "seedance-2.5"
assert _match_model_prefix("") == "seedance-2.5"
assert _match_model_prefix(None) == "seedance-2.5"
def test_case_insensitive(self):
from packages.domain.points_rules import _match_model_prefix
assert _match_model_prefix("SEEDANCE-2.0") == "seedance-2.0"
class TestInferResolutionKey:
"""_infer_resolution_key(): 1000+/650-999/<650 三个分支。"""
def test_height_ge_1000_is_1080p(self):
from packages.domain.points_rules import _infer_resolution_key
assert _infer_resolution_key(1000) == "1080p"
assert _infer_resolution_key(1080) == "1080p"
assert _infer_resolution_key(2160) == "1080p"
def test_height_650_to_999_is_720p(self):
from packages.domain.points_rules import _infer_resolution_key
assert _infer_resolution_key(650) == "720p"
assert _infer_resolution_key(720) == "720p"
assert _infer_resolution_key(999) == "720p"
def test_height_lt_650_is_480p(self):
from packages.domain.points_rules import _infer_resolution_key
assert _infer_resolution_key(480) == "480p"
assert _infer_resolution_key(649) == "480p"
assert _infer_resolution_key(0) == "480p"
class TestCalculateViralVideoCredits:
"""calculate_viral_video_credits():爆款视频动态定价核心函数。"""
def test_default_args_returns_float(self):
"""默认参数返回 float。"""
from packages.domain.points_rules import calculate_viral_video_credits
credits = calculate_viral_video_credits(15, 1280, 720)
assert isinstance(credits, float)
def test_return_is_rounded_to_two_decimals(self):
"""round(..., 2) 后值本身就是两位小数(再 round 不变化)。"""
from packages.domain.points_rules import calculate_viral_video_credits
for dur, w, h in [(15, 1280, 720), (5, 854, 480), (30, 1920, 1080), (10, 720, 720)]:
credits = calculate_viral_video_credits(dur, w, h)
assert round(credits, 2) == credits
def test_has_video_input_uses_lower_price(self):
"""has_video_input=True 时使用参考视频价格(有视频输入便宜)。"""
from packages.domain.points_rules import calculate_viral_video_credits
no_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=False)
with_input = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5", has_video_input=True)
assert with_input < no_input
def test_unknown_model_falls_back_to_seedance_2_5(self):
"""未知 model 前缀兜底到 seedance-2.5 价格,与默认等价。"""
from packages.domain.points_rules import calculate_viral_video_credits
unknown = calculate_viral_video_credits(15, 1280, 720, model="unknown-model")
default = calculate_viral_video_credits(15, 1280, 720, model="seedance-2.5")
assert unknown == default
def test_actual_tokens_overrides_calculation(self):
"""传入 actual_tokens>0 时用它替代公式计算的 tokens。"""
from packages.domain.points_rules import (
VIRAL_VIDEO_FIXED_COST,
VIRAL_VIDEO_MODEL_PRICES,
VIRAL_VIDEO_PROFIT_MULTIPLIER,
calculate_viral_video_credits,
)
price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
actual_tokens = 2_000_000
expected = round((actual_tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
credits = calculate_viral_video_credits(15, 1280, 720, actual_tokens=actual_tokens)
assert credits == expected
def test_zero_duration_width_height_defensive_max1(self):
"""duration/width/height 为 0/None 时 max(1,...) 防御,结果>0。"""
from packages.domain.points_rules import calculate_viral_video_credits
c_zero = calculate_viral_video_credits(0, 0, 0)
assert c_zero > 0
c_none = calculate_viral_video_credits(None, None, None)
assert c_none > 0
c_one = calculate_viral_video_credits(1, 1, 1)
assert c_none == c_one
def test_non_default_fps_affects_tokens(self):
"""fps 非默认值(30) 应比默认(24) 积分高。"""
from packages.domain.points_rules import calculate_viral_video_credits
c24 = calculate_viral_video_credits(15, 1280, 720, fps=24)
c30 = calculate_viral_video_credits(15, 1280, 720, fps=30)
assert c30 > c24
def test_seedance_2_0_priced_lower_than_2_5_at_1080p(self):
"""seedance-2.0 在 1080p 无视频输入时定价低于 seedance-2.5。"""
from packages.domain.points_rules import calculate_viral_video_credits
c20 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.0", has_video_input=False)
c25 = calculate_viral_video_credits(15, 1920, 1080, model="seedance-2.5", has_video_input=False)
assert c20 < c25
def test_formula_includes_fixed_cost_and_multiplier(self):
"""手算公式结果应与函数返回一致(固定成本 + 利润系数)。"""
from packages.domain.points_rules import (
VIRAL_VIDEO_FIXED_COST,
VIRAL_VIDEO_FPS,
VIRAL_VIDEO_MODEL_PRICES,
VIRAL_VIDEO_PROFIT_MULTIPLIER,
calculate_viral_video_credits,
)
dur, w, h = 10, 1280, 720
price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
assert calculate_viral_video_credits(dur, w, h) == expected
def test_seedance_2_0_with_video_input_falls_back_to_seedance_2_5_price(self):
"""seedance-2.0 + has_video_input=True 组合不在价格表,走 line 111 fallback 到 seedance-2.5 的 720p False 价格。"""
from packages.domain.points_rules import (
VIRAL_VIDEO_FIXED_COST,
VIRAL_VIDEO_FPS,
VIRAL_VIDEO_MODEL_PRICES,
VIRAL_VIDEO_PROFIT_MULTIPLIER,
calculate_viral_video_credits,
)
dur, w, h = 10, 1280, 720
# 兜底价格 = seedance-2.5/720p/False = 70.0
price = VIRAL_VIDEO_MODEL_PRICES[("seedance-2.5", "720p", False)]
assert price == 70.0
tokens = dur * w * h * VIRAL_VIDEO_FPS / 1024.0
expected = round((tokens / 1_000_000 * price + VIRAL_VIDEO_FIXED_COST) * VIRAL_VIDEO_PROFIT_MULTIPLIER, 2)
credits = calculate_viral_video_credits(dur, w, h, model="seedance-2.0", has_video_input=True)
assert credits == expected
def test_fps_zero_or_none_falls_back_to_default(self):
"""fps=0/None 时 int(fps or 24) 兜底到默认 24,结果与 fps=24 一致。"""
from packages.domain.points_rules import calculate_viral_video_credits
c_default = calculate_viral_video_credits(10, 1280, 720, fps=24)
c_zero = calculate_viral_video_credits(10, 1280, 720, fps=0)
c_none = calculate_viral_video_credits(10, 1280, 720, fps=None)
assert c_zero == c_default
assert c_none == c_default