张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
66 lines
3.3 KiB
Python
66 lines
3.3 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from app.core.contracts import RequestContext
|
|
from app.core.errors import UnauthorizedAgentError
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.repository.identity_repository import IdentityRepository
|
|
|
|
|
|
@pytest.mark.integration
|
|
async def test_roles_revocation_expiry_account_status_and_assignments():
|
|
user_id = uuid4().int % 10**15 + 10**15
|
|
role_id, permission_id = user_id + 1, user_id + 2
|
|
async with SessionFactory() as session:
|
|
try:
|
|
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,:name,:name,'test-only','员工','未申请','未开户','正常',UTC_TIMESTAMP(),
|
|
UTC_TIMESTAMP())
|
|
"""), {"id": user_id, "name": f"auth-{user_id}"})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_role VALUES
|
|
(:id,:code,'test','active',UTC_TIMESTAMP(),UTC_TIMESTAMP())
|
|
"""), {"id": role_id, "code": f"role-{role_id}"})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_user_role (user_id,role_id,assigned_at)
|
|
VALUES (:uid,:rid,UTC_TIMESTAMP()-INTERVAL 1 SECOND)
|
|
"""), {"uid": user_id, "rid": role_id})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_permission VALUES
|
|
(:id,:code,'test','read','own_customers',NULL,UTC_TIMESTAMP(),UTC_TIMESTAMP())
|
|
"""), {"id": permission_id, "code": f"test:{permission_id}"})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_role_permission(role_id,permission_id,created_at)
|
|
VALUES (:rid,:pid,UTC_TIMESTAMP())
|
|
"""), {"rid": role_id, "pid": permission_id})
|
|
await session.execute(text("""
|
|
INSERT INTO sys_customer_assignment VALUES
|
|
(:id,:customer,:employee,:role,UTC_TIMESTAMP()-INTERVAL 1 SECOND,NULL)
|
|
"""), {"id": user_id, "customer": user_id + 100, "employee": user_id,
|
|
"role": f"test-{user_id}"})
|
|
identity = RequestContext(user_id=str(user_id), trace_id="test", roles=("admin",),
|
|
permissions=("forged",), portal="forged")
|
|
repo = IdentityRepository(session)
|
|
current = await repo.load_context(identity)
|
|
assert current.roles == (f"role-{role_id}",)
|
|
assert current.permissions == (f"test:{permission_id}",)
|
|
assert current.permission_scopes[f"test:{permission_id}"] == "own_customers"
|
|
assert current.customer_ids == (str(user_id + 100),)
|
|
assert current.portal == "api"
|
|
await session.execute(text("""
|
|
UPDATE sys_user_role SET expires_at=UTC_TIMESTAMP()-INTERVAL 1 SECOND
|
|
WHERE user_id=:id
|
|
"""), {"id": user_id})
|
|
assert (await repo.load_context(identity)).roles == ()
|
|
await session.execute(text("UPDATE sys_user SET status='禁用' WHERE id=:id"),
|
|
{"id": user_id})
|
|
with pytest.raises(UnauthorizedAgentError):
|
|
await repo.load_context(identity)
|
|
finally:
|
|
await session.rollback()
|