shaguabijia-app-server/tests/test_admin_write.py

"""Admin M3 写接口测试:调金币/封号/反馈/提现/admin账号。

验证:写操作落审计 + 金币写流水 + 扣负拒绝 + 角色守卫 + 提现复用 wallet(mock wxpay)。
"""
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.core.security import hash_password
from app.db.session import SessionLocal
from app.models.admin import AdminAuditLog
from app.models.feedback import Feedback
from app.models.user import User
from app.models.wallet import CoinAccount, CoinTransaction, WithdrawOrder
from app.repositories import user as user_repo


@pytest.fixture()
def admin_client() -> TestClient:
    return TestClient(admin_app)


def _token(username: str, role: str) -> str:
    db = SessionLocal()
    try:
        a = admin_repo.get_by_username(db, username)
        if a is None:
            admin_repo.create_admin(db, username=username, password="pass1234", role=role)
        else:
            a.password_hash = hash_password("pass1234")
            a.role = role
            a.status = "active"
            db.commit()
    finally:
        db.close()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": username, "password": "pass1234"}
    ).json()["access_token"]


@pytest.fixture()
def super_token() -> str:
    return _token("w_super", "super_admin")


@pytest.fixture()
def finance_token() -> str:
    return _token("w_finance", "finance")


@pytest.fixture()
def operator_token() -> str:
    return _token("w_operator", "operator")


def _auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


def _seed_user(phone: str) -> int:
    db = SessionLocal()
    try:
        return user_repo.upsert_user_for_login(db, phone=phone, register_channel="sms").id
    finally:
        db.close()


def _seed_feedback(phone: str) -> int:
    uid = _seed_user(phone)
    db = SessionLocal()
    try:
        fb = Feedback(user_id=uid, content="测试", contact="wx", status="new")
        db.add(fb)
        db.commit()
        return fb.id
    finally:
        db.close()


# ===== 调金币 =====

def test_grant_coins_writes_txn_and_audit(admin_client: TestClient, finance_token: str) -> None:
    uid = _seed_user("13900000001")
    r = admin_client.post(
        f"/admin/api/users/{uid}/coins", json={"amount": 1000, "reason": "补偿"},
        headers=_auth(finance_token),
    )
    assert r.status_code == 200, r.text
    db = SessionLocal()
    try:
        assert db.get(CoinAccount, uid).coin_balance == 1000
        txns = db.execute(
            select(CoinTransaction).where(
                CoinTransaction.user_id == uid, CoinTransaction.biz_type == "admin_grant"
            )
        ).scalars().all()
        assert len(txns) == 1 and txns[0].amount == 1000
        logs = db.execute(
            select(AdminAuditLog).where(
                AdminAuditLog.action == "user.coins.grant", AdminAuditLog.target_id == str(uid)
            )
        ).scalars().all()
        assert len(logs) == 1 and logs[0].detail["amount"] == 1000
    finally:
        db.close()


def test_deduct_below_zero_rejected(admin_client: TestClient, finance_token: str) -> None:
    uid = _seed_user("13900000002")
    r = admin_client.post(
        f"/admin/api/users/{uid}/coins", json={"amount": -999999, "reason": "扣"},
        headers=_auth(finance_token),
    )
    assert r.status_code == 400
    db = SessionLocal()
    try:
        txns = db.execute(
            select(CoinTransaction).where(CoinTransaction.user_id == uid)
        ).scalars().all()
        assert len(txns) == 0  # 拒绝后无流水(原子:都不发生)
    finally:
        db.close()


def test_grant_zero_rejected(admin_client: TestClient, finance_token: str) -> None:
    uid = _seed_user("13900000008")
    r = admin_client.post(
        f"/admin/api/users/{uid}/coins", json={"amount": 0, "reason": "x"},
        headers=_auth(finance_token),
    )
    assert r.status_code == 400


# ===== 封号 =====

def test_set_user_status_and_audit(admin_client: TestClient, operator_token: str) -> None:
    uid = _seed_user("13900000003")
    r = admin_client.post(
        f"/admin/api/users/{uid}/status", json={"status": "disabled"}, headers=_auth(operator_token)
    )
    assert r.status_code == 200, r.text
    db = SessionLocal()
    try:
        assert db.get(User, uid).status == "disabled"
        logs = db.execute(
            select(AdminAuditLog).where(
                AdminAuditLog.action == "user.status.set", AdminAuditLog.target_id == str(uid)
            )
        ).scalars().all()
        assert logs[0].detail == {"before": "active", "after": "disabled"}
    finally:
        db.close()
    assert admin_client.post(
        f"/admin/api/users/{uid}/status", json={"status": "active"}, headers=_auth(operator_token)
    ).status_code == 200


# ===== 一键开启新手引导 =====

