shaguabijia-app-server/tests/test_coupon_slots.py

"""每券成功率(§13):app_env 落库 + coupon_slot_report 聚合。"""
from __future__ import annotations

from datetime import UTC, date, datetime

from sqlalchemy import delete, select

from app.admin.repositories.coupon_data import coupon_slot_report
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimRecord, CouponSession
from app.repositories.coupon_state import record_claims, session_app_env


def test_session_app_env_lookup() -> None:
    db = SessionLocal()
    trace = "slot-env-1"
    try:
        db.add(CouponSession(
            trace_id=trace, device_id="d-slot", status="started", app_env="dev",
            started_at=datetime(2020, 2, 1, tzinfo=UTC), started_date=date(2020, 2, 1),
        ))
        db.commit()
        assert session_app_env(db, trace) == "dev"
        assert session_app_env(db, "no-such-trace") is None
        assert session_app_env(db, None) is None
    finally:
        db.execute(delete(CouponSession).where(CouponSession.trace_id == trace))
        db.commit()
        db.close()


def test_record_claims_stamps_app_env() -> None:
    db = SessionLocal()
    dev = "d-slot-stamp"
    try:
        record_claims(db, dev, None, "t-stamp",
                      [{"coupon_id": "mt_x", "status": "success", "name": "美团券"}],
                      app_env="prod")
        row = db.execute(select(CouponClaimRecord).where(
            CouponClaimRecord.device_id == dev)).scalar_one()
        assert row.app_env == "prod"
        # UPDATE 路:app_env=None 不覆盖已有值
        record_claims(db, dev, None, "t-stamp",
                      [{"coupon_id": "mt_x", "status": "already_claimed", "name": "美团券"}],
                      app_env=None)
        db.expire_all()
        row = db.execute(select(CouponClaimRecord).where(
            CouponClaimRecord.device_id == dev)).scalar_one()
        assert row.app_env == "prod"
        assert row.status == "already_claimed"
    finally:
        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.device_id == dev))
        db.commit()
        db.close()


def _claim(db, device, coupon_id, status, app_env, name=None, d=date(2020, 2, 2)):
    db.add(CouponClaimRecord(
        device_id=device, coupon_id=coupon_id, claim_date=d,
        status=status, coupon_name=name, app_env=app_env,
    ))


def test_coupon_slot_report_aggregates() -> None:
    """按 coupon_id 聚合:tried=成功+失败(skipped 排除)、rate、env 过滤、tried 倒序。"""
    db = SessionLocal()
    try:
        # mt_a(prod):dev1 success + dev2 already_claimed + dev3 failed → 2/3;dev4 skipped 不计
        _claim(db, "sdev1", "mt_a", "success", "prod", "美团外卖红包天天领")
        _claim(db, "sdev2", "mt_a", "already_claimed", "prod", "美团外卖红包天天领")
        _claim(db, "sdev3", "mt_a", "failed", "prod", "美团外卖红包天天领")
        _claim(db, "sdev4", "mt_a", "skipped", "prod", "美团外卖红包天天领")
        _claim(db, "sdev1", "tb_b", "success", "prod", "淘宝券")          # 1/1
        _claim(db, "sdev9", "mt_a", "failed", "dev", "美团外卖红包天天领")  # 仅 dev/全部
        db.commit()

        prod = coupon_slot_report(
            db, date_from="2020-02-02", date_to="2020-02-02", app_env="prod"
        )["items"]
        by_id = {r["coupon_id"]: r for r in prod}
        assert by_id["mt_a"]["tried"] == 3            # skipped 排除
        assert by_id["mt_a"]["succeeded"] == 2
        assert by_id["mt_a"]["success_rate"] == round(2 / 3, 4)
        assert by_id["mt_a"]["platform"] == "meituan-waimai"
        assert by_id["mt_a"]["coupon_name"] == "美团外卖红包天天领"
        assert by_id["tb_b"]["success_rate"] == 1.0
        assert by_id["tb_b"]["platform"] == "taobao-shanguang"
        assert [r["coupon_id"] for r in prod] == ["mt_a", "tb_b"]  # tried 倒序

        dev = coupon_slot_report(
            db, date_from="2020-02-02", date_to="2020-02-02", app_env="dev"
        )["items"]
        assert {r["coupon_id"]: r["tried"] for r in dev} == {"mt_a": 1}
        allenv = coupon_slot_report(
            db, date_from="2020-02-02", date_to="2020-02-02", app_env=None
        )["items"]
        assert {r["coupon_id"]: r["tried"] for r in allenv}["mt_a"] == 4
    finally:
        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 2)))
        db.commit()
        db.close()


def test_coupon_slots_endpoint() -> None:
    from fastapi.testclient import TestClient

    from app.admin.main import admin_app
    from app.admin.repositories import admin_user as admin_repo
    from app.admin.security import create_admin_token

    db = SessionLocal()
    try:
        _claim(db, "epdev1", "mt_ep", "success", "prod", "端点券", d=date(2020, 2, 3))
        _claim(db, "epdev2", "mt_ep", "failed", "prod", "端点券", d=date(2020, 2, 3))
        admin = admin_repo.create_admin(
            db, username="slot_admin", password="pass1234", role="super_admin"
        )
        token, _exp = create_admin_token(admin_id=admin.id, role=admin.role)
        db.commit()
    finally:
        db.close()
    try:
        c = TestClient(admin_app)
        r = c.get(
            "/admin/api/coupon-data/coupons",
            params={"date_from": "2020-02-03", "date_to": "2020-02-03", "app_env": "prod"},
            headers={"Authorization": f"Bearer {token}"},
        )
        assert r.status_code == 200, r.text
        row = next(x for x in r.json()["items"] if x["coupon_id"] == "mt_ep")
        assert row["tried"] == 2 and row["succeeded"] == 1 and row["success_rate"] == 0.5
        assert row["coupon_name"] == "端点券" and row["platform"] == "meituan-waimai"
    finally:
        db = SessionLocal()
        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.claim_date == date(2020, 2, 3)))
        db.commit()
        db.close()