shaguabijia-app-server/tests/test_admin_roles.py

"""Admin RBAC 角色/权限测试:可见页下发、角色 CRUD 仅 super_admin、内建/在用保护。"""
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 hash_password
from app.db.session import SessionLocal


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


def _token(username: str, role: str) -> str:
    db = SessionLocal()
    try:
        a = admin_repo.get_by_username(db, username)
        if a is None:
            admin_repo.create_admin(db, username=username, password="pass1234", role=role)
        else:
            a.password_hash = hash_password("pass1234")
            a.role = role
            a.status = "active"
            db.commit()
    finally:
        db.close()
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": username, "password": "pass1234"}
    ).json()["access_token"]


@pytest.fixture()
def super_token() -> str:
    return _token("r_super", "super_admin")


@pytest.fixture()
def operator_token() -> str:
    return _token("r_operator", "operator")


@pytest.fixture()
def tech_token() -> str:
    return _token("r_tech", "tech")


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


def test_super_pages_all_operator_limited(admin_client, super_token, operator_token) -> None:
    su = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
    assert "admins" in su["pages"] and "analytics-health" in su["pages"]  # 超管全页
    op = admin_client.get("/admin/api/auth/me", headers=_auth(operator_token)).json()
    assert "dashboard" in op["pages"] and "admins" not in op["pages"]  # 运营看不到管理员页


def test_monitoring_audit_catalog_and_api_permissions(
    admin_client, super_token, operator_token, tech_token
) -> None:
    catalog = admin_client.get(
        "/admin/api/roles/catalog", headers=_auth(super_token)
    ).json()
    monitoring = next(group for group in catalog if group["group"] == "监控审计")
    assert [page["key"] for page in monitoring["pages"]] == [
        "device-liveness", "analytics-health", "event-logs", "audit-logs",
    ]

    # 运营默认只能查设备存活，不能绕过导航直调技术/审计接口。
    assert admin_client.get(
        "/admin/api/device-liveness/stats", headers=_auth(operator_token)
    ).status_code == 200
    for path in (
        "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
        "/admin/api/event-logs",
        "/admin/api/audit-logs",
    ):
        assert admin_client.get(path, headers=_auth(operator_token)).status_code == 403

    # 技术角色默认拥有监控审计组全部四项权限。
    for path in (
        "/admin/api/device-liveness/stats",
        "/admin/api/analytics-health/overview?date_from=2026-07-01T00:00:00Z&date_to=2026-07-02T00:00:00Z",
        "/admin/api/event-logs",
        "/admin/api/audit-logs",
    ):
        assert admin_client.get(path, headers=_auth(tech_token)).status_code == 200


def test_custom_admin_api_permission_uses_pages_override(admin_client) -> None:
    db = SessionLocal()
    try:
        admin = admin_repo.get_by_username(db, "r_monitoring_custom")
        if admin is None:
            admin = admin_repo.create_admin(
                db, username="r_monitoring_custom", password="pass1234", role="custom"
            )
        admin.password_hash = hash_password("pass1234")
        admin.role = "custom"
        admin.pages_override = ["event-logs"]
        admin.status = "active"
        db.commit()
    finally:
        db.close()

    token = admin_client.post(
        "/admin/api/auth/login",
        json={"username": "r_monitoring_custom", "password": "pass1234"},
    ).json()["access_token"]
    assert admin_client.get("/admin/api/event-logs", headers=_auth(token)).status_code == 200
    assert admin_client.get("/admin/api/audit-logs", headers=_auth(token)).status_code == 403


def test_roles_endpoints_super_only(admin_client, super_token, operator_token) -> None:
    assert admin_client.get("/admin/api/roles", headers=_auth(super_token)).status_code == 200
    assert admin_client.get("/admin/api/roles", headers=_auth(operator_token)).status_code == 403


def test_role_crud(admin_client, super_token) -> None:
    cat = admin_client.get("/admin/api/roles/catalog", headers=_auth(super_token)).json()
    assert any(g["group"] == "看板" for g in cat)

    r = admin_client.post(
        "/admin/api/roles", json={"name": "审核", "pages": ["dashboard", "feedbacks", "xx-bad"]},
        headers=_auth(super_token),
    )
    assert r.status_code == 200, r.text
    rid = r.json()["id"]
    assert set(r.json()["pages"]) == {"dashboard", "feedbacks"}  # 非法 key 被过滤

    # 重名 409
    assert admin_client.post(
        "/admin/api/roles", json={"name": "审核", "pages": []}, headers=_auth(super_token)
    ).status_code == 409

    # 改可见页
    u = admin_client.patch(
        f"/admin/api/roles/{rid}", json={"pages": ["dashboard"]}, headers=_auth(super_token)
    )
    assert u.status_code == 200 and u.json()["pages"] == ["dashboard"]

    # 删
    assert admin_client.delete(f"/admin/api/roles/{rid}", headers=_auth(super_token)).status_code == 200


def test_builtin_super_admin_protected(admin_client, super_token) -> None:
    roles = admin_client.get("/admin/api/roles", headers=_auth(super_token)).json()
    sa = next(r for r in roles if r["name"] == "super_admin")
    assert sa["is_builtin"] and len(sa["pages"]) >= 10  # 全部页
    assert admin_client.patch(
        f"/admin/api/roles/{sa['id']}", json={"pages": []}, headers=_auth(super_token)
    ).status_code == 400
    assert admin_client.delete(
        f"/admin/api/roles/{sa['id']}", headers=_auth(super_token)
    ).status_code == 400