def test_set_force_onboarding_and_audit(admin_client: TestClient, operator_token: str) -> None:
    uid = _seed_user("13900000009")
    r = admin_client.post(
        f"/admin/api/users/{uid}/force-onboarding", json={"enabled": True},
        headers=_auth(operator_token),
    )
    assert r.status_code == 200, r.text
    db = SessionLocal()
    try:
        assert db.get(User, uid).force_onboarding is True
        logs = db.execute(
            select(AdminAuditLog).where(
                AdminAuditLog.action == "user.force_onboarding.set",
                AdminAuditLog.target_id == str(uid),
            )
        ).scalars().all()
        assert logs[0].detail == {"before": False, "after": True}
    finally:
        db.close()
    # 取消(回到 false)
    assert admin_client.post(
        f"/admin/api/users/{uid}/force-onboarding", json={"enabled": False},
        headers=_auth(operator_token),
    ).status_code == 200
    db = SessionLocal()
    try:
        assert db.get(User, uid).force_onboarding is False
    finally:
        db.close()


def test_finance_cannot_force_onboarding(admin_client: TestClient, finance_token: str) -> None:
    uid = _seed_user("13900000010")
    r = admin_client.post(
        f"/admin/api/users/{uid}/force-onboarding", json={"enabled": True},
        headers=_auth(finance_token),
    )
    assert r.status_code == 403


# ===== 角色守卫 =====

def test_operator_cannot_grant_coins(admin_client: TestClient, operator_token: str) -> None:
    uid = _seed_user("13900000004")
    r = admin_client.post(
        f"/admin/api/users/{uid}/coins", json={"amount": 100, "reason": "x"},
        headers=_auth(operator_token),
    )
    assert r.status_code == 403


def test_finance_cannot_manage_admins(admin_client: TestClient, finance_token: str) -> None:
    assert admin_client.get("/admin/api/admins", headers=_auth(finance_token)).status_code == 403


def test_super_admin_can_grant_coins(admin_client: TestClient, super_token: str) -> None:
    uid = _seed_user("13900000005")
    r = admin_client.post(
        f"/admin/api/users/{uid}/coins", json={"amount": 50, "reason": "x"},
        headers=_auth(super_token),
    )
    assert r.status_code == 200


# ===== 反馈处理 =====

def test_handle_feedback(admin_client: TestClient, operator_token: str) -> None:
    fid = _seed_feedback("13900000006")
    r = admin_client.post(f"/admin/api/feedbacks/{fid}/handle", headers=_auth(operator_token))
    assert r.status_code == 200
    db = SessionLocal()
    try:
        assert db.get(Feedback, fid).status == "handled"
    finally:
        db.close()


# ===== admin 账号管理(super_admin) =====

def test_create_and_update_admin(admin_client: TestClient, super_token: str) -> None:
    r = admin_client.post(
        "/admin/api/admins",
        json={"username": "new_op", "password": "pass1234", "role": "operator"},
        headers=_auth(super_token),
    )
    assert r.status_code == 200, r.text
    new_id = r.json()["id"]
    r = admin_client.patch(
        f"/admin/api/admins/{new_id}", json={"role": "finance"}, headers=_auth(super_token)
    )
    assert r.status_code == 200 and r.json()["role"] == "finance"
    r = admin_client.post(
        "/admin/api/admins", json={"username": "new_op", "password": "pass1234"},
        headers=_auth(super_token),
    )
    assert r.status_code == 409


def test_cannot_disable_self(admin_client: TestClient, super_token: str) -> None:
    me = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
    r = admin_client.patch(
        f"/admin/api/admins/{me['id']}", json={"status": "disabled"}, headers=_auth(super_token)
    )
    assert r.status_code == 400


# ===== 提现重试 / 对账(mock wxpay,不真调微信) =====

def test_withdraw_refresh(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
    uid = _seed_user("13900000007")
    db = SessionLocal()
    try:
        db.add(WithdrawOrder(
            user_id=uid, out_bill_no="adminrefresh0001", amount_cents=100, status="pending"
        ))
        db.commit()
    finally:
        db.close()
    from app.repositories import wallet as wr
    monkeypatch.setattr(
        wr.wxpay, "query_transfer",
        lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
    )
    r = admin_client.post(
        "/admin/api/withdraws/adminrefresh0001/refresh", headers=_auth(finance_token)
    )
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "success"
    db = SessionLocal()
    try:
        logs = db.execute(
            select(AdminAuditLog).where(AdminAuditLog.action == "withdraw.refresh")
        ).scalars().all()
        assert any(x.target_id == "adminrefresh0001" for x in logs)
    finally:
        db.close()


def test_withdraw_refresh_404(admin_client: TestClient, finance_token: str) -> None:
    assert admin_client.post(
        "/admin/api/withdraws/nope999notexist/refresh", headers=_auth(finance_token)
    ).status_code == 404


def test_withdraw_reconcile(admin_client: TestClient, finance_token: str, monkeypatch) -> None:
    from app.repositories import wallet as wr
    monkeypatch.setattr(
        wr.wxpay, "query_transfer",
        lambda obn: {"status_code": 200, "data": {"state": "SUCCESS"}},
    )
    r = admin_client.post(
        "/admin/api/withdraws/reconcile", params={"older_than_minutes": 0},
        headers=_auth(finance_token),
    )
    assert r.status_code == 200, r.text
    assert "checked" in r.json() and "resolved" in r.json()