shaguabijia-app-server/tests/test_admin.py

"""Admin 后台 M1 测试:登录闭环 + 与 App 用户鉴权的彻底隔离。

admin app 是独立的 FastAPI(app.admin.main:admin_app),用独立 TestClient。
admin 表由 conftest 的 Base.metadata.create_all 一起建好(models/__init__ 已登记)。
"""
from __future__ import annotations

import pytest
from fastapi.testclient import TestClient

from app.admin.main import admin_app
from app.admin.repositories import admin_user as admin_repo
from app.core.security import create_token, hash_password
from app.db.session import SessionLocal


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


def _ensure_admin(
    username: str = "test_admin",
    password: str = "admin-pass-123",
    role: str = "super_admin",
) -> tuple[str, str]:
    """create-or-reset 一个 admin(测试 DB 跨用例共享,需幂等)。"""
    db = SessionLocal()
    try:
        a = admin_repo.get_by_username(db, username)
        if a is None:
            admin_repo.create_admin(db, username=username, password=password, role=role)
        else:
            a.password_hash = hash_password(password)
            a.role = role
            a.status = "active"
            db.commit()
    finally:
        db.close()
    return username, password


def test_admin_login_and_me(admin_client: TestClient) -> None:
    username, password = _ensure_admin()

    r = admin_client.post(
        "/admin/api/auth/login", json={"username": username, "password": password}
    )
    assert r.status_code == 200, r.text
    data = r.json()
    assert "access_token" in data
    assert data["admin"]["username"] == username
    assert data["admin"]["role"] == "super_admin"

    token = data["access_token"]
    r = admin_client.get("/admin/api/auth/me", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200, r.text
    assert r.json()["username"] == username


def test_admin_login_wrong_password(admin_client: TestClient) -> None:
    username, _ = _ensure_admin()
    r = admin_client.post(
        "/admin/api/auth/login", json={"username": username, "password": "definitely-wrong"}
    )
    assert r.status_code == 401


def test_admin_me_requires_token(admin_client: TestClient) -> None:
    assert admin_client.get("/admin/api/auth/me").status_code == 401


def test_disabled_admin_cannot_login(admin_client: TestClient) -> None:
    username, password = _ensure_admin(username="disabled_admin")
    db = SessionLocal()
    try:
        a = admin_repo.get_by_username(db, username)
        a.status = "disabled"
        db.commit()
    finally:
        db.close()
    r = admin_client.post(
        "/admin/api/auth/login", json={"username": username, "password": password}
    )
    assert r.status_code == 403


# ============================ 关键:鉴权隔离 ============================

def test_app_user_token_cannot_access_admin(admin_client: TestClient) -> None:
    """App 用户的 access_token(用 JWT_SECRET_KEY 签)不能访问 admin。

    admin 用独立 ADMIN_JWT_SECRET 验签 → App token 直接验签失败 → 401。
    这是后台防越权的第一道线。
    """
    user_token, _ = create_token(user_id=1, token_type="access")
    r = admin_client.get(
        "/admin/api/auth/me", headers={"Authorization": f"Bearer {user_token}"}
    )
    assert r.status_code == 401


def test_admin_token_cannot_access_app_api(client: TestClient, admin_client: TestClient) -> None:
    """反向:admin token 也不能访问 App 用户接口(App 用 JWT_SECRET_KEY 验签,admin token 失败)。"""
    username, password = _ensure_admin()
    r = admin_client.post(
        "/admin/api/auth/login", json={"username": username, "password": password}
    )
    admin_token = r.json()["access_token"]
    r = client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {admin_token}"})
    assert r.status_code == 401


# ============================ 回归:review 修掉的两个 bug ============================

def test_long_password_does_not_crash(admin_client: TestClient) -> None:
    """>72 UTF-8 字节的密码(如多个中文)不能让建账号/登录崩(bcrypt 72 字节限制)。"""
    username = "longpw_admin"
    long_pw = "超长密码测试" * 8  # 6 中文 ×8 = 48 字 ≈ 144 字节 UTF-8,远超 72
    _ensure_admin(username=username, password=long_pw)  # 建账号不应抛 ValueError
    r = admin_client.post(
        "/admin/api/auth/login", json={"username": username, "password": long_pw}
    )
    assert r.status_code == 200, r.text


def test_audit_log_pagination_no_gap() -> None:
    """审计游标分页跨页不丢/不重(回归 next_cursor off-by-one)。"""
    from app.admin.repositories import admin_user as admin_repo
    from app.admin.repositories import audit_log as audit_repo

    _ensure_admin()  # 确保有 test_admin 供 FK 引用
    db = SessionLocal()
    try:
        admin = admin_repo.get_by_username(db, "test_admin")
        action = "test.pagination.probe"
        created_ids = []
        for i in range(5):
            log = audit_repo.add_audit_log(
                db, admin_id=admin.id, admin_username=admin.username,
                action=action, target_type="probe", target_id=str(i),
            )
            created_ids.append(log.id)

        # limit=2 翻 5 条,收集所有 id,应正好覆盖创建的 5 条(无丢无重)
        seen: list[int] = []
        cursor = None
        for _ in range(10):  # 上限防死循环
            rows, cursor = audit_repo.list_audit_logs(db, action=action, limit=2, cursor=cursor)
            seen.extend(r.id for r in rows)
            if cursor is None:
                break
        assert sorted(seen) == sorted(created_ids), f"分页丢/重: want={created_ids} got={seen}"
    finally:
        db.close()