From 6812fbe3174c1ec19d41bea3f54c3bd0f106a5b6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Fri, 11 Sep 2026 21:22:56 +0800 Subject: [PATCH] =?UTF-8?q?B1=EF=BC=9ARBAC=20=E5=8F=AA=E8=AF=BB=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E6=8E=A5=E5=8F=A3=EF=BC=884=20=E4=B8=AA=EF=BC=89+=20d?= =?UTF-8?q?ocs/05=20=E7=99=BB=E8=AE=B0=20A035-A038?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在此之前，权限只能靠脚本改，平台里**没有任何地方能看"谁能访问什么"**。 B1 先把"看得见"做出来： - GET /api/v1/admin/roles 角色清单 + 权限数 + 在用人数 - GET /api/v1/admin/roles/{role_code} 角色详情 - GET /api/v1/admin/roles/{role_code}/permissions 权限清单（按权限码排序， 便于与各 Service 的 require() 对照） - GET /api/v1/admin/users/{user_id}/roles 某人**实际解析出来**的角色/权限/数据范围 几个刻意的决定： 1. 权限码复用 `audit:read` 而不新增 `rbac:read`：这份清单本身就是审计材料，且复用是零数据 改动、立刻可用（新增权限码要先改 sys_permission，而它目前由 seed_test_rbac.py 以 DELETE 重建语义管理）。将来要细分再加，不冲突。 2. `/users/{id}/roles` 直接复用 IdentityService.resolve，不自己拼 SQL —— 那正是请求进来时 走的链路（status 检查、assigned_at/expires_at 时间窗、data_scope 取最高、客户分配）。 自己写一遍必然漂移，而"这里查出的权限"与"实际能用的权限"不一致比没有这个接口更糟。 测试里加了一条交叉验证：该接口的 roles/data_scope 必须与登录响应完全一致。 3. 停用账号返回**空权限集**而不是 404 —— 用户存在但拿不到权限，如实呈现比 404 更利于排障。 4. 角色详情与权限清单分开：权限为空的角色不该被误判成"角色不存在"。 5. 全部只读、不写审计（它们返回的就是审计材料本身），docs/05 §19 登记时审计列标"否"。 权限变更（提权/降权）仍无接口 —— docs/05 §19 已注明那不是遗漏，而是需要单独评审 （审计留痕 + 禁止自我提权 + 保护内置角色三条红线）。 另：确认了 load_context 对 sys_user_role.expires_at 是有过滤的（assigned_at<=now AND (expires_at IS NULL OR expires_at>now)），我上一轮只读了半段 SQL 差点误报。 验证：ruff 干净 / mypy 185 文件 0 错 / 文档守卫 38 份无编号冲突 / unit+contract 1140 passed / integration 98 passed（本批新增 8 个）。 --- app/api/controllers/rbac.py | 84 +++++++++ app/main.py | 2 + app/service/rbac_query_service.py | 202 ++++++++++++++++++++++ docs/05-接口文档.md | 15 ++ tests/integration/test_rbac_read_mysql.py | 184 ++++++++++++++++++++ 5 files changed, 487 insertions(+) create mode 100644 app/api/controllers/rbac.py create mode 100644 app/service/rbac_query_service.py create mode 100644 tests/integration/test_rbac_read_mysql.py diff --git a/app/api/controllers/rbac.py b/app/api/controllers/rbac.py new file mode 100644 index 0000000..275b7d7 --- /dev/null +++ b/app/api/controllers/rbac.py @@ -0,0 +1,84 @@ +"""角色与权限的只读接口（B1：先让管理员看得见）。 + +前缀与 `admin.py` 相同（`/api/v1/admin`），但**独立成文件**：`admin.py` 是表驱动的 +配置面 CRUD 工厂（`RESOURCES` 单表增删改），而角色/权限是**多对多关系查询**， +塞进那个工厂既不自然、也会让"配置管理"和"身份管理"两件事混在一个文件里。 + +四个接口全部只读，且**不写审计**——它们返回的就是审计材料本身（"谁能访问什么"， +合规检查要看的正是这份清单）。`docs/05` §19 登记时"审计"列标"否"。 + +所有响应都走 `docs/05` §3.3 的统一信封（复用 `app/api/views/envelope.py`）： +列表的 `data` 是纯数组、分页元数据在 `meta`；单对象的 `meta` 只有 `trace_id`。 +""" + +from fastapi import APIRouter, Depends, Path +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.dependencies.auth import build_request_context +from app.api.dependencies.database import get_session +from app.api.dependencies.rate_limit import enforce_rate_limit +from app.api.views.envelope import envelope, list_envelope +from app.core.contracts import RequestContext +from app.service.rbac_query_service import RbacQueryService + +router = APIRouter( + prefix="/api/v1/admin", + tags=["platform-admin"], + dependencies=[Depends(enforce_rate_limit)], +) + +#: 角色码在 `sys_role` 里形如 `customer` / `risk_operator` / `admin`。 +ROLE_CODE = Path(min_length=1, max_length=64, pattern=r"^[a-z0-9_]+$") +#: `sys_user.id` 是 BIGINT UNSIGNED。 +USER_ID = Path(pattern=r"^[0-9]{1,20}$") + + +@router.get("/roles") +async def list_roles( + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """所有角色 + 每个角色的权限数与在用人数。""" + page = await RbacQueryService(session).list_roles(context) + return list_envelope(page, context) + + +@router.get("/roles/{role_code}") +async def get_role( + role_code: str = ROLE_CODE, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """单个角色的详情。 + + 与 `/permissions` 分开是必要的：权限为空的角色（例如刚建好还没授权的）在一个合并 + 接口里很容易被"查不到权限"误判成"角色不存在"。 + """ + data = await RbacQueryService(session).get_role_detail(context, role_code) + return envelope(data, context) + + +@router.get("/roles/{role_code}/permissions") +async def list_role_permissions( + role_code: str = ROLE_CODE, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """该角色的权限清单，按权限码排序，便于与代码里各 Service 的 `require(...)` 对照。""" + page = await RbacQueryService(session).list_role_permissions(context, role_code) + return list_envelope(page, context) + + +@router.get("/users/{user_id}/roles") +async def get_user_roles( + user_id: str = USER_ID, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + """某个用户**实际解析出来**的角色 / 权限 / 数据范围 / 可见客户。 + + 这是四个接口里最有用的一个：它直接回答"这个人为什么 403"—— + 走的是 `IdentityService.resolve`，与请求期完全同一条链路。 + """ + data = await RbacQueryService(session).get_user_identity(context, user_id) + return envelope(data, context) diff --git a/app/main.py b/app/main.py index 167c41d..2c248d0 100644 --- a/app/main.py +++ b/app/main.py @@ -14,6 +14,7 @@ from app.api.controllers.offsite_fund import operation_router as offsite_operati from app.api.controllers.offsite_fund import router as offsite_fund_router from app.api.controllers.promotion_material import router as promotion_material_router from app.api.controllers.public_platform import router as public_platform_router +from app.api.controllers.rbac import router as rbac_router from app.api.controllers.risk import router as risk_router from app.api.middleware import attach_trace_id from app.core.config import get_settings @@ -105,6 +106,7 @@ def create_app() -> FastAPI: application.include_router(agent_runs_router) application.include_router(conversations_router) application.include_router(public_platform_router) + application.include_router(rbac_router) application.include_router(risk_router) application.include_router(offsite_fund_router) application.include_router(offsite_operation_router) diff --git a/app/service/rbac_query_service.py b/app/service/rbac_query_service.py new file mode 100644 index 0000000..c03adb0 --- /dev/null +++ b/app/service/rbac_query_service.py @@ -0,0 +1,202 @@ +"""角色与权限的只读查询（B1：先让管理员**看得见**）。 + +## 为什么先做只读 + +权限变更（提权 / 降权）必须带三条红线：**审计留痕**、**禁止自我提权**、 +**保护内置角色**（否则可能把自己锁在门外）。那是一条需要单独评审的写路径。 +而运维的大半诉求其实是"这个角色到底有哪些权限""这个人为什么 403"——只读就能回答。 + +## 权限码为什么复用 `audit:read`，而不新增 `rbac:read` + +1. `audit:read` 的 `data_scope` 已经是 `all`，且只发给管理员； +2. "谁能访问什么"本身就是**审计材料**——合规检查要看的正是这份清单； +3. 复用是**零数据改动、立刻可用**：新增权限码要先往 `sys_permission` 插行再授权， + 而 `sys_role_permission` 目前由 `seed_test_rbac.py` 以 **DELETE 重建**语义管理 + （见该脚本 95-98 行），为一个只读接口去动权限表不划算。 + 将来要细分时再加 `rbac:read`，与现在不冲突。 + +## 为什么 `user_identity()` 复用 `IdentityService.resolve` + +那是请求进来时走的**同一条链路**：含 `status` 检查、`assigned_at` / `expires_at` +时间窗、`data_scope` 取最高、客户分配（`identity_repository.load_context`）。 +自己再拼一遍 SQL 必然与它漂移，而"这里查出来的权限"与"实际能用的权限"不一致， +比没有这个接口更糟——排障时会被引到错的方向。 +""" + +from __future__ import annotations + +from typing import Any + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.contracts import RequestContext +from app.core.errors import GenericResourceNotFoundError, UnauthorizedAgentError +from app.service.authorization_service import AuthorizationService +from app.service.identity_service import IdentityService + +#: 这三个接口共用的权限闸门，理由见模块文档。 +READ_PERMISSION = "audit:read" + + +class RbacQueryService: + """RBAC 只读查询。不写任何表，也不改任何状态。""" + + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def list_roles(self, context: RequestContext) -> dict[str, Any]: + """列出所有角色，带权限数与在用人数。 + + 角色数量是个位数量级，不分页；返回 `{items, next_cursor, has_more}` 是为了 + 与其它列表接口共用 `list_envelope`（`docs/05` §3.3），而不是真会翻页。 + """ + await AuthorizationService.require(context, READ_PERMISSION) + rows = ( + await self.session.execute( + text( + """ + SELECT r.id, r.role_code, r.role_name, r.status, + (SELECT COUNT(*) FROM sys_role_permission rp + WHERE rp.role_id = r.id) AS permission_count, + (SELECT COUNT(*) FROM sys_user_role ur + WHERE ur.role_id = r.id) AS user_count + FROM sys_role r + ORDER BY r.id + """ + ) + ) + ).mappings().all() + return { + "items": [ + { + "role_id": str(row["id"]), + "role_code": str(row["role_code"]), + "role_name": str(row["role_name"]), + "status": str(row["status"]), + "permission_count": int(row["permission_count"]), + "user_count": int(row["user_count"]), + } + for row in rows + ], + "next_cursor": None, + "has_more": False, + } + + async def list_role_permissions( + self, context: RequestContext, role_code: str + ) -> dict[str, Any]: + """某个角色拥有的全部权限（按权限码排序，便于与代码里 `require(...)` 对照）。""" + await AuthorizationService.require(context, READ_PERMISSION) + role = ( + await self.session.execute( + text( + "SELECT id, role_code, role_name, status FROM sys_role " + "WHERE role_code = :code" + ), + {"code": role_code}, + ) + ).mappings().first() + if role is None: + raise GenericResourceNotFoundError("角色不存在") + rows = ( + await self.session.execute( + text( + """ + SELECT p.permission_code, p.resource, p.action, p.data_scope + FROM sys_role_permission rp + JOIN sys_permission p ON p.id = rp.permission_id + WHERE rp.role_id = :role_id + ORDER BY p.permission_code + """ + ), + {"role_id": int(role["id"])}, + ) + ).mappings().all() + return { + "items": [ + { + "permission_code": str(row["permission_code"]), + "resource": str(row["resource"]), + "action": str(row["action"]), + "data_scope": str(row["data_scope"]), + } + for row in rows + ], + "next_cursor": None, + "has_more": False, + } + + async def get_role_detail( + self, context: RequestContext, role_code: str + ) -> dict[str, Any]: + """角色本身的信息（与权限清单分开：详情接口不应因权限为空就 404）。""" + await AuthorizationService.require(context, READ_PERMISSION) + row = ( + await self.session.execute( + text( + """ + SELECT r.id, r.role_code, r.role_name, r.status, r.created_at, r.updated_at, + (SELECT COUNT(*) FROM sys_role_permission rp + WHERE rp.role_id = r.id) AS permission_count, + (SELECT COUNT(*) FROM sys_user_role ur + WHERE ur.role_id = r.id) AS user_count + FROM sys_role r WHERE r.role_code = :code + """ + ), + {"code": role_code}, + ) + ).mappings().first() + if row is None: + raise GenericResourceNotFoundError("角色不存在") + return { + "role_id": str(row["id"]), + "role_code": str(row["role_code"]), + "role_name": str(row["role_name"]), + "status": str(row["status"]), + "permission_count": int(row["permission_count"]), + "user_count": int(row["user_count"]), + } + + async def get_user_identity( + self, context: RequestContext, user_id: str + ) -> dict[str, Any]: + """某个用户**实际解析出来**的身份：角色、权限、数据范围、可见客户。 + + 直接复用 `IdentityService.resolve`，见模块文档。它抛 `UnauthorizedAgentError` + 表示"账号不存在或未启用"，对这个只读接口来说是**查不到**（404）， + 不是调用方鉴权失败——所以在这里转成 404，避免排障时误以为是权限问题。 + """ + await AuthorizationService.require(context, READ_PERMISSION) + row = ( + await self.session.execute( + text( + "SELECT id, user_no, username, user_type, status " + "FROM sys_user WHERE id = :user_id" + ), + {"user_id": int(user_id)}, + ) + ).mappings().first() + if row is None: + raise GenericResourceNotFoundError("用户不存在") + + try: + resolved = await IdentityService().resolve( + RequestContext(user_id=user_id, trace_id=context.trace_id) + ) + except UnauthorizedAgentError: + # 账号被停用（status != '正常'）——用户存在但拿不到任何权限， + # 如实返回空权限集，比 404 更有助于排障。 + resolved = None + + return { + "user_id": str(row["id"]), + "user_no": str(row["user_no"]), + "username": str(row["username"]), + "user_type": str(row["user_type"]), + "status": str(row["status"]), + "roles": list(resolved.roles) if resolved is not None else [], + "permissions": list(resolved.permissions) if resolved is not None else [], + "data_scope": resolved.data_scope if resolved is not None else None, + "customer_ids": list(resolved.customer_ids) if resolved is not None else [], + } diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index 516512c..989e248 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -1099,10 +1099,25 @@ GET /internal/metrics | A032 | `PUT /api/v1/admin/negative-word-rules/{rule_id}` | `config:write` | 必须 | `200` | 禁止表达 | | A033 | `GET /api/v1/admin/audit-records` | `audit:read` | 否 | `200` | 否 | | A034 | `POST /api/v1/auth/tokens` | 公开（登录前无身份） | 否 | `200` | 登录成功/失败 | +| A035 | `GET /api/v1/admin/roles` | `audit:read` | 否 | `200` | 否 | +| A036 | `GET /api/v1/admin/roles/{role_code}` | `audit:read` | 否 | `200` | 否 | +| A037 | `GET /api/v1/admin/roles/{role_code}/permissions` | `audit:read` | 否 | `200` | 否 | +| A038 | `GET /api/v1/admin/users/{user_id}/roles` | `audit:read` | 否 | `200` | 否 | | O001 | `GET /internal/health/live` | 内网 | 否 | `200` | 否 | | O002 | `GET /internal/health/ready` | 内网 | 否 | `200/503` | 否 | | O003 | `GET /internal/metrics` | 监控系统 | 否 | `200` | 否 | +> **A034 – A038 的两点说明**： +> +> - `POST /api/v1/auth/tokens` 是平台内**唯一的登录入口**（§11 已注明这是"平台不重复实现 +> 签发"的唯一例外；**刷新与注销仍归统一身份认证模块**）。 +> - A035 – A038 是 RBAC 的**只读**查询，供管理员回答"谁能访问什么""这个人为什么 403"。 +> 它们复用 `audit:read` 而**不新增** `rbac:read`：这份清单本身就是审计材料，且复用是 +> 零数据改动、立刻可用（新增权限码得先改 `sys_permission`，而它目前由 +> `seed_test_rbac.py` 以 DELETE 重建语义管理）。 +> **权限变更（提权 / 降权）尚无接口** —— 那条写路径必须带三条红线 +> （审计留痕、禁止自我提权、保护内置角色），需要单独评审，不是遗漏。 + 业务域接口 `/customer-service/handover-tickets/**`、`/advisory-plans/**`、`/sim-orders/**`、`/risk-scans/**` 和 `/risk-alerts/**` 的具体方法、请求体、领域状态机和错误码分别由对应业务文档登记；它们仍必须遵守本文第 3-5、11 和 12 节。 ## 20. 变更流程 diff --git a/tests/integration/test_rbac_read_mysql.py b/tests/integration/test_rbac_read_mysql.py new file mode 100644 index 0000000..66fe99d --- /dev/null +++ b/tests/integration/test_rbac_read_mysql.py @@ -0,0 +1,184 @@ +"""RBAC 只读接口的端到端验证（真实 MySQL + 真实 HTTP 栈）。 + +重点不在"能不能查出数据"，而在三件事： + +1. **门槛对不对** —— 这三个接口暴露的是"谁能访问什么"，属管理员级只读， + 客户令牌必须 403。权限码复用 `audit:read`（理由见 `rbac_query_service` 的模块文档）。 +2. **两个接口说的是不是同一件事** —— `/admin/users/{id}/roles` 走 + `IdentityService.resolve`，而登录响应的 `roles` 也走同一条链路；两者必须一致， + 否则排障时会被引到错方向。 +3. **信封形状** —— 列表的 `data` 是纯数组、分页元数据在 `meta`（`docs/05` §3.3）。 + +前置：`tools/seed_test_rbac.py` + `tools/set_user_password.py`。 +""" + +from typing import Any + +import httpx +import pytest + +from app.main import app + +pytestmark = pytest.mark.integration + +LOGIN_PATH = "/api/v1/auth/tokens" + + +class _AlwaysAllowBackend: + """恒放行，隔离跨用例的限流计数累积（同 test_auth_login_mysql.py）。""" + + async def increment(self, key: str, window_seconds: int) -> tuple[int, int] | None: + del key, window_seconds + return (1, 0) + + +@pytest.fixture(autouse=True) +def _replace_rate_limit_backend(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + "app.api.dependencies.rate_limit.get_counter_backend", + lambda: _AlwaysAllowBackend(), + ) + + +def client() -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=30 + ) + + +async def token_for(http: httpx.AsyncClient, username: str, password: str) -> str: + response = await http.post( + LOGIN_PATH, json={"username": username, "password": password} + ) + assert response.status_code == 200, response.text + return str(response.json()["data"]["access_token"]) + + +def auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +@pytest.mark.asyncio +async def test_admin_can_list_roles_with_shape() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + response = await http.get("/api/v1/admin/roles", headers=auth(token)) + + assert response.status_code == 200, response.text + body: dict[str, Any] = response.json() + # §3.3：列表的 data 是纯数组，分页元数据在 meta。 + assert isinstance(body["data"], list) + assert set(body["meta"]) == {"trace_id", "next_cursor", "has_more"} + codes = {row["role_code"] for row in body["data"]} + assert {"customer", "risk_operator", "admin"} <= codes, f"角色清单缺内置角色：{codes}" + for row in body["data"]: + assert isinstance(row["permission_count"], int) + assert isinstance(row["user_count"], int) + assert row["status"] + + +@pytest.mark.asyncio +async def test_role_permissions_are_listed_and_sorted() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + response = await http.get( + "/api/v1/admin/roles/customer/permissions", headers=auth(token) + ) + + assert response.status_code == 200, response.text + body = response.json() + assert isinstance(body["data"], list) and body["data"], "客户角色不该没有任何权限" + codes = [row["permission_code"] for row in body["data"]] + assert codes == sorted(codes), "权限清单应按权限码排序，便于与代码里的 require() 对照" + assert {"agent:run", "suitability:read"} <= set(codes) + + +@pytest.mark.asyncio +async def test_role_detail_is_separate_from_permissions() -> None: + """权限为空的角色也必须能查到详情，不能被当成"角色不存在"。""" + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + detail = await http.get("/api/v1/admin/roles/operator", headers=auth(token)) + + assert detail.status_code == 200, detail.text + assert detail.json()["data"]["role_code"] == "operator" + + +@pytest.mark.asyncio +async def test_customer_token_is_denied() -> None: + """门槛验证：客户不能读"谁能访问什么"。""" + async with client() as http: + token = await token_for(http, "cust_t", "123456") + for path in ( + "/api/v1/admin/roles", + "/api/v1/admin/roles/customer/permissions", + "/api/v1/admin/users/9001/roles", + ): + response = await http.get(path, headers=auth(token)) + assert response.status_code == 403, f"{path} 不该让客户访问（{response.status_code}）" + assert response.json()["error"]["code"] == "AGENT_PERMISSION_DENIED" + + +@pytest.mark.asyncio +async def test_anonymous_is_unauthorized() -> None: + async with client() as http: + response = await http.get("/api/v1/admin/roles") + + assert response.status_code == 401 + assert response.json()["error"]["code"] == "AUTHENTICATION_REQUIRED" + + +@pytest.mark.asyncio +async def test_unknown_role_and_user_are_not_found() -> None: + async with client() as http: + token = await token_for(http, "admin_t", "88888888") + role = await http.get("/api/v1/admin/roles/no_such_role", headers=auth(token)) + user = await http.get("/api/v1/admin/users/99999999/roles", headers=auth(token)) + + assert role.status_code == 404 + assert user.status_code == 404 + + +@pytest.mark.asyncio +async def test_user_identity_agrees_with_login_response() -> None: + """交叉验证：两个接口必须说同一件事。 + + 登录响应的 `roles` 与 `/admin/users/{id}/roles` 的 `roles` 都来自 + `IdentityService.resolve`；若哪天有人给其中一条路径加了缓存或另写一份 SQL， + 这个断言会立刻发现。 + """ + async with client() as http: + login = await http.post( + LOGIN_PATH, json={"username": "risk_t", "password": "666666"} + ) + assert login.status_code == 200, login.text + login_data = login.json()["data"] + + admin_token = await token_for(http, "admin_t", "88888888") + identity = await http.get( + f"/api/v1/admin/users/{login_data['user_id']}/roles", headers=auth(admin_token) + ) + + assert identity.status_code == 200, identity.text + data = identity.json()["data"] + assert data["roles"] == login_data["roles"], "两个接口解析出的角色不一致" + assert data["data_scope"] == login_data["data_scope"], "两个接口解析出的数据范围不一致" + assert data["username"] == "risk_t" + assert "audit:read" in data["permissions"] + + +@pytest.mark.asyncio +async def test_deactivated_account_reports_empty_permissions_not_404() -> None: + """被停用的账号：**存在**但没有权限。返回空权限集比 404 更有助于排障。""" + async with client() as http: + admin_token = await token_for(http, "admin_t", "88888888") + # 9004(review_t) 是种子里的账号：仅绑了角色但没设密码，且此处不依赖密码。 + response = await http.get( + "/api/v1/admin/users/9004/roles", headers=auth(admin_token) + ) + + assert response.status_code == 200, response.text + data = response.json()["data"] + assert data["username"] == "review_t" + # 它没有 sys_user_role 绑定，因此 roles 为空 —— 但不该是 404。 + assert data["roles"] == []