shaguabijia-app-server/tests/test_admin_config.py

"""Admin M5 配置后台化测试:列出/改配置 + 改配**真生效** + 校验 + 审计。

autouse 清理每个用例后清空 app_config,避免改配污染其他文件的福利测试。
"""
from __future__ import annotations

from collections.abc import Iterator

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

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.admin import AdminAuditLog
from app.models.app_config import AppConfig
from app.repositories import ad_reward, signin
from app.repositories import user as user_repo


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


@pytest.fixture()
def token() -> str:
    db = SessionLocal()
    try:
        if admin_repo.get_by_username(db, "cfg_admin") is None:
            admin_repo.create_admin(
                db, username="cfg_admin", password="cfgpass12", role="super_admin"
            )
    finally:
        db.close()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": "cfg_admin", "password": "cfgpass12"}
    ).json()["access_token"]


@pytest.fixture(autouse=True)
def _clean_config() -> Iterator[None]:
    """每个用例后清空 app_config,避免改配污染其他文件的福利测试(它们假设默认值)。"""
    yield
    db = SessionLocal()
    try:
        db.execute(delete(AppConfig))
        db.commit()
    finally:
        db.close()


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


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 test_list_config(admin_client: TestClient, token: str) -> None:
    r = admin_client.get("/admin/api/config", headers=_auth(token))
    assert r.status_code == 200, r.text
    items = {i["key"]: i for i in r.json()}
    assert "signin_rewards" in items and "ad_daily_limit" in items
    assert items["signin_rewards"]["value"] == [
        200, 200, 300, 200, 400, 400, 800,
    ]
    assert items["signin_rewards"]["overridden"] is False


def test_update_signin_takes_effect(admin_client: TestClient, token: str) -> None:
    r = admin_client.patch(
        "/admin/api/config/signin_rewards",
        json={"value": [100, 200, 300, 400, 500, 600, 700]},
        headers=_auth(token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["value"][0] == 100 and r.json()["overridden"] is True

    # 业务真的用上新值(配置后台化的核心验证)
    uid = _seed_user("13912340001")
    db = SessionLocal()
    try:
        st = signin.get_status(db, uid)
        assert st.steps[0].coin == 100
        assert st.today_coin == 100  # 首签=第 1 天=100(新值)
        logs = db.execute(
            select(AdminAuditLog).where(
                AdminAuditLog.action == "config.set",
                AdminAuditLog.target_id == "signin_rewards",
            )
        ).scalars().all()
        assert len(logs) == 1 and logs[0].detail["after"][0] == 100
    finally:
        db.close()


def test_update_ad_limit_takes_effect(admin_client: TestClient, token: str) -> None:
    r = admin_client.patch(
        "/admin/api/config/ad_daily_limit", json={"value": 5}, headers=_auth(token)
    )
    assert r.status_code == 200, r.text
    uid = _seed_user("13912340002")
    db = SessionLocal()
    try:
        # today_status 现返 7 元组(末两位为观看时长闸:已看秒数 / 上限),取次数上限位
        _used, limit, _coin, _rc, _cd, _ws, _wl = ad_reward.today_status(db, uid)
        assert limit == 5  # 新配置生效
    finally:
        db.close()


def test_config_validation(admin_client: TestClient, token: str) -> None:
    # 签到档位长度≠7
    assert admin_client.patch(
        "/admin/api/config/signin_rewards", json={"value": [1, 2, 3]}, headers=_auth(token)
    ).status_code == 400
    # int 负数
    assert admin_client.patch(
        "/admin/api/config/ad_daily_limit", json={"value": -5}, headers=_auth(token)
    ).status_code == 400
    # 未知 key
    assert admin_client.patch(
        "/admin/api/config/nope", json={"value": 1}, headers=_auth(token)
    ).status_code == 404