ajyy/backend/tests/test_role_menus.py

"""用户管理 / 角色授权 / 操作审计 API 测试"""
from app.core.database import get_db
from app.core.security import hash_password
from app.crud import user as user_crud
from app.main import app as fastapi_app


def _login(client, username: str, password: str) -> dict:
    r = client.post(
        "/api/web/auth/login", json={"username": username, "password": password}
    )
    assert r.status_code == 200, r.text
    token = r.json()["data"]["token"]
    return {"Authorization": f"Bearer {token}"}


def _client_db():
    gen = fastapi_app.dependency_overrides[get_db]()
    return next(gen)


def _ensure_role_user(client, username: str, role_code: str, password: str = "Pass@123"):
    db = _client_db()
    role = user_crud.get_role_by_code(db, role_code)
    assert role is not None
    existing = user_crud.get_user_by_username(db, username)
    if not existing:
        user_crud.create_user(
            db,
            username=username,
            password_hash=hash_password(password),
            display_name=username,
            role_id=role.id,
        )
    return _login(client, username, password)


def test_role_grant_updates_menus(client):
    admin_headers = _login(client, "admin", "Admin@123")

    roles = client.get("/api/web/admin/roles", headers=admin_headers).json()["data"]
    viewer = next(r for r in roles if r["roleCode"] == "viewer")

    menus_before = client.get(
        f"/api/web/admin/roles/{viewer['id']}/menus", headers=admin_headers
    ).json()["data"]
    menu_ids = list(menus_before["menuIds"])
    assert len(menu_ids) >= 1

    tree = client.get("/api/web/admin/menus/tree", headers=admin_headers).json()["data"]

    def _flatten(nodes):
        out = []
        for n in nodes or []:
            out.append(n)
            out.extend(_flatten(n.get("children") or []))
        return out

    flat = _flatten(tree)
    suggestion = next(m for m in flat if m["menuCode"] == "suggestion")
    new_ids = [mid for mid in menu_ids if mid != suggestion["id"]]
    assert suggestion["id"] in menu_ids

    put = client.put(
        f"/api/web/admin/roles/{viewer['id']}/menus",
        headers=admin_headers,
        json={"menuIds": new_ids},
    )
    assert put.status_code == 200, put.text

    viewer_headers = _ensure_role_user(client, "viewer_grant", "viewer")
    menus = client.get("/api/web/auth/menus", headers=viewer_headers).json()["data"]
    tree_codes = {m["menuCode"] for m in _flatten(menus["tree"])}
    assert "suggestion" not in tree_codes
    assert "dashboard" in tree_codes
    assert "feedback" in tree_codes


def test_user_manage_crud(client):
    admin_headers = _login(client, "admin", "Admin@123")
    roles = client.get("/api/web/admin/roles", headers=admin_headers).json()["data"]
    handler = next(r for r in roles if r["roleCode"] == "handler")

    created = client.post(
        "/api/web/admin/users",
        headers=admin_headers,
        json={
            "username": "u_task7",
            "password": "Pass@123",
            "displayName": "任务七用户",
            "roleId": handler["id"],
        },
    )
    assert created.status_code == 200, created.text
    user = created.json()["data"]
    assert user["username"] == "u_task7"
    assert user["displayName"] == "任务七用户"
    assert user["roleId"] == handler["id"]

    listed = client.get("/api/web/admin/users", headers=admin_headers)
    assert listed.status_code == 200
    usernames = {u["username"] for u in listed.json()["data"]}
    assert "u_task7" in usernames

    updated = client.put(
        f"/api/web/admin/users/{user['id']}",
        headers=admin_headers,
        json={"displayName": "已改名", "status": "1"},
    )
    assert updated.status_code == 200
    assert updated.json()["data"]["displayName"] == "已改名"
    assert updated.json()["data"]["status"] == "1"

    reset = client.put(
        f"/api/web/admin/users/{user['id']}/reset-password",
        headers=admin_headers,
        json={"password": "NewPass@123"},
    )
    assert reset.status_code == 200

    # 启停后不可登录；先启再用新密码登录
    client.put(
        f"/api/web/admin/users/{user['id']}",
        headers=admin_headers,
        json={"status": "0"},
    )
    login = client.post(
        "/api/web/auth/login",
        json={"username": "u_task7", "password": "NewPass@123"},
    )
    assert login.status_code == 200


def test_oper_logs_list_requires_audit(client):
    admin_headers = _login(client, "admin", "Admin@123")
    r = client.get(
        "/api/web/admin/oper-logs",
        headers=admin_headers,
        params={"page": 1, "pageSize": 10},
    )
    assert r.status_code == 200
    data = r.json()["data"]
    assert "list" in data
    assert "total" in data
    assert data["page"] == 1
    assert data["pageSize"] == 10
    assert data["total"] >= 1  # login 已写日志

    viewer_headers = _ensure_role_user(client, "viewer_audit", "viewer")
    denied = client.get(
        "/api/web/admin/oper-logs",
        headers=viewer_headers,
        params={"page": 1, "pageSize": 10},
    )
    assert denied.status_code == 403


