feat(admin): app_config 运营配置后台(rewards 等常量可后台配置) (#14)

- 新增 app_config 表 + model/repository/config_schema + admin config 路由与 schema + 测试
- rewards.py 及 ad_reward/signin/task/wallet/comparison_milestone repositories 接入可配置项;ad.py / admin/main.py 配套
- alembic merge(ebb6af5c0b56)合并本地 app_config(cfg1a2b3c4d5)与 incoming price_report/best_deeplink(b7e2c1a9f4d3)两 head

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Reviewed-on: #14
This commit was merged in pull request #14.
This commit is contained in:
2026-06-06 04:13:08 +08:00
parent 4506677483
commit 67a9a15775
17 changed files with 561 additions and 49 deletions
+129
View File
@@ -0,0 +1,129 @@
"""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"] == [10, 20, 30, 50, 80, 120, 200]
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:
_used, limit, _coin, _rc, _cd = 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