OuYingJun1024
995908cd0b
- 看视频奖励 LT 因子(因子2)按「账号累计第 N 次看视频」递减,不再按天重置: ad_reward 新增 _granted_cumulative、ad_feed_reward 单位序号去掉当天过滤、 rewards 形参 today_count_after_this → count_after_this;每日次数上限/冷却仍按当日统计 - 广告金币审计:加 only_mismatch 筛选只看 ✗ 行;total/mismatch_count 改全量统计 (不受 limit/筛选影响、截断前算)+ 新增 truncated 标记展示集是否被截断 - admin 用户金币/现金接口加 mode=delta|set:set=设为目标值(读余额算差值、仍写一笔流水, 沿用原子/审计/扣负保护);新增 admin-user-cash 文档 + 更新 API 索引/coins 等文档 + 补 admin read/write 测试
375 lines
13 KiB
Python
375 lines
13 KiB
Python
"""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_coins_writes_delta_txn(admin_client: TestClient, finance_token: str) -> None:
|
|
"""set 模式:先有余额,再设为目标值,只写一笔差值流水,审计带 mode/target/before。"""
|
|
uid = _seed_user("13900000009")
|
|
# 先增到 300
|
|
admin_client.post(
|
|
f"/admin/api/users/{uid}/coins", json={"amount": 300, "reason": "底"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
# 设为 100 → 差值 -200(扣减)
|
|
r = admin_client.post(
|
|
f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": 100, "reason": "设值"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
db = SessionLocal()
|
|
try:
|
|
assert db.get(CoinAccount, uid).coin_balance == 100
|
|
txns = db.execute(
|
|
select(CoinTransaction).where(CoinTransaction.user_id == uid)
|
|
.order_by(CoinTransaction.id.desc())
|
|
).scalars().all()
|
|
# 两笔:+300(admin_grant)、-200(admin_deduct)
|
|
assert txns[0].amount == -200 and txns[0].biz_type == "admin_deduct"
|
|
logs = db.execute(
|
|
select(AdminAuditLog).where(
|
|
AdminAuditLog.action == "user.coins.grant", AdminAuditLog.target_id == str(uid)
|
|
).order_by(AdminAuditLog.id.desc())
|
|
).scalars().all()
|
|
assert logs[0].detail["mode"] == "set"
|
|
assert logs[0].detail["target"] == 100
|
|
assert logs[0].detail["before"] == 300
|
|
assert logs[0].detail["amount"] == -200
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_set_coins_equal_balance_rejected(admin_client: TestClient, finance_token: str) -> None:
|
|
"""set 为当前余额(差值 0)→ 拒绝,不写流水。"""
|
|
uid = _seed_user("13900000010")
|
|
admin_client.post(
|
|
f"/admin/api/users/{uid}/coins", json={"amount": 50, "reason": "底"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
r = admin_client.post(
|
|
f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": 50, "reason": "x"},
|
|
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) == 1 # 仅初始那笔,设值被拒后无新流水
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_set_coins_negative_target_rejected(admin_client: TestClient, finance_token: str) -> None:
|
|
uid = _seed_user("13900000011")
|
|
r = admin_client.post(
|
|
f"/admin/api/users/{uid}/coins", json={"mode": "set", "amount": -1, "reason": "x"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_set_cash_writes_delta_txn(admin_client: TestClient, finance_token: str) -> None:
|
|
"""现金 set 模式:设为目标分值,写一笔差值流水。"""
|
|
uid = _seed_user("13900000012")
|
|
admin_client.post(
|
|
f"/admin/api/users/{uid}/cash", json={"amount_cents": 500, "reason": "底"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
r = admin_client.post(
|
|
f"/admin/api/users/{uid}/cash",
|
|
json={"mode": "set", "amount_cents": 200, "reason": "设值"},
|
|
headers=_auth(finance_token),
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
db = SessionLocal()
|
|
try:
|
|
assert db.get(CoinAccount, uid).cash_balance_cents == 200
|
|
logs = db.execute(
|
|
select(AdminAuditLog).where(
|
|
AdminAuditLog.action == "user.cash.grant", AdminAuditLog.target_id == str(uid)
|
|
).order_by(AdminAuditLog.id.desc())
|
|
).scalars().all()
|
|
assert logs[0].detail["mode"] == "set"
|
|
assert logs[0].detail["target_cents"] == 200
|
|
assert logs[0].detail["before_cents"] == 500
|
|
assert logs[0].detail["amount_cents"] == -300
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# ===== 封号 =====
|
|
|
|
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_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()
|