"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。""" from __future__ import annotations from datetime import UTC, datetime import pytest from fastapi.testclient import TestClient from sqlalchemy import event from app.admin.main import admin_app from app.admin.repositories import admin_user as admin_repo from app.admin.repositories import queries from app.db.session import SessionLocal, engine from app.models.comparison import ComparisonRecord from app.models.feedback import Feedback from app.models.invite import InviteRelation from app.models.savings import SavingsRecord from app.models.user import User 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_dashboard_period_comparison_is_aggregated_by_backend( admin_client: TestClient, admin_token: str ) -> None: created_at = datetime(2037, 1, 15, 12) rows = [ ("dashboard-aggregate-success", "success", 101, 0.1), ("dashboard-aggregate-failed", "failed", 200, 0.2), ("dashboard-aggregate-cancelled", "cancelled", 300, 0.3), ("dashboard-aggregate-running", "running", 400, 0.4), ] db = SessionLocal() try: for trace_id, status, total_ms, llm_cost_yuan in rows: db.add( ComparisonRecord( trace_id=trace_id, status=status, total_ms=total_ms, llm_cost_yuan=llm_cost_yuan, created_at=created_at, ) ) db.commit() finally: db.close() response = admin_client.get( "/admin/api/stats/overview", params={"date_from": "2037-01-15", "date_to": "2037-01-15"}, headers=_auth(admin_token), ) assert response.status_code == 200, response.text comparison = response.json()["period"]["comparison"] assert comparison["total"] == 4 assert comparison["completed"] == 2 assert comparison["cancelled"] == 1 assert comparison["success"] == 1 assert comparison["success_rate"] == 0.3333 assert comparison["median_duration_ms"] == 151 assert comparison["p95_duration_ms"] == 195 assert comparison["token_cost_total_yuan"] == pytest.approx(1.0) 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_reward_detail_does_not_select_unrelated_new_ad_columns( admin_client: TestClient, admin_token: str ) -> None: """旧库缺少无关新列时，提现详情的统计和金币记录仍应可读。""" uid = _seed_user_with_data("13800000022") def reject_full_ad_reward_projection( _conn, _cursor, statement: str, _parameters, _context, _executemany ) -> None: if "ad_reward_record.boost_round_id" in statement: raise AssertionError("提现详情不应查询未使用的 boost_round_id") if "FROM user" in statement and "user.phone" in statement: raise AssertionError("奖励统计的用户存在性检查不应展开完整 user 表") event.listen(engine, "before_cursor_execute", reject_full_ad_reward_projection) try: stats = admin_client.get( f"/admin/api/users/{uid}/reward-stats", headers=_auth(admin_token) ) records = admin_client.get( f"/admin/api/users/{uid}/coin-records", params={"limit": 10, "cursor": 0}, headers=_auth(admin_token), ) finally: event.remove(engine, "before_cursor_execute", reject_full_ad_reward_projection) assert stats.status_code == 200, stats.text assert records.status_code == 200, records.text def test_user_reward_stats_can_scope_withdrawals_by_account( admin_client: TestClient, admin_token: str ) -> None: uid = _seed_user_with_data("13800000023") db = SessionLocal() try: account = wallet_repo.get_or_create_account(db, uid) account.cash_balance_cents = 123 account.invite_cash_balance_cents = 456 db.add_all( [ WithdrawOrder( user_id=uid, out_bill_no="rewardstatsinvite00000001", amount_cents=250, source="invite_cash", status="success", ), WithdrawOrder( user_id=uid, out_bill_no="rewardstatsinvite00000002", amount_cents=50, source="invite_cash", status="reviewing", ), ] ) db.commit() finally: db.close() coin = admin_client.get( f"/admin/api/users/{uid}/reward-stats", params={"withdraw_source": "coin_cash"}, headers=_auth(admin_token), ) invite = admin_client.get( f"/admin/api/users/{uid}/reward-stats", params={"withdraw_source": "invite_cash"}, headers=_auth(admin_token), ) assert coin.status_code == 200, coin.text assert invite.status_code == 200, invite.text assert coin.json()["withdraw_success_cents"] == 100 assert coin.json()["withdraw_total"] == 1 assert coin.json()["cash_balance_cents"] == 123 assert invite.json()["withdraw_success_cents"] == 250 assert invite.json()["withdraw_total"] == 2 assert invite.json()["cash_balance_cents"] == 456 def test_user_reward_stats_draw_ecpm_uses_all_filtered_impressions( admin_client: TestClient, admin_token: str ) -> None: """Draw 平均 eCPM 应与收益报表一致，不能只平均成功发奖记录。""" from app.models.ad_ecpm import AdEcpmRecord from app.models.ad_feed_reward import AdFeedRewardRecord uid = _seed_user_with_data("13800000024") created_at = datetime(2038, 1, 15, 4, tzinfo=UTC) db = SessionLocal() try: db.add_all( [ AdFeedRewardRecord( client_event_id="reward-stats-granted-high", ad_session_id="reward-stats-granted-high", user_id=uid, reward_date="2038-01-15", duration_seconds=10, unit_count=1, ecpm_raw="9000", ad_type="draw", feed_scene="coupon", app_env="prod", our_code_id="104098712", coin=9, status="granted", created_at=created_at, ), AdEcpmRecord( user_id=uid, ad_type="draw", feed_scene="coupon", ad_session_id="reward-stats-impression-low", app_env="prod", our_code_id="104098712", ecpm_raw="1000", report_date="2038-01-15", created_at=created_at, ), AdEcpmRecord( user_id=uid, ad_type="feed", feed_scene="coupon", ad_session_id="reward-stats-impression-mid", app_env="prod", our_code_id="104098712", ecpm_raw="3000", report_date="2038-01-15", created_at=created_at, ), # 同用户但不同场景/环境/非业务代码位，均不应进入本次详情筛选。 AdEcpmRecord( user_id=uid, ad_type="draw", feed_scene="comparison", ad_session_id="reward-stats-other-scene", app_env="prod", our_code_id="104098712", ecpm_raw="7000", report_date="2038-01-15", created_at=created_at, ), AdEcpmRecord( user_id=uid, ad_type="draw", feed_scene="coupon", ad_session_id="reward-stats-test-env", app_env="test", our_code_id="104127529", ecpm_raw="8000", report_date="2038-01-15", created_at=created_at, ), AdEcpmRecord( user_id=uid, ad_type="draw", feed_scene="coupon", ad_session_id="reward-stats-non-business", app_env="prod", our_code_id="demo-slot", ecpm_raw="9000", report_date="2038-01-15", created_at=created_at, ), ] ) db.commit() finally: db.close() response = admin_client.get( f"/admin/api/users/{uid}/reward-stats", params={ "date_from": "2038-01-15T00:00:00Z", "date_to": "2038-01-15T23:59:59Z", "app_env": "prod", "revenue_scope": "business", "feed_scene": "coupon", }, headers=_auth(admin_token), ) assert response.status_code == 200, response.text data = response.json() assert data["feed_count"] == 1 # 全部真实展示 (1000 + 3000) / 2；不能返回成功发奖记录的 9000。 assert data["feed_avg_ecpm"] == 2000.0 def test_user_coin_record_sort_accepts_mixed_timezone_datetimes() -> None: """线上 PostgreSQL 返回 aware，SQLite/历史转换可能返回 naive，二者必须可混排。""" naive = datetime(2038, 1, 1, 8, 0) aware = datetime(2038, 1, 1, 7, 0, tzinfo=UTC) rows = [{"created_at": aware}, {"created_at": naive}] rows.sort(key=queries._coin_record_sort_key, reverse=True) assert rows == [{"created_at": naive}, {"created_at": aware}] 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 # 直接从 ORM 查询，验证列表富化返回人工风险结论。 user = db.get(User, uid) assert user is not None user.is_high_risk = True user.high_risk_note = "邀请行为异常" 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.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}coinreview", amount_cents=50, status="reviewing", source="coin_cash")) db.add(WithdrawOrder(user_id=uid, out_bill_no=f"src{uid}invitereview", amount_cents=80, status="reviewing", 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) assert all(it["is_high_risk"] is True for it in items) assert all(it["high_risk_note"] == "邀请行为异常" for it in items) # 累计提现按当前审核页账户隔离，不把普通现金的 100 分串进邀请金。 assert all(it["cumulative_success_cents"] == 200 for it in items) invite_summary = admin_client.get( "/admin/api/withdraws/summary", params={"source": "invite_cash"}, headers=_auth(admin_token), ).json() other_summary = admin_client.get( "/admin/api/withdraws/summary", params={"source": "coin_cash"}, headers=_auth(admin_token), ).json() assert invite_summary["reviewing_count"] >= 1 assert other_summary["reviewing_count"] >= 1 assert invite_summary["reviewing_amount_cents"] >= 80 assert other_summary["reviewing_amount_cents"] >= 50 def test_invite_withdraw_detail_contains_invitee_first_actions( admin_client: TestClient, admin_token: str ) -> None: db = SessionLocal() try: inviter = user_repo.upsert_user_for_login( db, phone="13800009031", register_channel="sms" ) invitee = user_repo.upsert_user_for_login( db, phone="13800009032", register_channel="sms" ) db.add( InviteRelation( inviter_user_id=inviter.id, invitee_user_id=invitee.id, status="effective", compare_reward_granted=True, compare_reward_cents=50, ) ) db.add( ComparisonRecord( user_id=invitee.id, trace_id="invite-detail-first-compare", status="success", store_name="首次比价商家", product_names="商品甲、商品乙", ) ) db.add( SavingsRecord( user_id=invitee.id, order_amount_cents=1888, saved_amount_cents=300, source="compare", shop_name="首次下单商家", title="首次下单商品", dishes=[], ) ) bill = "inviteoverviewbill0001" db.add( WithdrawOrder( user_id=inviter.id, out_bill_no=bill, amount_cents=100, status="reviewing", source="invite_cash", ) ) db.commit() finally: db.close() response = admin_client.get( f"/admin/api/withdraws/{bill}", headers=_auth(admin_token) ) assert response.status_code == 200, response.text overview = response.json()["invite_overview"] assert overview["invite_total"] == 1 assert overview["invite_success_total"] == 1 item = overview["items"][0] assert item["phone"] == "13800009032" assert item["first_compare_store"] == "首次比价商家" assert item["first_compare_products"] == "商品甲、商品乙" assert item["first_order_store"] == "首次下单商家" assert item["first_order_products"] == "首次下单商品" assert item["first_order_amount_cents"] == 1888 filtered = admin_client.get( f"/admin/api/withdraws/{bill}", params={ "date_from": "2099-01-01T00:00:00Z", "date_to": "2099-01-02T00:00:00Z", }, headers=_auth(admin_token), ) assert filtered.status_code == 200, filtered.text assert filtered.json()["invite_overview"]["invite_total"] == 0 def test_withdraw_summary_counts_match_status_lists_by_source( admin_client: TestClient, admin_token: str ) -> None: """每个审核页签的汇总数必须等于相同 source、status 的列表总数。""" db = SessionLocal() try: uid = user_repo.upsert_user_for_login( db, phone="13800009003", register_channel="sms" ).id for index, status in enumerate( ("reviewing", "pending", "success", "rejected", "failed"), start=1 ): db.add( WithdrawOrder( user_id=uid, out_bill_no=f"summary{uid}invite{index}", amount_cents=index * 10, status=status, source="invite_cash", ) ) db.add( WithdrawOrder( user_id=uid, out_bill_no=f"summary{uid}coinfailed", amount_cents=99, status="failed", source="coin_cash", ) ) db.commit() finally: db.close() status_fields = { "reviewing": "reviewing_count", "pending": "pending_count", "success": "success_count", "rejected": "rejected_count", "failed": "failed_count", } for source in ("coin_cash", "invite_cash"): summary_response = admin_client.get( "/admin/api/withdraws/summary", params={"source": source}, headers=_auth(admin_token), ) assert summary_response.status_code == 200, summary_response.text summary = summary_response.json() for status, field in status_fields.items(): list_response = admin_client.get( "/admin/api/withdraws", params={"source": source, "status": status, "limit": 1}, headers=_auth(admin_token), ) assert list_response.status_code == 200, list_response.text assert summary[field] == list_response.json()["total"] reviewing_response = admin_client.get( "/admin/api/withdraws", params={"source": source, "status": "reviewing", "limit": 100}, headers=_auth(admin_token), ) assert reviewing_response.status_code == 200, reviewing_response.text reviewing_body = reviewing_response.json() assert reviewing_body["total"] == len(reviewing_body["items"]) assert summary["reviewing_amount_cents"] == sum( item["amount_cents"] for item in reviewing_body["items"] ) all_response = admin_client.get( "/admin/api/withdraws", params={"source": source, "limit": 1}, headers=_auth(admin_token), ) assert all_response.status_code == 200, all_response.text assert summary["total_count"] == all_response.json()["total"] assert summary["total_count"] == sum( summary[field] for field in status_fields.values() ) assert admin_client.get( "/admin/api/withdraws/summary", params={"source": "unknown"}, headers=_auth(admin_token), ).status_code == 422 def test_comparison_records_show_readable_device_and_rom_version( admin_client: TestClient, admin_token: str ) -> None: db = SessionLocal() try: user = user_repo.upsert_user_for_login( db, phone="13800009041", register_channel="sms" ) record = ComparisonRecord( user_id=user.id, trace_id="comparison-readable-device", status="success", device_model="V2166BA", device_manufacturer="vivo", rom_vendor="vivo", rom_name="OriginOS", rom_version=4, android_version="14", ) db.add(record) db.commit() record_id = record.id user_id = user.id finally: db.close() response = admin_client.get( "/admin/api/comparison-records", params={"user_id": user_id}, headers=_auth(admin_token), ) assert response.status_code == 200, response.text item = response.json()["items"][0] assert item["device_model"] == "V2166BA" assert item["device_model_name"] == "vivo Y77e" assert item["rom_name"] == "OriginOS" assert item["rom_version"] == 4 detail = admin_client.get( f"/admin/api/comparison-records/{record_id}", headers=_auth(admin_token), ) assert detail.status_code == 200, detail.text assert detail.json()["device_model_name"] == "vivo Y77e" assert detail.json()["rom_version"] == 4 def test_comparison_records_show_real_order_status( admin_client: TestClient, admin_token: str ) -> None: db = SessionLocal() try: user = user_repo.upsert_user_for_login( db, phone="13800009042", register_channel="sms" ) ordered_record = ComparisonRecord( user_id=user.id, trace_id="comparison-ordered-shop", status="success", store_name="真实下单店", ) unordered_record = ComparisonRecord( user_id=user.id, trace_id="comparison-demo-order-shop", status="success", store_name="演示下单店", ) db.add_all([ordered_record, unordered_record]) db.flush() db.add_all( [ SavingsRecord( user_id=user.id, order_amount_cents=1800, saved_amount_cents=300, shop_name="真实下单店", source="compare", client_event_id="admin-comparison-real-order", ), SavingsRecord( user_id=user.id, order_amount_cents=1500, saved_amount_cents=200, shop_name="演示下单店", source="demo", ), ] ) db.commit() ordered_record_id = ordered_record.id user_id = user.id finally: db.close() response = admin_client.get( "/admin/api/comparison-records", params={"user_id": user_id}, headers=_auth(admin_token), ) assert response.status_code == 200, response.text by_trace = {item["trace_id"]: item for item in response.json()["items"]} assert by_trace["comparison-ordered-shop"]["ordered"] is True assert by_trace["comparison-demo-order-shop"]["ordered"] is False detail = admin_client.get( f"/admin/api/comparison-records/{ordered_record_id}", headers=_auth(admin_token), ) assert detail.status_code == 200, detail.text assert detail.json()["ordered"] is True 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 # 仍计入激励视频卡 def test_period_regular_task_coin_uses_explicit_allowlist( admin_client: TestClient, admin_token: str ) -> None: """常规任务金币只加明确任务来源，领券/比价及未知新类型不得自动混入。""" from datetime import datetime from app.models.wallet import CoinTransaction d = "2021-06-18" included = { "signin": 100, "task_enable_notification": 300, "task_other": 400, "price_report_reward": 500, "feedback_reward": 600, } excluded = { "signin_boost": 200, "feed_ad_reward_coupon": 700, "feed_ad_reward_comparison": 800, "feed_ad_reward": 900, "reward_video": 1000, "admin_grant": 1100, "invite_inviter": 1200, "future_unknown_reward": 1300, } db = SessionLocal() try: uid = user_repo.upsert_user_for_login( db, phone="13800008804", register_channel="sms" ).id balance = 0 rows = [] for index, (biz_type, amount) in enumerate((included | excluded).items(), start=1): balance += amount rows.append(CoinTransaction( user_id=uid, amount=amount, balance_after=balance, biz_type=biz_type, ref_id=f"regular-task-{index}", created_at=datetime(2021, 6, 18, 12, 0, index), )) db.add_all(rows) db.commit() finally: db.close() response = admin_client.get( "/admin/api/stats/overview", params={"date_from": d, "date_to": d}, headers=_auth(admin_token), ) assert response.status_code == 200, response.text coins = response.json()["period"]["coins"] assert coins["regular_task_coin_total"] == sum(included.values()) assert coins["task_coin_total"] == 700 assert coins["reward_video_coin_total"] == 1200 def test_period_signin_boost_moves_to_reward_video_without_double_count( admin_client: TestClient, admin_token: str ) -> None: """历史签到膨胀归看视频桶，不再进常规任务；本期发放总额不变且不重复计算。""" from datetime import datetime from app.models.wallet import CoinTransaction d = "2021-06-19" rows = [ ("signin_boost", 200), ("reward_video", 100), ("signin", 50), ] db = SessionLocal() try: uid = user_repo.upsert_user_for_login( db, phone="13800008805", register_channel="sms" ).id balance = 0 for index, (biz_type, amount) in enumerate(rows, start=1): balance += amount db.add(CoinTransaction( user_id=uid, amount=amount, balance_after=balance, biz_type=biz_type, ref_id=f"signin-boost-route-{index}", created_at=datetime(2021, 6, 19, 12, 0, index), )) db.commit() finally: db.close() response = admin_client.get( "/admin/api/stats/overview", params={"date_from": d, "date_to": d}, headers=_auth(admin_token), ) assert response.status_code == 200, response.text coins = response.json()["period"]["coins"] assert coins["granted_total"] == 350 assert coins["reward_video_coin_total"] == 300 assert coins["regular_task_coin_total"] == 50 assert coins["signin_boost_coin_total"] == 200