shaguabijia-app-server/tests/test_admin_read.py

"""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="new"))
        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 False


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_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": "new"}, headers=_auth(admin_token))
    assert r.status_code == 200
    assert all(f["status"] == "new" for f in r.json()["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