"""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_feedback_list_filter_by_source(admin_client: TestClient, admin_token: str) -> None: """反馈列表按反馈类型(source)筛选;返回项带 source/scene 字段。""" db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800009001", register_channel="sms").id db.add(Feedback(user_id=uid, content="比价反馈", contact="", status="pending", source="comparison", scene="找错商品")) db.add(Feedback(user_id=uid, content="普通反馈", contact="", status="pending", source="profile")) db.commit() finally: db.close() r = admin_client.get( "/admin/api/feedbacks", params={"user_id": uid, "source": "comparison"}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text items = r.json()["items"] assert items and all(it["source"] == "comparison" for it in items) assert any(it["scene"] == "找错商品" for it in items) r = admin_client.get( "/admin/api/feedbacks", params={"user_id": uid, "source": "profile"}, headers=_auth(admin_token), ) assert all(it["source"] == "profile" for it in r.json()["items"]) # 非法 source 值 → 422 assert admin_client.get( "/admin/api/feedbacks", params={"source": "bogus"}, headers=_auth(admin_token) ).status_code == 422 def test_withdraw_list_exposes_source_and_filters( admin_client: TestClient, admin_token: str ) -> None: """提现列表带 source 字段(coin_cash/invite_cash),并可按提现类型筛选。""" db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800009002", register_channel="sms").id db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coin0001", amount_cents=100, status="success", source="coin_cash")) db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invite001", amount_cents=200, status="success", source="invite_cash")) db.commit() finally: db.close() r = admin_client.get("/admin/api/withdraws", params={"user_id": uid}, headers=_auth(admin_token)) assert r.status_code == 200, r.text assert {it["source"] for it in r.json()["items"]} == {"coin_cash", "invite_cash"} r = admin_client.get( "/admin/api/withdraws", params={"user_id": uid, "source": "invite_cash"}, headers=_auth(admin_token), ) items = r.json()["items"] assert items and all(it["source"] == "invite_cash" for it in 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 def test_period_comparison_reward_coin_from_feed_scene( admin_client: TestClient, admin_token: str ) -> None: """比价奖励金币口径:按 ad_feed_reward_record.feed_scene='comparison' 的实发金币 (status=granted)汇总,而非查从不写入的 biz_type 桶(修复大盘该卡恒 0)。 too_short(未发奖)不计。""" from app.models.ad_feed_reward import AdFeedRewardRecord d = "2021-06-15" # 独立历史日,隔离其它用例数据 db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800008801", register_channel="sms").id db.add(AdFeedRewardRecord( client_event_id="cmp-fs-granted", user_id=uid, reward_date=d, ecpm_raw="0", coin=123, feed_scene="comparison", status="granted", )) db.add(AdFeedRewardRecord( client_event_id="cmp-fs-tooshort", user_id=uid, reward_date=d, ecpm_raw="0", coin=99, feed_scene="comparison", status="too_short", )) db.commit() finally: db.close() r = admin_client.get( "/admin/api/stats/overview", params={"date_from": d, "date_to": d}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text assert r.json()["period"]["coins"]["comparison_reward_coin_total"] == 123 def test_period_coupon_reward_coin_from_feed_scene( admin_client: TestClient, admin_token: str ) -> None: """领券奖励金币口径:按 ad_feed_reward_record.feed_scene='coupon' 的实发金币汇总。""" from app.models.ad_feed_reward import AdFeedRewardRecord d = "2021-06-16" db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800008802", register_channel="sms").id db.add(AdFeedRewardRecord( client_event_id="cpn-fs-granted", user_id=uid, reward_date=d, ecpm_raw="0", coin=456, feed_scene="coupon", status="granted", )) db.commit() finally: db.close() r = admin_client.get( "/admin/api/stats/overview", params={"date_from": d, "date_to": d}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text assert r.json()["period"]["coins"]["coupon_reward_coin_total"] == 456 def test_period_coupon_reward_excludes_reward_video( admin_client: TestClient, admin_token: str ) -> None: """领券奖励金币不再把激励视频金币算进来(reward_video/ad_reward 从领券桶拆出); 激励视频仍单独计入 reward_video_coin_total。""" from datetime import datetime from app.models.wallet import CoinTransaction d = "2021-06-17" db = SessionLocal() try: uid = user_repo.upsert_user_for_login(db, phone="13800008803", register_channel="sms").id db.add(CoinTransaction( user_id=uid, amount=50, balance_after=50, biz_type="reward_video", ref_id="rv-split-1", created_at=datetime(2021, 6, 17, 12, 0, 0), )) db.commit() finally: db.close() r = admin_client.get( "/admin/api/stats/overview", params={"date_from": d, "date_to": d}, headers=_auth(admin_token), ) assert r.status_code == 200, r.text coins = r.json()["period"]["coins"] assert coins["coupon_reward_coin_total"] == 0 # 激励视频不计入领券奖励 assert coins["reward_video_coin_total"] == 50 # 仍计入激励视频卡