shaguabijia-app-server/tests/test_huawei_review.py

"""华为审核开关:admin 读写 + 客户端公开端点 + 审计 + 空库回退。

背景:华为应用市场审核要求新手引导的「快速设置」权限步必须可被用户关闭。开关切到 review 后
客户端(仅华为 ROM)在该步左上角显示退出按钮。默认 default = 上线至今的现状(不可关闭)。

autouse 清理每个用例后清空 app_config,避免污染其他文件里假设默认值的用例(同 test_admin_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


@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, "hw_admin") is None:
            admin_repo.create_admin(
                db, username="hw_admin", password="hwpass123", role="super_admin"
            )
    finally:
        db.close()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": "hw_admin", "password": "hwpass123"}
    ).json()["access_token"]


@pytest.fixture(autouse=True)
def _clean_config() -> Iterator[None]:
    yield
    db = SessionLocal()
    try:
        db.execute(delete(AppConfig))
        # 审计行同样要清:同库跨用例累积会让 test_switch_writes_audit 数到前面用例写的行
        # (按 action 限定,不碰其他模块可能已写入的审计)。
        db.execute(delete(AdminAuditLog).where(AdminAuditLog.action == "huawei_review.set"))
        db.commit()
    finally:
        db.close()


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


def test_public_default_not_closable(client: TestClient) -> None:
    """空库(从未切过)→ 客户端拿到 default / 不可关闭 = 维持现状;且不需要鉴权。"""
    r = client.get("/api/v1/platform/huawei-review")
    assert r.status_code == 200, r.text
    assert r.json() == {"mode": "default", "onboarding_closable": False}


def test_admin_get_default(admin_client: TestClient, token: str) -> None:
    r = admin_client.get("/admin/api/huawei-review", headers=_auth(token))
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["mode"] == "default"
    assert body["updated_at"] is None  # 从未切过


def test_switch_to_review_takes_effect(
    admin_client: TestClient, client: TestClient, token: str
) -> None:
    """admin 切 review → 公开端点立刻下发可关闭(跨进程/跨 app 一致,因为落在 DB 而非内存)。"""
    r = admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}, headers=_auth(token)
    )
    assert r.status_code == 200, r.text
    assert r.json()["mode"] == "review"
    assert r.json()["updated_at"] is not None

    pub = client.get("/api/v1/platform/huawei-review").json()
    assert pub == {"mode": "review", "onboarding_closable": True}

    # 切回 default → 客户端恢复不可关闭(审核过了要能一键收回)
    admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "default"}, headers=_auth(token)
    )
    assert client.get("/api/v1/platform/huawei-review").json()["onboarding_closable"] is False


def test_switch_writes_audit(admin_client: TestClient, token: str) -> None:
    admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}, headers=_auth(token)
    )
    db = SessionLocal()
    try:
        logs = db.execute(
            select(AdminAuditLog).where(AdminAuditLog.action == "huawei_review.set")
        ).scalars().all()
        assert len(logs) == 1
        assert logs[0].detail == {"before": "default", "after": "review"}
    finally:
        db.close()


def test_invalid_mode_rejected(admin_client: TestClient, token: str) -> None:
    """mode 是 Literal,非法值由 FastAPI 校验挡在 422(不会落库)。"""
    r = admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "nope"}, headers=_auth(token)
    )
    assert r.status_code == 422, r.text
    db = SessionLocal()
    try:
        assert db.get(AppConfig, "huawei_review") is None
    finally:
        db.close()


def test_requires_admin_auth(admin_client: TestClient) -> None:
    assert admin_client.get("/admin/api/huawei-review").status_code == 401
    assert admin_client.patch(
        "/admin/api/huawei-review", json={"mode": "review"}
    ).status_code == 401