xsl
ecdfbb64c3
POST /quota/tasks/{task_key}/claim：配置驱动任务注册表、
grant_bonus 纯域函数、ledger 幂等键防重、task_bonus_max 封顶（FR-BILL-03/04）。
实现：packages/domain/billing/quota.py 新增 BonusDecision/grant_bonus（mypy strict，
恰好到顶允许、超一格整体拒绝）、grant_idempotency_key、PERIOD_TYPES；
packages/billing/service.py 新增 claim_task_bonus（幂等快捷路径 → 行锁 → 封顶判定
→ 同事务更新 bonus_granted 并写 system_grant 流水，delta=0 且 balance_after 按
remaining 口径；IntegrityError 并发兜底回滚为 40910）；响应模型 TaskBonusClaimResponse
放新建的 packages/billing/schemas.py；路由挂在 apps/api/routers/quota.py，
rules 由 settings.task_bonus_rules 注入（amount 为占位值，产品定稿后改 env）。
测试：新增单测 11 个（封顶边界、amount<=0、不动 used、hard_cap 独立生效、
默认规则形状与 env 覆盖通路）；新增集成 4 个（领取+重复 40910+纯 DB 流水断言、
五类任务全部到账、未知任务 40410、amount=11>max=10 → 40310 且无新流水）。
门禁全绿：ruff check / ruff format --check / mypy 两段 / lint-imports；
单测 180 passed，集成 7 passed。
偏差说明：1) 任务文档提到的 BillingService._load_rule 实际名为 _require_rule，
直接复用现有助手；2) packages/billing/repo.py 无需新增函数（现有
get_ledger_by_idempotency_key/lock_or_create_quota/add_ledger 已覆盖本任务）；
3) 文档设想的「TASK_BONUS_RULES_JSON 空值容错」经 env 源实测不成立——
pydantic-settings 对 dict 字段先做 JSON 解码，空白字符串在 Settings 构造时即
报错（共享校验器/config.py 属禁改，未动）；单测改为覆盖受支持的空配置形态 "{}"。
75 lines
3.0 KiB
Python
75 lines
3.0 KiB
Python
"""grant_bonus 纯域函数：封顶边界与不变量（FR-BILL-03）."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from packages.domain.billing import quota
|
||
|
||
|
||
def test_grant_up_to_the_cap_exactly_is_allowed() -> None:
|
||
# bonus + amount == max：恰好到顶，允许。
|
||
decision = quota.grant_bonus(quota.QuotaState(used=0, base_limit=10, bonus=9), 1, 10)
|
||
assert decision.allowed
|
||
assert decision.bonus == 10
|
||
assert decision.effective_limit == 20
|
||
assert decision.remaining == 20
|
||
|
||
|
||
def test_grant_one_over_the_cap_is_rejected_as_a_whole() -> None:
|
||
# bonus + amount == max + 1：整体拒绝，不做部分发放。
|
||
state = quota.QuotaState(used=4, base_limit=10, bonus=9)
|
||
decision = quota.grant_bonus(state, 2, 10)
|
||
assert not decision.allowed
|
||
# 拒绝时保持发放前的值 —— 调用方不得据此落库。
|
||
assert decision.bonus == 9
|
||
assert decision.remaining == state.remaining
|
||
assert decision.effective_limit == state.effective_limit
|
||
|
||
|
||
def test_zero_cap_rejects_every_grant() -> None:
|
||
# hd_export / report_basic 的 task_bonus_max=0：任务奖励一律拒绝。
|
||
decision = quota.grant_bonus(quota.QuotaState(used=0, base_limit=1), 1, 0)
|
||
assert not decision.allowed
|
||
|
||
|
||
def test_grant_rejects_non_positive_amounts() -> None:
|
||
state = quota.QuotaState(used=0, base_limit=10)
|
||
with pytest.raises(ValueError, match="positive"):
|
||
quota.grant_bonus(state, 0, 10)
|
||
with pytest.raises(ValueError, match="positive"):
|
||
quota.grant_bonus(state, -1, 10)
|
||
|
||
|
||
def test_grant_never_touches_used_or_base_limit() -> None:
|
||
# 任务奖励抬高的是额度不是用量：remaining 的增量恰好等于 amount。
|
||
state = quota.QuotaState(used=7, base_limit=10, bonus=2)
|
||
decision = quota.grant_bonus(state, 3, 10)
|
||
assert decision.allowed
|
||
assert decision.remaining - state.remaining == 3
|
||
assert decision.bonus == 5
|
||
|
||
|
||
def test_task_bonus_cap_is_independent_of_hard_cap() -> None:
|
||
# 两道天花板独立生效：task_bonus_max 放行、hard_cap 把 effective_limit 削回来。
|
||
state = quota.QuotaState(used=0, base_limit=3, bonus=3, hard_cap=5)
|
||
decision = quota.grant_bonus(state, 3, 10)
|
||
assert decision.allowed # bonus 6 <= task_bonus_max 10
|
||
assert decision.effective_limit == 5 # base 3 + bonus 6 被 hard_cap 削到 5
|
||
assert decision.remaining == 5
|
||
|
||
|
||
def test_grant_key_is_derived_per_user_task_and_claim_period() -> None:
|
||
key = quota.grant_idempotency_key(
|
||
user_id=7, task_key="sign_in", claim_period_key="d:2026-08-14"
|
||
)
|
||
assert key == "grant:7:sign_in:d:2026-08-14"
|
||
# 派生键确定：同一周期重复 claim 得到同一个键，唯一约束才能兜底。
|
||
assert key == quota.grant_idempotency_key(
|
||
user_id=7, task_key="sign_in", claim_period_key="d:2026-08-14"
|
||
)
|
||
# 换周期换键：下个周期可以再领。
|
||
assert key != quota.grant_idempotency_key(
|
||
user_id=7, task_key="sign_in", claim_period_key="d:2026-08-15"
|
||
)
|