unknown
50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
"""admin 领券明细逐场点位分数。"""
|
|
from datetime import date
|
|
|
|
from app.admin.repositories.coupon_data import _point_scores_by_trace
|
|
from app.db.session import SessionLocal
|
|
from app.models.coupon_state import CouponClaimRecord
|
|
|
|
|
|
def test_point_scores_by_trace() -> None:
|
|
"""已领算成功、失败算尝试、跳过不进分母。"""
|
|
db = SessionLocal()
|
|
trace = "point-score-trace"
|
|
try:
|
|
db.add_all([
|
|
CouponClaimRecord(
|
|
device_id="score-device",
|
|
coupon_id=f"mt-score-{status}",
|
|
claim_date=date(2020, 1, 2),
|
|
status=status,
|
|
trace_id=trace,
|
|
)
|
|
for status in ("success", "already_claimed", "failed", "skipped")
|
|
])
|
|
db.flush()
|
|
|
|
assert _point_scores_by_trace(db, [trace]) == {trace: (2, 3)}
|
|
finally:
|
|
db.rollback()
|
|
db.close()
|
|
|
|
|
|
def test_point_scores_omit_trace_without_attempts() -> None:
|
|
"""仅有 skipped 或完全无记录时不伪造 0/0。"""
|
|
db = SessionLocal()
|
|
trace = "point-score-skipped"
|
|
try:
|
|
db.add(CouponClaimRecord(
|
|
device_id="score-device-skipped",
|
|
coupon_id="mt-score-skipped-only",
|
|
claim_date=date(2020, 1, 2),
|
|
status="skipped",
|
|
trace_id=trace,
|
|
))
|
|
db.flush()
|
|
|
|
assert _point_scores_by_trace(db, [trace, "missing-trace"]) == {}
|
|
finally:
|
|
db.rollback()
|
|
db.close()
|