shaguabijia-app-server/tests/test_coupon_point_score.py

"""admin 领券明细逐场点位分数与按需明细。"""
from datetime import UTC, date, datetime

from fastapi.testclient import TestClient
from sqlalchemy import delete

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.admin.repositories.coupon_data import (
    _point_scores_by_trace,
    coupon_data_report,
    coupon_point_details,
)
from app.admin.security import create_admin_token
from app.db.session import SessionLocal
from app.models.coupon_state import CouponClaimRecord, CouponSession


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,
                coupon_name=f"测试点位-{status}",
                reason="测试失败" if status == "failed" else None,
                trace_id=trace,
            )
            for status in ("success", "already_claimed", "failed", "skipped")
        ])
        db.flush()

        stats = _point_scores_by_trace(db, [trace])[trace]
        assert stats["succeeded"] == 2
        assert stats["tried"] == 3
        details = coupon_point_details(db, trace_id=trace)
        assert [item["status"] for item in details] == [
            "success", "already_claimed", "failed", "skipped"
        ]
        assert details[2]["reason"] == "测试失败"
    finally:
        db.rollback()
        db.close()


def test_skipped_detail_does_not_create_a_score() -> 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()

        scores = _point_scores_by_trace(db, [trace, "missing-trace"])
        assert trace not in scores
        assert "missing-trace" not in scores
        assert coupon_point_details(db, trace_id=trace)[0]["status"] == "skipped"
    finally:
        db.rollback()
        db.close()


def test_coupon_data_report_returns_scores_without_embedding_details() -> None:
    """主列表只返回聚合分数，逐券记录必须走按 trace 的明细查询。"""
    db = SessionLocal()
    trace = "point-score-report"
    report_date = date(2020, 1, 4)
    try:
        db.add(CouponSession(
            trace_id=trace,
            device_id="score-report-device",
            status="completed",
            app_env="prod",
            platforms=["meituan-waimai"],
            started_at=datetime(2020, 1, 4, tzinfo=UTC),
            started_date=report_date,
        ))
        db.add_all([
            CouponClaimRecord(
                device_id="score-report-device",
                coupon_id=f"mt-report-{status}",
                claim_date=report_date,
                status=status,
                trace_id=trace,
            )
            for status in ("success", "failed")
        ])
        db.flush()

        report = coupon_data_report(
            db,
            date_from=report_date.isoformat(),
            date_to=report_date.isoformat(),
            app_env="prod",
        )
        row = next(item for item in report["items"] if item["trace_id"] == trace)
        assert row["point_success_count"] == 1
        assert row["point_total_count"] == 2
        assert "point_details" not in row
        assert len(coupon_point_details(db, trace_id=trace)) == 2
    finally:
        db.rollback()
        db.close()


def test_coupon_point_details_endpoint() -> None:
    """前端点击使用的接口按约定返回 trace_id 和逐券 items。"""
    db = SessionLocal()
    trace = "point-details-endpoint"
    try:
        admin = admin_repo.get_by_username(db, "point_details_admin")
        if admin is None:
            admin = admin_repo.create_admin(
                db,
                username="point_details_admin",
                password="pass1234",
                role="super_admin",
            )
        token, _expires_at = create_admin_token(admin_id=admin.id, role=admin.role)
        db.add(CouponClaimRecord(
            device_id="point-details-endpoint-device",
            coupon_id="mt-point-details-endpoint",
            coupon_name="接口测试券",
            claim_date=date(2020, 1, 5),
            status="failed",
            reason="接口测试失败",
            trace_id=trace,
        ))
        db.commit()

        response = TestClient(admin_app).get(
            "/admin/api/coupon-data/point-details",
            params={"trace_id": trace},
            headers={"Authorization": f"Bearer {token}"},
        )
        assert response.status_code == 200, response.text
        assert response.json() == {
            "trace_id": trace,
            "items": [{
                "coupon_id": "mt-point-details-endpoint",
                "coupon_name": "接口测试券",
                "status": "failed",
                "reason": "接口测试失败",
            }],
        }
    finally:
        db.rollback()
        db.execute(delete(CouponClaimRecord).where(CouponClaimRecord.trace_id == trace))
        db.commit()
        db.close()