def test_delete_role_in_use_blocked(admin_client, super_token) -> None:
    admin_client.post(
        "/admin/api/admins",
        json={"username": "role_holder", "password": "pass1234", "role": "operator"},
        headers=_auth(super_token),
    )
    roles = admin_client.get("/admin/api/roles", headers=_auth(super_token)).json()
    op = next(r for r in roles if r["name"] == "operator")
    assert op["in_use"] >= 1
    assert admin_client.delete(
        f"/admin/api/roles/{op['id']}", headers=_auth(super_token)
    ).status_code == 400


def test_builtin_roles_labels_and_pages(admin_client, super_token) -> None:
    roles = {r["name"]: r for r in admin_client.get("/admin/api/roles", headers=_auth(super_token)).json()}
    # 中文展示名(key 不变)
    assert roles["super_admin"]["label"] == "管理员"
    assert roles["operator"]["label"] == "运营"
    assert roles["finance"]["label"] == "财务"
    assert roles["tech"]["label"] == "技术"
    # 页集对齐 Prototypes/dashboard/permissions.md 的 ROLES
    assert set(roles["finance"]["pages"]) == {"dashboard", "ad-revenue-report", "cps", "withdraws"}
    assert set(roles["tech"]["pages"]) == {
        "dashboard", "device-liveness", "analytics-health", "config", "ad-revenue", "huawei-review",
        "event-logs", "audit-logs",
    }


def test_ui_created_admin_shows_password_script_admin_hidden(admin_client, super_token) -> None:
    # UI 建号:账号列表回显明文登录密码(供权限管理页编辑复看)
    admin_client.post(
        "/admin/api/admins",
        json={"username": "pw_show_user", "password": "pass1234", "role": "operator"},
        headers=_auth(super_token),
    )
    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
    assert admins["pw_show_user"]["password"] == "pass1234"
    # 经 repo/脚本直建的账号(r_super,无留存明文)→ password 为 None,前端不显示
    assert admins["r_super"]["password"] is None
    # 登录 / me 不下发明文(保持 None)
    me = admin_client.get("/admin/api/auth/me", headers=_auth(super_token)).json()
    assert me.get("password") is None


def test_create_admin_rejects_unknown_role(admin_client, super_token) -> None:
    r = admin_client.post(
        "/admin/api/admins",
        json={"username": "bad_role_user", "password": "pass1234", "role": "不存在的角色"},
        headers=_auth(super_token),
    )
    assert r.status_code == 400


def _login_pages(username: str, password: str = "pass1234") -> list[str]:
    """以某账号登录,取下发的有效可见页(左侧导航过滤依据)。"""
    c = TestClient(admin_app)
    return c.post(
        "/admin/api/auth/login", json={"username": username, "password": password}
    ).json()["admin"]["pages"]


def test_custom_role_create_pages_take_effect(admin_client, super_token) -> None:
    # 建「自定义」账号:role=custom + 勾选页(含一个非法 key,应被过滤)
    r = admin_client.post(
        "/admin/api/admins",
        json={
            "username": "cust_user", "password": "pass1234", "role": "custom",
            "pages_override": ["dashboard", "withdraws", "xx-bad"],
        },
        headers=_auth(super_token),
    )
    assert r.status_code == 200, r.text
    # 登录下发的 pages == 勾选集(非法 key 过滤),真正驱动左侧导航
    assert set(_login_pages("cust_user")) == {"dashboard", "withdraws"}
    # 账号列表回显原始勾选集(供编辑回填),同样已过滤
    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
    assert set(admins["cust_user"]["pages_override"]) == {"dashboard", "withdraws"}
    assert admins["cust_user"]["role"] == "custom"


def test_custom_role_update_and_switch_back_clears_override(admin_client, super_token) -> None:
    admin_client.post(
        "/admin/api/admins",
        json={
            "username": "cust_sw", "password": "pass1234", "role": "custom",
            "pages_override": ["dashboard"],
        },
        headers=_auth(super_token),
    )
    aid = next(
        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
        if a["username"] == "cust_sw"
    )
    # 只改勾选集(角色不变)→ 生效
    u = admin_client.patch(
        f"/admin/api/admins/{aid}", json={"pages_override": ["dashboard", "feedbacks"]},
        headers=_auth(super_token),
    )
    assert u.status_code == 200, u.text
    assert set(_login_pages("cust_sw")) == {"dashboard", "feedbacks"}
    # 切回普通角色 operator → override 清空,pages 跟随角色(运营页集,且不含 admins)
    u2 = admin_client.patch(
        f"/admin/api/admins/{aid}", json={"role": "operator"}, headers=_auth(super_token)
    )
    assert u2.status_code == 200, u2.text
    admins = {a["username"]: a for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()}
    assert admins["cust_sw"]["pages_override"] is None
    pages = set(_login_pages("cust_sw"))
    assert "feedbacks" in pages and "admins" not in pages  # 运营口径,自定义页已不生效


def test_switch_operator_to_custom_sets_override(admin_client, super_token) -> None:
    admin_client.post(
        "/admin/api/admins",
        json={"username": "op2cust", "password": "pass1234", "role": "operator"},
        headers=_auth(super_token),
    )
    aid = next(
        a["id"] for a in admin_client.get("/admin/api/admins", headers=_auth(super_token)).json()
        if a["username"] == "op2cust"
    )
    u = admin_client.patch(
        f"/admin/api/admins/{aid}",
        json={"role": "custom", "pages_override": ["config"]},
        headers=_auth(super_token),
    )
    assert u.status_code == 200, u.text
    assert set(_login_pages("op2cust")) == {"config"}