lowmaster-chen
f46a38bf36
改了什么：新增京东联盟订单查询客户端、cps_order 京东字段迁移、CPS 对账接口 platform=all/jd、数据大盘京东订单与佣金聚合。 为什么改：数据大盘需要展示京东 CPS 真实订单佣金，不再保持占位。 验证方式：python -m pytest tests\\test_admin_read.py::test_dashboard_overview tests\\test_cps_admin.py::test_jd_reconcile_updates_dashboard -q；本地真实拉取 2026-06-27 京东订单 3 行，页面显示京东有效 1 单/佣金 ¥0.13。
262 lines
9.4 KiB
Python
262 lines
9.4 KiB
Python
"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
|
|
from __future__ import annotations
|
|
|
|
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.db.session import SessionLocal
|
|
from app.models.feedback import Feedback
|
|
from app.models.wallet import CashTransaction, WithdrawOrder
|
|
from app.repositories import user as user_repo
|
|
from app.repositories import wallet as wallet_repo
|
|
|
|
|
|
@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, "m2_admin") is None:
|
|
admin_repo.create_admin(db, username="m2_admin", password="m2-pass", role="super_admin")
|
|
finally:
|
|
db.close()
|
|
c = TestClient(admin_app)
|
|
r = c.post("/admin/api/auth/login", json={"username": "m2_admin", "password": "m2-pass"})
|
|
return r.json()["access_token"]
|
|
|
|
|
|
def _auth(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _seed_user_with_data(phone: str) -> int:
|
|
"""造一个用户 + 金币流水 + 现金流水 + 提现单 + 反馈,返回 user_id。"""
|
|
db = SessionLocal()
|
|
try:
|
|
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms")
|
|
uid = user.id
|
|
wallet_repo.grant_coins(db, uid, 5000, biz_type="signin", remark="测试发金币")
|
|
db.commit()
|
|
db.add(CashTransaction(
|
|
user_id=uid, amount_cents=-100, balance_after_cents=0, biz_type="withdraw", remark="t"
|
|
))
|
|
db.add(WithdrawOrder(
|
|
user_id=uid, out_bill_no=f"test{uid}billno0001", amount_cents=100, status="success"
|
|
))
|
|
db.add(Feedback(user_id=uid, content="测试反馈内容", contact="wx_test", status="pending"))
|
|
db.commit()
|
|
return uid
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
|
|
_seed_user_with_data("13800000001")
|
|
r = admin_client.get("/admin/api/stats/overview", headers=_auth(admin_token))
|
|
assert r.status_code == 200, r.text
|
|
data = r.json()
|
|
assert data["users"]["total"] >= 1
|
|
assert data["coins"]["granted_total"] >= 5000
|
|
assert "success_rate" in data["comparison"]
|
|
assert data["cps"]["available"] is True
|
|
assert "meituan_order_count" in data["cps"]
|
|
assert "jd_order_count" in data["cps"]
|
|
|
|
|
|
def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:
|
|
uid = _seed_user_with_data("13800000002")
|
|
r = admin_client.get("/admin/api/users", headers=_auth(admin_token))
|
|
assert r.status_code == 200, r.text
|
|
assert "items" in r.json()
|
|
|
|
r = admin_client.get(f"/admin/api/users/{uid}", headers=_auth(admin_token))
|
|
assert r.status_code == 200, r.text
|
|
d = r.json()
|
|
assert d["user"]["id"] == uid
|
|
assert d["coin_balance"] == 5000
|
|
assert d["withdraw_total"] >= 1
|
|
assert d["feedback_total"] >= 1
|
|
|
|
assert admin_client.get("/admin/api/users/999999", headers=_auth(admin_token)).status_code == 404
|
|
|
|
|
|
def test_user_filter_by_status(admin_client: TestClient, admin_token: str) -> None:
|
|
_seed_user_with_data("13800000003")
|
|
r = admin_client.get("/admin/api/users", params={"status": "active"}, headers=_auth(admin_token))
|
|
assert r.status_code == 200
|
|
assert all(u["status"] == "active" for u in r.json()["items"])
|
|
|
|
|
|
def test_user_list_sort_filter_range(admin_client: TestClient, admin_token: str) -> None:
|
|
"""用户列表:渠道/昵称筛选 + id 升降序 + 时间范围 + 非法 sort_by 422。"""
|
|
db = SessionLocal()
|
|
try:
|
|
u1 = user_repo.upsert_user_for_login(db, phone="13811110001", register_channel="sms")
|
|
u1.nickname = "排序测试甲"
|
|
u2 = user_repo.upsert_user_for_login(db, phone="13811110002", register_channel="wechat")
|
|
u2.nickname = "排序测试乙"
|
|
db.commit()
|
|
id1, id2 = u1.id, u2.id
|
|
finally:
|
|
db.close()
|
|
|
|
# 渠道精确筛选
|
|
r = admin_client.get(
|
|
"/admin/api/users", params={"register_channel": "wechat"}, headers=_auth(admin_token)
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
assert all(u["register_channel"] == "wechat" for u in r.json()["items"])
|
|
|
|
# 昵称模糊筛选
|
|
r = admin_client.get(
|
|
"/admin/api/users", params={"nickname": "排序测试", "limit": 100}, headers=_auth(admin_token)
|
|
)
|
|
ids = {u["id"] for u in r.json()["items"]}
|
|
assert id1 in ids and id2 in ids
|
|
|
|
# id 升序 / 降序
|
|
asc_ids = [
|
|
u["id"]
|
|
for u in admin_client.get(
|
|
"/admin/api/users",
|
|
params={"sort_by": "id", "sort_order": "asc", "limit": 100},
|
|
headers=_auth(admin_token),
|
|
).json()["items"]
|
|
]
|
|
assert asc_ids == sorted(asc_ids)
|
|
desc_ids = [
|
|
u["id"]
|
|
for u in admin_client.get(
|
|
"/admin/api/users",
|
|
params={"sort_by": "id", "sort_order": "desc", "limit": 100},
|
|
headers=_auth(admin_token),
|
|
).json()["items"]
|
|
]
|
|
assert desc_ids == sorted(desc_ids, reverse=True)
|
|
|
|
# 时间范围:2000 年之前无人;之后有人(验证范围条件确实生效)
|
|
assert (
|
|
admin_client.get(
|
|
"/admin/api/users", params={"created_to": "2000-01-01T00:00:00Z"},
|
|
headers=_auth(admin_token),
|
|
).json()["items"]
|
|
== []
|
|
)
|
|
assert (
|
|
len(
|
|
admin_client.get(
|
|
"/admin/api/users", params={"created_from": "2000-01-01T00:00:00Z", "limit": 100},
|
|
headers=_auth(admin_token),
|
|
).json()["items"]
|
|
)
|
|
>= 1
|
|
)
|
|
|
|
# 非法 sort_by → 422
|
|
assert (
|
|
admin_client.get(
|
|
"/admin/api/users", params={"sort_by": "phone"}, headers=_auth(admin_token)
|
|
).status_code
|
|
== 422
|
|
)
|
|
|
|
|
|
def test_wallet_and_withdraw_lists(admin_client: TestClient, admin_token: str) -> None:
|
|
uid = _seed_user_with_data("13800000004")
|
|
r = admin_client.get(
|
|
"/admin/api/wallet/coin-transactions", params={"user_id": uid}, headers=_auth(admin_token)
|
|
)
|
|
assert r.status_code == 200
|
|
assert len(r.json()["items"]) >= 1
|
|
|
|
r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token))
|
|
assert r.status_code == 200
|
|
assert len(r.json()["items"]) >= 1
|
|
assert all(o["status"] == "success" for o in r.json()["items"])
|
|
|
|
|
|
def test_feedback_list(admin_client: TestClient, admin_token: str) -> None:
|
|
_seed_user_with_data("13800000005")
|
|
r = admin_client.get(
|
|
"/admin/api/feedbacks", params={"status": "pending"}, headers=_auth(admin_token)
|
|
)
|
|
assert r.status_code == 200
|
|
assert all(f["status"] == "pending" for f in r.json()["items"])
|
|
|
|
|
|
def test_ad_coin_audit_full_count_truncate_and_only_mismatch(
|
|
admin_client: TestClient, admin_token: str
|
|
) -> None:
|
|
"""A+B:total/mismatch_count 按全量统计(不受 limit 影响),truncated 旗标 + only_mismatch 过滤。
|
|
|
|
用 capped 行造确定性数据(应发恒 0,coin==0 即一致),不依赖发奖公式:
|
|
3 条 coin=0(一致) + 2 条 coin=7(不一致) → 全量 total=5、mismatch=2。
|
|
"""
|
|
from app.models.ad_reward import AdRewardRecord
|
|
|
|
d = "2020-01-15" # 固定历史日 + 独立 user,隔离其它用例数据
|
|
db = SessionLocal()
|
|
try:
|
|
uid = user_repo.upsert_user_for_login(db, phone="13800009999", register_channel="sms").id
|
|
for i in range(3):
|
|
db.add(AdRewardRecord(
|
|
trans_id=f"adaudit-ok-{i}", user_id=uid, coin=0, status="capped",
|
|
reward_scene="reward_video", reward_date=d,
|
|
))
|
|
for i in range(2):
|
|
db.add(AdRewardRecord(
|
|
trans_id=f"adaudit-bad-{i}", user_id=uid, coin=7, status="capped",
|
|
reward_scene="reward_video", reward_date=d,
|
|
))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
base = {"date": d, "user_id": uid}
|
|
|
|
# 全量:total=5、mismatch=2、不截断、返回 5 条
|
|
body = admin_client.get(
|
|
"/admin/api/ad-coin-audit", params={**base, "limit": 100}, headers=_auth(admin_token)
|
|
).json()
|
|
assert body["total"] == 5
|
|
assert body["mismatch_count"] == 2
|
|
assert body["truncated"] is False
|
|
assert len(body["items"]) == 5
|
|
|
|
# 截断:limit=2 → 统计仍全量、truncated=True、只回 2 条
|
|
body = admin_client.get(
|
|
"/admin/api/ad-coin-audit", params={**base, "limit": 2}, headers=_auth(admin_token)
|
|
).json()
|
|
assert body["total"] == 5 and body["mismatch_count"] == 2
|
|
assert body["truncated"] is True
|
|
assert len(body["items"]) == 2
|
|
|
|
# only_mismatch:只回 ✗ 行(2 条),统计仍全量、不截断
|
|
body = admin_client.get(
|
|
"/admin/api/ad-coin-audit",
|
|
params={**base, "limit": 100, "only_mismatch": True},
|
|
headers=_auth(admin_token),
|
|
).json()
|
|
assert body["total"] == 5 and body["mismatch_count"] == 2
|
|
assert body["truncated"] is False
|
|
assert len(body["items"]) == 2
|
|
assert all(it["matched"] is False for it in body["items"])
|
|
|
|
|
|
def test_read_apis_require_auth(admin_client: TestClient) -> None:
|
|
"""所有 M2 读接口未带 token → 401(router 级 get_current_admin 守卫)。"""
|
|
for path in [
|
|
"/admin/api/stats/overview",
|
|
"/admin/api/users",
|
|
"/admin/api/wallet/coin-transactions",
|
|
"/admin/api/withdraws",
|
|
"/admin/api/feedbacks",
|
|
]:
|
|
assert admin_client.get(path).status_code == 401, path
|