def test_oper_logs_filter_by_username_and_keyword(client):
    from app.crud import oper_log as oper_log_crud

    db = _client_db()
    oper_log_crud.create_oper_log(
        db,
        module="用户管理",
        action="创建",
        username="filter_alice",
        detail="创建用户 alice_unique_detail",
    )
    oper_log_crud.create_oper_log(
        db,
        module="角色管理",
        action="更新",
        username="filter_bob",
        detail="更新角色 bob_other_detail",
    )

    admin_headers = _login(client, "admin", "Admin@123")

    by_user = client.get(
        "/api/web/admin/oper-logs",
        headers=admin_headers,
        params={"username": "filter_alice", "page": 1, "pageSize": 10},
    )
    assert by_user.status_code == 200
    user_data = by_user.json()["data"]
    assert user_data["total"] >= 1
    assert all("filter_alice" in (row["username"] or "") for row in user_data["list"])

    by_keyword = client.get(
        "/api/web/admin/oper-logs",
        headers=admin_headers,
        params={"keyword": "alice_unique_detail", "page": 1, "pageSize": 10},
    )
    assert by_keyword.status_code == 200
    kw_data = by_keyword.json()["data"]
    assert kw_data["total"] >= 1
    assert all(
        "alice_unique_detail" in (row["detail"] or "") for row in kw_data["list"]
    )
    assert not any(
        "bob_other_detail" in (row["detail"] or "") for row in kw_data["list"]
    )


def test_role_create_and_update(client):
    admin_headers = _login(client, "admin", "Admin@123")

    created = client.post(
        "/api/web/admin/roles",
        headers=admin_headers,
        json={
            "roleCode": "custom_role",
            "roleName": "自定义角色",
            "remark": "备注A",
        },
    )
    assert created.status_code == 200, created.text
    role = created.json()["data"]
    assert role["roleCode"] == "custom_role"
    assert role["roleName"] == "自定义角色"
    assert role["remark"] == "备注A"
    assert role["id"]

    dup = client.post(
        "/api/web/admin/roles",
        headers=admin_headers,
        json={"roleCode": "custom_role", "roleName": "重复编码"},
    )
    assert dup.status_code == 400

    updated = client.put(
        f"/api/web/admin/roles/{role['id']}",
        headers=admin_headers,
        json={"roleName": "已改名", "remark": "备注B"},
    )
    assert updated.status_code == 200, updated.text
    data = updated.json()["data"]
    assert data["roleCode"] == "custom_role"
    assert data["roleName"] == "已改名"
    assert data["remark"] == "备注B"

    listed = client.get("/api/web/admin/roles", headers=admin_headers).json()["data"]
    found = next(r for r in listed if r["id"] == role["id"])
    assert found["roleName"] == "已改名"


def test_role_delete(client):
    admin_headers = _login(client, "admin", "Admin@123")

    created = client.post(
        "/api/web/admin/roles",
        headers=admin_headers,
        json={"roleCode": "to_delete", "roleName": "待删角色"},
    )
    assert created.status_code == 200, created.text
    role = created.json()["data"]

    # 授权后再删，菜单关联应一并清理
    tree = client.get("/api/web/admin/menus/tree", headers=admin_headers).json()["data"]

    def _collect(nodes):
        ids = []
        for n in nodes:
            ids.append(n["id"])
            ids.extend(_collect(n.get("children") or []))
        return ids

    menu_ids = _collect(tree)[:1]
    assert menu_ids
    grant = client.put(
        f"/api/web/admin/roles/{role['id']}/menus",
        headers=admin_headers,
        json={"menuIds": menu_ids},
    )
    assert grant.status_code == 200, grant.text

    deleted = client.delete(
        f"/api/web/admin/roles/{role['id']}", headers=admin_headers
    )
    assert deleted.status_code == 200, deleted.text

    listed = client.get("/api/web/admin/roles", headers=admin_headers).json()["data"]
    assert all(r["id"] != role["id"] for r in listed)

    # 再次删除应报不存在
    again = client.delete(
        f"/api/web/admin/roles/{role['id']}", headers=admin_headers
    )
    assert again.status_code == 400


def test_role_delete_rejects_admin_and_in_use(client):
    admin_headers = _login(client, "admin", "Admin@123")
    roles = client.get("/api/web/admin/roles", headers=admin_headers).json()["data"]
    admin_role = next(r for r in roles if r["roleCode"] == "admin")
    viewer = next(r for r in roles if r["roleCode"] == "viewer")

    deny_admin = client.delete(
        f"/api/web/admin/roles/{admin_role['id']}", headers=admin_headers
    )
    assert deny_admin.status_code == 400

    # viewer 已被测试用户占用时不可删
    _ensure_role_user(client, "viewer_keep", "viewer")
    deny_in_use = client.delete(
        f"/api/web/admin/roles/{viewer['id']}", headers=admin_headers
    )
    assert deny_in_use.status_code == 400
    assert "用户" in deny_in_use.json().get("message", "")