"""admin CPS 端点测试:每日明细 date 格式 + 按天按用户领券下钻(/day-users)。""" from __future__ import annotations from datetime import date, datetime, timedelta, timezone import pytest 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.repositories import cps as cps_repo from app.db.session import SessionLocal from app.models.cps_link import CpsClick from app.models.cps_order import CpsOrder from app.models.cps_wx_user import CpsWxUser from app.repositories import cps_link as cps_link_repo _BJ = timezone(timedelta(hours=8)) @pytest.fixture() def admin_client() -> TestClient: return TestClient(admin_app) @pytest.fixture() def admin_token() -> str: db = SessionLocal() try: if admin_repo.get_by_username(db, "cps_test_admin") is None: admin_repo.create_admin( db, username="cps_test_admin", password="cps-pass", role="super_admin" ) finally: db.close() c = TestClient(admin_app) r = c.post("/admin/api/auth/login", json={"username": "cps_test_admin", "password": "cps-pass"}) return r.json()["access_token"] def _auth(token: str) -> dict: return {"Authorization": f"Bearer {token}"} def _seed_cps_day(day_bj: date) -> tuple[int, str]: """造淘宝群 + 2 个活动(券) + 2 条 link + day_bj 当天点击(含匿名 + 次日各 1) + 授权画像。 o_user(本群唯一):券A visit×3、券B visit×2、copy×2 → visit_count=5、copy_count=2。 另造 1 条匿名 visit(openid=None)与 1 条次日 visit,均不应计入当天该用户聚合。 clicked_at 统一存 tz-aware UTC(SQLite DateTime 按字段渲染、忽略 tzinfo; 转 UTC 后字段即 UTC 墙钟,与 repo 侧 _as_utc 比较口径一致)。返回 (group_id, openid)。 """ db = SessionLocal() try: gid = cps_repo.create_group(db, name="下钻测试群", platforms=["taobao"]).id openid = f"od_user_{gid}" # 按群唯一,规避 CpsWxUser.openid 唯一约束跨用例冲突 a1 = cps_repo.create_activity(db, name="618神券", platform="taobao", payload="tkl-1") a2 = cps_repo.create_activity(db, name="买一送一", platform="taobao", payload="tkl-2") link1 = cps_link_repo.create_link( db, group_id=gid, activity_id=a1.id, sid=None, target_url="t1", platform="taobao" ) link2 = cps_link_repo.create_link( db, group_id=gid, activity_id=a2.id, sid=None, target_url="t2", platform="taobao" ) db.add(CpsWxUser(openid=openid, nickname="张三", headimgurl="https://h/1")) db.commit() def at(hour: int) -> datetime: return datetime(day_bj.year, day_bj.month, day_bj.day, hour, tzinfo=_BJ).astimezone( timezone.utc ) for _ in range(3): db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=at(10))) for _ in range(2): db.add(CpsClick(link_id=link2.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=at(11))) for _ in range(2): db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="copy", openid=openid, clicked_at=at(12))) # 匿名点击 — 不计入 db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=None, clicked_at=at(13))) # 次日点击 — 验证时间窗,不计入当天 nxt = (datetime(day_bj.year, day_bj.month, day_bj.day, 10, tzinfo=_BJ) + timedelta(days=1)).astimezone(timezone.utc) db.add(CpsClick(link_id=link1.id, group_id=gid, sid=None, event_type="visit", openid=openid, clicked_at=nxt)) db.commit() return gid, openid finally: db.close() def test_daily_date_is_full_iso(admin_client: TestClient, admin_token: str) -> None: """/daily 每行 date 改为 YYYY-MM-DD(10 字符、两个连字符),不再是 MM-DD。""" gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/daily", params={"days": 3}, headers=_auth(admin_token) ) assert r.status_code == 200, r.text rows = r.json()["rows"] assert rows, "应有按天补零行" for row in rows: assert len(row["date"]) == 10 and row["date"].count("-") == 2, row["date"] def test_day_users_aggregates(admin_client: TestClient, admin_token: str) -> None: """当天该群:仅授权用户;领券=copy、点击=visit;coupons=visit 券×次数倒序、合计=点击次数。""" gid, openid = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026-06-25"}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text body = r.json() assert body["group_id"] == gid assert body["date"] == "2026-06-25" users = body["users"] assert len(users) == 1 # 匿名不计、次日不计 u = users[0] assert u["openid"] == openid assert u["nickname"] == "张三" assert u["headimgurl"] == "https://h/1" assert u["copy_count"] == 2 assert u["visit_count"] == 5 assert [c["name"] for c in u["coupons"]] == ["618神券", "买一送一"] assert [c["count"] for c in u["coupons"]] == [3, 2] assert sum(c["count"] for c in u["coupons"]) == u["visit_count"] def test_day_users_group_not_found(admin_client: TestClient, admin_token: str) -> None: r = admin_client.get( "/admin/api/cps/groups/999999/day-users", params={"date": "2026-06-25"}, headers=_auth(admin_token), ) assert r.status_code == 404 def test_day_users_bad_date(admin_client: TestClient, admin_token: str) -> None: gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026/06/25"}, # 非 YYYY-MM-DD headers=_auth(admin_token), ) assert r.status_code == 400 def test_day_users_requires_auth(admin_client: TestClient) -> None: r = admin_client.get( "/admin/api/cps/groups/1/day-users", params={"date": "2026-06-25"} ) assert r.status_code == 401 def test_day_users_empty_when_no_clicks(admin_client: TestClient, admin_token: str) -> None: gid, _ = _seed_cps_day(date(2026, 6, 25)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2026-06-20"}, # 该群当天无任何点击 headers=_auth(admin_token), ) assert r.status_code == 200 assert r.json()["users"] == [] def test_day_users_cross_year(admin_client: TestClient, admin_token: str) -> None: """跨年:YYYY-MM-DD 才能精确定位 12-31(MM-DD 会丢年份);次日(次年 01-01)不计入。""" gid, openid = _seed_cps_day(date(2025, 12, 31)) r = admin_client.get( f"/admin/api/cps/groups/{gid}/day-users", params={"date": "2025-12-31"}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text users = r.json()["users"] assert len(users) == 1 assert users[0]["openid"] == openid assert users[0]["visit_count"] == 5 # 次年 01-01 那条被时间窗排除 def test_jd_reconcile_updates_dashboard( admin_client: TestClient, admin_token: str, monkeypatch: pytest.MonkeyPatch ) -> None: """京东拉单:按订单行入库;大盘只统计有效 validCode 的佣金。""" def fake_query_order_rows(**kwargs): start = kwargs["start_time"].astimezone(_BJ) if start.hour != 10 or kwargs["page_index"] != 1: return {"rows": [], "has_more": False} return { "rows": [ { "id": "pytest-jd-row-valid", "orderId": "pytest-jd-order-1", "skuId": "sku-1", "skuName": "京东测试商品", "orderTime": "2026-06-25 10:10:00", "modifyTime": "2026-06-25 10:20:00", "validCode": "16", "estimateCosPrice": "19.90", "estimateFee": "1.23", "actualFee": "2.34", "commissionRate": "10.00", }, { "id": "pytest-jd-row-invalid", "orderId": "pytest-jd-order-2", "skuId": "sku-2", "skuName": "京东无效订单", "orderTime": "2026-06-25 10:15:00", "modifyTime": "2026-06-25 10:25:00", "validCode": "4", "estimateCosPrice": "9.90", "estimateFee": "0.50", "commissionRate": "5.00", }, ], "has_more": False, } monkeypatch.setattr("app.integrations.jd_union.query_order_rows", fake_query_order_rows) r = admin_client.post( "/admin/api/cps/orders/reconcile", params={ "platform": "jd", "date_from": "2026-06-25", "date_to": "2026-06-25", "query_time_type": 1, }, headers=_auth(admin_token), ) assert r.status_code == 200, r.text assert r.json()["fetched"] == 2 with SessionLocal() as db: rows = db.query(CpsOrder).filter(CpsOrder.platform == "jd").all() assert len([o for o in rows if o.order_id.startswith("jd:pytest-jd-row")]) == 2 overview = admin_client.get( "/admin/api/stats/overview", params={"date_from": "2026-06-25", "date_to": "2026-06-25"}, headers=_auth(admin_token), ) assert overview.status_code == 200, overview.text cps = overview.json()["cps"] assert cps["jd_order_count"] >= 1 assert cps["jd_commission_cents"] >= 234 assert cps["jd_invalid_count"] >= 1