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()}
    # 非 hidden 项照常返回;看广告组保留可见的:每日上限 / 单次金币上限 / 关闭后冷却。
    assert "signin_rewards" in items and "ad_daily_limit" in items and "ad_cooldown_sec" in items
    # hidden 项(任务/里程碑、首页轮播数据源、看广告组的单次金币/每轮次数/信息流广告开关)不在配置页返回。
    for hidden_key in (
        "task_rewards", "record_milestones", "marquee_feed_mode",
        "ad_reward_coin", "ad_round_count", "comparing_ad_enabled",
    ):
        assert hidden_key not in items, f"{hidden_key} 应被 hidden 过滤"
    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_update_bool_config(admin_client: TestClient, token: str) -> None:
    # 提现自动对账开关默认 True
    items = {
        i["key"]: i
        for i in admin_client.get("/admin/api/config", headers=_auth(token)).json()
    }
    assert items["withdraw_auto_reconcile_enabled"]["type"] == "bool"
    assert items["withdraw_auto_reconcile_enabled"]["value"] is True

    # 关掉 → DB 落 False、业务读到 False
    r = admin_client.patch(
        "/admin/api/config/withdraw_auto_reconcile_enabled",
        json={"value": False},
        headers=_auth(token),
    )
    assert r.status_code == 200, r.text
    assert r.json()["value"] is False and r.json()["overridden"] is True

    db = SessionLocal()
    try:
        from app.repositories import app_config

        assert app_config.get_value(db, "withdraw_auto_reconcile_enabled") is False
    finally:
        db.close()

    # bool 项不接受非布尔值
    assert admin_client.patch(
        "/admin/api/config/withdraw_auto_reconcile_enabled",
        json={"value": 1},
        headers=_auth(token),
    ).status_code == 400


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