guke
2eb44fe947
- 新增 GET /admin/api/cps/groups/{id}/day-users?date=YYYY-MM-DD:
按 openid 聚合当天领券(copy)/点击(visit) + 每人 visit 过的券(券×次数)
---------
Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #79
183 lines
7.3 KiB
Python
183 lines
7.3 KiB
Python
"""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_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 那条被时间窗排除
|