perf(admin): 按需加载领券点位明细

tests/test_coupon_point_score.py

@@ -1,9 +1,19 @@
-"""admin 领券明细逐场点位分数。"""
-from datetime import date
+"""admin 领券明细逐场点位分数与按需明细。"""
+from datetime import UTC, date, datetime
 
-from app.admin.repositories.coupon_data import _point_stats_by_trace
+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
+from app.models.coupon_state import CouponClaimRecord, CouponSession
 
 
 def test_point_scores_by_trace() -> None:
@@ -25,20 +35,21 @@ def test_point_scores_by_trace() -> None:
         ])
         db.flush()
 
-        stats = _point_stats_by_trace(db, [trace])[trace]
+        stats = _point_scores_by_trace(db, [trace])[trace]
         assert stats["succeeded"] == 2
         assert stats["tried"] == 3
-        assert [item["status"] for item in stats["details"]] == [
+        details = coupon_point_details(db, trace_id=trace)
+        assert [item["status"] for item in details] == [
             "success", "already_claimed", "failed", "skipped"
         ]
-        assert stats["details"][2]["reason"] == "测试失败"
+        assert details[2]["reason"] == "测试失败"
     finally:
         db.rollback()
         db.close()
 
 
-def test_point_stats_keep_skipped_detail_without_faking_score() -> None:
-    """仅有 skipped 时保留明细，但分数仍为 0/0，由 schema 层展示为空。"""
+def test_skipped_detail_does_not_create_a_score() -> None:
+    """仅有 skipped 时按需明细仍可查到，但列表没有虚假的 0/0 分数。"""
     db = SessionLocal()
     trace = "point-score-skipped"
     try:
@@ -51,11 +62,100 @@ def test_point_stats_keep_skipped_detail_without_faking_score() -> None:
         ))
         db.flush()
 
-        stats = _point_stats_by_trace(db, [trace, "missing-trace"])
-        assert stats[trace]["succeeded"] == 0
-        assert stats[trace]["tried"] == 0
-        assert stats[trace]["details"][0]["status"] == "skipped"
-        assert "missing-trace" not in stats
+        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()