"""添加一个可登录的测试账号（用户 + 角色 + 密码），并**验证它真的能拿到权限**。 ## 为什么需要它 `sys_user` / `sys_user_role` / `sys_role` 这些表**没有 ORM 模型**（全项目用裸 SQL 访问， 见 `tools/seed_test_rbac.py`），手写 INSERT 要凑齐 10 个字段、还要自己算密码哈希。 更要紧的是这里有个**静默陷阱**： MySQL 的 DATETIME(0) 会把微秒**四舍五入到秒**。若 `sys_user_role.assigned_at` 用"当前时间"写入，进位后可能落在未来，而授权校验是 `assigned_at <= now` —— 于是刚建好的账号**一个角色都拿不到**，表现为 `roles=()`，**不报错也不失败**。 `seed_test_rbac.py` 已经踩过一次（它的注释里写着）。本脚本统一把 `assigned_at` 往前留 5 秒，并在最后**用 `IdentityService.resolve` 打印真实解析结果**而不是 "插入成功" —— 后者根本不能说明这个账号能用。 ## 用法 python tools/create_test_user.py --list python tools/create_test_user.py --id 9010 --username test_cust \\ --role customer --password abc12345 重复执行同一个 `--id` 是**覆盖**语义：更新用户名 / 密码 / 角色，不会产生重复行。 """ from __future__ import annotations import argparse import asyncio import sys from datetime import UTC, datetime, timedelta from sqlalchemy import text from app.core.contracts import RequestContext from app.infrastructure.db import SessionFactory from app.service.auth_service import hash_password from app.service.identity_service import IdentityService if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] #: 库里现成的角色。新用户复用它们。 #: 要引入**新角色**得同时定义它的权限集合（`sys_role_permission`）—— #: `advisor` 就是由 `tools/grant_advisor_role.py` 建立的。 ROLE_IDS: dict[str, int] = { "customer": 9001, "risk_operator": 9002, "admin": 9003, "advisor": 9004, } #: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。 ROLE_USER_TYPE: dict[str, str] = { "customer": "customer", "risk_operator": "employee", "admin": "employee", "advisor": "employee", } #: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。 FUND_ACCOUNT_STATUS: dict[str, str] = { "customer": "已开户", "employee": "closed", } #: `assigned_at` 往前留的秒数，见模块文档里的静默陷阱。 ASSIGN_BACKDATE_SECONDS = 5 async def list_users() -> None: """列出所有账号、它的角色与密码状态。""" async with SessionFactory() as session: rows = ( await session.execute( text( """ SELECT u.id, u.username, u.user_type, u.status, u.password_hash, GROUP_CONCAT(r.role_code ORDER BY r.role_code) AS roles FROM sys_user u LEFT JOIN sys_user_role ur ON ur.user_id = u.id LEFT JOIN sys_role r ON r.id = ur.role_id GROUP BY u.id, u.username, u.user_type, u.status, u.password_hash ORDER BY u.id """ ) ) ).mappings().all() print(f"{'id':<8}{'username':<18}{'user_type':<12}{'status':<8}{'roles':<28}密码") for row in rows: has_password = str(row["password_hash"] or "").startswith(("$2a$", "$2b$", "$2y$")) print( f"{row['id']:<8}{str(row['username']):<18}{str(row['user_type']):<12}" f"{str(row['status']):<8}{str(row['roles'] or '(无角色)') or '(无角色)':<28}" f"{'已设' if has_password else '占位符，无法登录'}" ) async def upsert_user( *, user_id: int, username: str, role: str, password: str ) -> int: """建/更新账号并绑定角色，最后验证权限能解析出来。""" role_id = ROLE_IDS[role] user_type = ROLE_USER_TYPE[role] now = datetime.now(UTC).replace(tzinfo=None) assigned_at = now - timedelta(seconds=ASSIGN_BACKDATE_SECONDS) async with SessionFactory() as session, session.begin(): role_exists = await session.scalar( text("SELECT id FROM sys_role WHERE id = :role_id"), {"role_id": role_id} ) if role_exists is None: print(f"[失败] 角色 {role}（id={role_id}）不存在，先跑 tools/seed_test_rbac.py") return 1 # 覆盖语义：同一个 id 重跑不会堆出第二行。 await session.execute( text( """ INSERT INTO sys_user (id, user_no, username, password_hash, user_type, professional_investor_status, fund_account_status, status, created_at, updated_at) VALUES (:id, :user_no, :username, :password_hash, :user_type, 'none', :fund_status, '正常', :now, :now) AS new ON DUPLICATE KEY UPDATE username = new.username, password_hash = new.password_hash, user_type = new.user_type, fund_account_status = new.fund_account_status, status = '正常', updated_at = new.updated_at """ ), { "id": user_id, "user_no": f"T-{username.upper()[:20]}", "username": username, "password_hash": hash_password(password), "user_type": user_type, "fund_status": FUND_ACCOUNT_STATUS[user_type], "now": now, }, ) # 角色绑定先清后插。表上其实**有** `uk_sys_user_role (user_id, role_id)` 唯一键 # （`tools/generate_baseline_sql.py:128`），所以直接 INSERT 不会堆重复行 —— # 先清的理由是**支持改角色**：不清的话换角色会在表里留下两条绑定， # `IdentityService` 解析出的 roles 就变成两个（既是 customer 又是 admin）。 await session.execute( text("DELETE FROM sys_user_role WHERE user_id = :user_id"), {"user_id": user_id} ) await session.execute( text( "INSERT INTO sys_user_role (user_id, role_id, assigned_at)" " VALUES (:user_id, :role_id, :assigned_at)" ), {"user_id": user_id, "role_id": role_id, "assigned_at": assigned_at}, ) print(f"[OK] id={user_id} username={username} role={role} 已写入") return await verify(user_id, username, role) async def verify(user_id: int, username: str, role: str) -> int: """用真实链路解析身份 —— 这是唯一能证明"这个账号能用的"方式。""" context = await IdentityService().resolve( RequestContext(user_id=str(user_id), trace_id="create-test-user") ) print(f" 解析结果：roles={context.roles} data_scope={context.data_scope}") print(f" 权限 {len(context.permissions)} 项") if role not in context.roles: print( "[失败] 角色没有解析出来。最可能的原因是 assigned_at 落在了未来" "（DATETIME(0) 的毫秒舍入），请重跑本脚本。" ) return 1 print(f"[OK] 可以用它登录：{{\"username\": \"{username}\", \"password\": \"<你刚设的>\"}}") return 0 async def main() -> int: parser = argparse.ArgumentParser(description="添加可登录的测试账号") parser.add_argument("--list", action="store_true", help="列出所有账号与角色") parser.add_argument("--id", type=int, help="用户 id（9001-9003 已被演示账号占用）") parser.add_argument("--username", help="登录用户名") parser.add_argument("--role", choices=sorted(ROLE_IDS), help="角色") parser.add_argument("--password", help="登录密码（仅限演示环境）") args = parser.parse_args() if args.list: await list_users() return 0 missing = [ name for name, value in ( ("--id", args.id), ("--username", args.username), ("--role", args.role), ("--password", args.password), ) if value is None ] if missing: print(f"[失败] 缺少参数：{' '.join(missing)}（或直接用 --list 看现有账号）") return 1 return await upsert_user( user_id=args.id, username=args.username, role=args.role, password=args.password, ) if __name__ == "__main__": sys.exit(asyncio.run(main()))