shaguabijia-app-server/tests/test_admin_read.py

"""Admin M2 读接口测试:大盘聚合 + 用户/流水/提现/反馈列表 + 鉴权拦截。"""
from __future__ import annotations

from datetime import 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.db.session import SessionLocal, engine
from app.models.comparison import ComparisonRecord
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_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")

    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_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        # 仍计入激励视频卡


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