Simon
613 lines
22 KiB
Python
613 lines
22 KiB
Python
# =============================================================================
|
|
# 企微IT智能服务台 — 管理后台账号密码+OTP登录测试
|
|
# =============================================================================
|
|
# 覆盖范围:
|
|
# 1. 超级管理员初始化逻辑 (init_super_admin)
|
|
# 2. 账号密码登录流程 (/agents/login)
|
|
# 3. 管理员 CRUD API (/admin/users) — 注意:nginx已剥离/api前缀
|
|
# 4. 密码验证 (bcrypt)
|
|
# 5. OTP/MFA 二次验证
|
|
# =============================================================================
|
|
|
|
import os
|
|
import pytest
|
|
import pytest_asyncio
|
|
import pyotp
|
|
import bcrypt
|
|
from sqlalchemy import select
|
|
|
|
from app.models.agent import Agent
|
|
from app.models.role import Role
|
|
from app.models.user_role import UserRole
|
|
from app.services.admin_user_service import AdminUserService, init_super_admin
|
|
from tests.conftest import create_test_agent
|
|
|
|
|
|
# -----------------------------------------------------------------------------
|
|
# 辅助函数
|
|
# -----------------------------------------------------------------------------
|
|
async def _seed_admin_role(db_session, employee_id: str, role_name: str = "admin") -> str:
|
|
"""为用户分配指定角色."""
|
|
stmt = select(Role).where(Role.name == role_name)
|
|
role = (await db_session.execute(stmt)).scalars().first()
|
|
if not role:
|
|
role = Role(
|
|
id=str(__import__("uuid").uuid4()),
|
|
name=role_name,
|
|
display_name={"admin": "管理员", "super_admin": "超级管理员"}.get(role_name, role_name),
|
|
is_default=False,
|
|
permissions=[],
|
|
)
|
|
db_session.add(role)
|
|
await db_session.flush()
|
|
|
|
# 检查关联是否已存在
|
|
stmt = select(UserRole).where(
|
|
UserRole.employee_id == employee_id,
|
|
UserRole.role_id == role.id,
|
|
)
|
|
existing = (await db_session.execute(stmt)).scalars().first()
|
|
if not existing:
|
|
user_role = UserRole(
|
|
id=str(__import__("uuid").uuid4()),
|
|
employee_id=employee_id,
|
|
role_id=role.id,
|
|
source="manual",
|
|
assigned_at=__import__("datetime").datetime.now(),
|
|
)
|
|
db_session.add(user_role)
|
|
await db_session.flush()
|
|
|
|
return role.id
|
|
|
|
|
|
def _bearer(token: str) -> dict:
|
|
"""构造 Authorization header."""
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def _login_and_get_token(client, user_id: str, name: str, password: str = None, otp_code: str = None) -> dict:
|
|
"""调用 /agents/login 拿 token.
|
|
|
|
Returns:
|
|
dict: 包含 token, require_otp 等字段的响应数据
|
|
"""
|
|
json_body = {"user_id": user_id, "name": name}
|
|
if password:
|
|
json_body["password"] = password
|
|
if otp_code:
|
|
json_body["otp_code"] = otp_code
|
|
|
|
response = await client.post("/agents/login", json=json_body)
|
|
assert response.status_code == 200, f"登录失败: {response.text}"
|
|
body = response.json()
|
|
return body
|
|
|
|
|
|
# =============================================================================
|
|
# 1. 超级管理员初始化测试
|
|
# =============================================================================
|
|
class TestSuperAdminInit:
|
|
"""测试 init_super_admin 函数"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_super_admin_no_env_vars(self, db_session):
|
|
"""未配置环境变量时，返回 None，不创建用户"""
|
|
# 确保环境变量未设置
|
|
with pytest.MonkeyPatch.context() as mp:
|
|
mp.delenv("ADMIN_USERNAME", raising=False)
|
|
mp.delenv("ADMIN_PASSWORD", raising=False)
|
|
mp.delenv("ADMIN_NAME", raising=False)
|
|
|
|
result = await init_super_admin(db_session)
|
|
assert result is None
|
|
|
|
# 确认没有创建任何管理员
|
|
stmt = select(Agent).where(Agent.role == "super_admin")
|
|
result = await db_session.execute(stmt)
|
|
assert result.scalars().first() is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_super_admin_with_env_vars(self, db_session):
|
|
"""配置环境变量时，创建超级管理员"""
|
|
with pytest.MonkeyPatch.context() as mp:
|
|
mp.setenv("ADMIN_USERNAME", "superadmin")
|
|
mp.setenv("ADMIN_PASSWORD", "superpass123")
|
|
mp.setenv("ADMIN_NAME", "超级管理员")
|
|
|
|
result = await init_super_admin(db_session)
|
|
assert result is not None
|
|
assert result.user_id == "superadmin"
|
|
assert result.name == "超级管理员"
|
|
assert result.role == "super_admin"
|
|
|
|
# 验证密码已哈希存储
|
|
assert result.password_hash is not None
|
|
assert bcrypt.checkpw("superpass123".encode("utf-8"), result.password_hash.encode("utf-8"))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_init_super_admin_already_exists(self, db_session):
|
|
"""超级管理员已存在时，跳过创建"""
|
|
# 先创建一个
|
|
agent = create_test_agent(user_id="superadmin", name="超级管理员")
|
|
agent.role = "super_admin"
|
|
agent.password_hash = bcrypt.hashpw("oldpass".encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
db_session.add(agent)
|
|
await db_session.flush()
|
|
|
|
with pytest.MonkeyPatch.context() as mp:
|
|
mp.setenv("ADMIN_USERNAME", "superadmin")
|
|
mp.setenv("ADMIN_PASSWORD", "newpass123")
|
|
|
|
result = await init_super_admin(db_session)
|
|
assert result is not None
|
|
assert result.user_id == "superadmin"
|
|
# 密码应该是原来的，不应该被覆盖
|
|
assert bcrypt.checkpw("oldpass".encode("utf-8"), result.password_hash.encode("utf-8"))
|
|
|
|
|
|
# =============================================================================
|
|
# 2. AdminUserService 单元测试
|
|
# =============================================================================
|
|
class TestAdminUserService:
|
|
"""AdminUserService 静态方法直接测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_admin_user(self, db_session):
|
|
"""创建管理员用户"""
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="admin001",
|
|
name="管理员1",
|
|
role="admin",
|
|
password="test123456"
|
|
)
|
|
|
|
assert agent.user_id == "admin001"
|
|
assert agent.name == "管理员1"
|
|
assert agent.role == "admin"
|
|
assert bcrypt.checkpw("test123456".encode("utf-8"), agent.password_hash.encode("utf-8"))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_admin_user_duplicate(self, db_session):
|
|
"""重复创建管理员应抛出异常"""
|
|
service = AdminUserService(db_session)
|
|
await service.create_admin_user(
|
|
user_id="admin001",
|
|
name="管理员1",
|
|
role="admin",
|
|
password="test123456"
|
|
)
|
|
|
|
# 重复创建应抛出异常
|
|
from app.utils.error_codes import ErrorCode
|
|
from app.utils.response import AppException
|
|
with pytest.raises(AppException) as exc_info:
|
|
await service.create_admin_user(
|
|
user_id="admin001",
|
|
name="管理员1",
|
|
role="admin",
|
|
password="test123456"
|
|
)
|
|
assert "已存在" in str(exc_info.value.message)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_password_correct(self, db_session):
|
|
"""密码验证 - 正确密码"""
|
|
service = AdminUserService(db_session)
|
|
await service.create_admin_user(
|
|
user_id="admin002",
|
|
name="管理员2",
|
|
role="admin",
|
|
password="correctpassword"
|
|
)
|
|
|
|
agent = await service.verify_password("admin002", "correctpassword")
|
|
assert agent is not None
|
|
assert agent.user_id == "admin002"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_password_wrong(self, db_session):
|
|
"""密码验证 - 错误密码"""
|
|
service = AdminUserService(db_session)
|
|
await service.create_admin_user(
|
|
user_id="admin003",
|
|
name="管理员3",
|
|
role="admin",
|
|
password="correctpassword"
|
|
)
|
|
|
|
agent = await service.verify_password("admin003", "wrongpassword")
|
|
assert agent is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_password_nonexistent_user(self, db_session):
|
|
"""密码验证 - 不存在的用户"""
|
|
service = AdminUserService(db_session)
|
|
agent = await service.verify_password("nonexistent", "anypassword")
|
|
assert agent is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_password(self, db_session):
|
|
"""重置密码"""
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="admin004",
|
|
name="管理员4",
|
|
role="admin",
|
|
password="oldpassword"
|
|
)
|
|
|
|
updated = await service.reset_password(agent.id, "newpassword")
|
|
assert bcrypt.checkpw("newpassword".encode("utf-8"), updated.password_hash.encode("utf-8"))
|
|
assert not bcrypt.checkpw("oldpassword".encode("utf-8"), updated.password_hash.encode("utf-8"))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_admin_user(self, db_session):
|
|
"""删除管理员"""
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="admin005",
|
|
name="管理员5",
|
|
role="admin",
|
|
password="test123"
|
|
)
|
|
|
|
result = await service.delete_admin_user(agent.id)
|
|
assert result is True
|
|
|
|
# 验证已删除
|
|
deleted = await service.get_user_by_id(agent.id)
|
|
assert deleted is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_super_admin_forbidden(self, db_session):
|
|
"""删除超级管理员应被拒绝"""
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="superadmin",
|
|
name="超级管理员",
|
|
role="super_admin",
|
|
password="test123"
|
|
)
|
|
|
|
from app.utils.error_codes import ErrorCode
|
|
from app.utils.response import AppException
|
|
with pytest.raises(AppException) as exc_info:
|
|
await service.delete_admin_user(agent.id)
|
|
assert "超级管理员" in str(exc_info.value.message)
|
|
|
|
|
|
# =============================================================================
|
|
# 3. 管理员 CRUD API 测试
|
|
# =============================================================================
|
|
class TestAdminUserAPI:
|
|
"""管理员用户 CRUD API 测试
|
|
|
|
注意: nginx 配置会将 /api 前缀剥离,所以实际路径是 /admin/users 而非 /api/admin/users
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_admin_users(self, client, db_session):
|
|
"""GET /admin/users - 获取管理员列表"""
|
|
# 创建测试管理员
|
|
service = AdminUserService(db_session)
|
|
await service.create_admin_user("admin_list_1", "管理员A", "admin", "pass123")
|
|
await service.create_admin_user("admin_list_2", "管理员B", "admin", "pass456")
|
|
await db_session.commit()
|
|
|
|
# 创建管理员用户并分配角色
|
|
admin_agent = create_test_agent(user_id="test_admin_user", name="测试管理员")
|
|
admin_agent.role = "admin"
|
|
db_session.add(admin_agent)
|
|
await db_session.flush()
|
|
await _seed_admin_role(db_session, "test_admin_user", "admin")
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "test_admin_user", "测试管理员")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 调用 API (无 /api 前缀)
|
|
response = await client.get("/admin/users", headers=_bearer(token))
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["code"] == 0
|
|
assert body["data"]["total"] >= 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_admin_user_api(self, client, db_session):
|
|
"""POST /admin/users - 创建管理员"""
|
|
# 创建超级管理员
|
|
super_agent = create_test_agent(user_id="test_super_admin", name="测试超级管理员")
|
|
super_agent.role = "super_admin"
|
|
db_session.add(super_agent)
|
|
await db_session.flush()
|
|
await _seed_admin_role(db_session, "test_super_admin", "super_admin")
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "test_super_admin", "测试超级管理员")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 创建管理员
|
|
response = await client.post(
|
|
"/admin/users",
|
|
headers=_bearer(token),
|
|
json={
|
|
"user_id": "new_admin",
|
|
"name": "新管理员",
|
|
"role": "admin",
|
|
"password": "newpass123"
|
|
}
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["code"] == 0
|
|
assert body["data"]["user_id"] == "new_admin"
|
|
assert body["data"]["role"] == "admin"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_admin_user(self, client, db_session):
|
|
"""GET /admin/users/{id} - 获取管理员详情"""
|
|
# 创建管理员
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user("admin_detail", "管理员详情", "admin", "pass123")
|
|
await db_session.commit()
|
|
|
|
# 查询用户并分配角色
|
|
stmt = select(Agent).where(Agent.user_id == "admin_detail")
|
|
result = await db_session.execute(stmt)
|
|
admin_agent = result.scalars().first()
|
|
|
|
# 分配 admin 角色
|
|
await _seed_admin_role(db_session, "admin_detail", "admin")
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "admin_detail", "管理员详情")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 获取详情
|
|
response = await client.get(f"/admin/users/{admin_agent.id}", headers=_bearer(token))
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["code"] == 0
|
|
assert body["data"]["user_id"] == "admin_detail"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_admin_user(self, client, db_session):
|
|
"""PUT /admin/users/{id} - 更新管理员"""
|
|
# 创建管理员
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user("admin_update", "管理员更新", "admin", "pass123")
|
|
await db_session.commit()
|
|
|
|
# 分配角色
|
|
await _seed_admin_role(db_session, "admin_update", "admin")
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "admin_update", "管理员更新")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 更新管理员
|
|
response = await client.put(
|
|
f"/admin/users/{agent.id}",
|
|
headers=_bearer(token),
|
|
json={"name": "新名字", "is_active": True}
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["code"] == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_admin_user_api(self, client, db_session):
|
|
"""DELETE /admin/users/{id} - 删除管理员"""
|
|
# 创建超级管理员
|
|
super_agent = create_test_agent(user_id="test_super_delete", name="测试超级管理员")
|
|
super_agent.role = "super_admin"
|
|
db_session.add(super_agent)
|
|
await db_session.flush()
|
|
await _seed_admin_role(db_session, "test_super_delete", "super_admin")
|
|
|
|
# 创建待删除的管理员
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user("admin_to_delete", "待删除管理员", "admin", "pass123")
|
|
await db_session.commit()
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "test_super_delete", "测试超级管理员")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 删除管理员
|
|
response = await client.delete(f"/admin/users/{agent.id}", headers=_bearer(token))
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["code"] == 0
|
|
|
|
|
|
# =============================================================================
|
|
# 4. 账号密码+OTP登录测试
|
|
# =============================================================================
|
|
class TestPasswordOTPLogin:
|
|
"""账号密码 + OTP 登录流程测试
|
|
|
|
登录流程说明:
|
|
1. 优先尝试企微通讯录验证
|
|
2. 企微不可达时，已注册坐席可降级登录（需验证本地密码）
|
|
3. 启用MFA后需要OTP验证
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_without_password(self, client, db_session):
|
|
"""登录 - 无密码的新坐席（企微验证通过）"""
|
|
# 创建管理员但不设置密码
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="login_test_user",
|
|
name="登录测试用户",
|
|
role="admin"
|
|
# 不设置 password
|
|
)
|
|
await db_session.commit()
|
|
|
|
# 登录（企微验证通过）
|
|
response = await _login_and_get_token(
|
|
client,
|
|
user_id="login_test_user",
|
|
name="登录测试用户"
|
|
)
|
|
|
|
body = response
|
|
assert body["code"] == 0
|
|
assert "token" in body["data"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_require_otp_when_mfa_enabled(self, client, db_session):
|
|
"""登录 - MFA 启用时需要 OTP 验证"""
|
|
# 创建管理员并启用 MFA
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="mfa_user",
|
|
name="MFA用户",
|
|
role="admin",
|
|
password="testpassword123"
|
|
)
|
|
# 模拟已绑定 MFA
|
|
secret = pyotp.random_base32()
|
|
agent.mfa_secret = secret
|
|
agent.mfa_enabled = True
|
|
await db_session.commit()
|
|
|
|
# 登录但不提供 OTP
|
|
response = await _login_and_get_token(
|
|
client,
|
|
user_id="mfa_user",
|
|
name="MFA用户",
|
|
password="testpassword123"
|
|
)
|
|
|
|
body = response
|
|
assert body["code"] == 0
|
|
assert body["data"]["require_otp"] is True
|
|
assert "token" not in body["data"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_with_correct_otp(self, client, db_session):
|
|
"""登录 - 提供正确的 OTP"""
|
|
# 创建管理员并启用 MFA
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="otp_user",
|
|
name="OTP用户",
|
|
role="admin",
|
|
password="testpassword123"
|
|
)
|
|
# 模拟已绑定 MFA
|
|
secret = pyotp.random_base32()
|
|
agent.mfa_secret = secret
|
|
agent.mfa_enabled = True
|
|
await db_session.commit()
|
|
|
|
# 生成当前有效的 OTP
|
|
totp = pyotp.TOTP(secret)
|
|
otp_code = totp.now()
|
|
|
|
# 登录并提供 OTP
|
|
response = await _login_and_get_token(
|
|
client,
|
|
user_id="otp_user",
|
|
name="OTP用户",
|
|
password="testpassword123",
|
|
otp_code=otp_code
|
|
)
|
|
|
|
body = response
|
|
assert body["code"] == 0
|
|
assert "token" in body["data"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_with_wrong_otp(self, client, db_session):
|
|
"""登录 - 提供错误的 OTP"""
|
|
# 创建管理员并启用 MFA
|
|
service = AdminUserService(db_session)
|
|
agent = await service.create_admin_user(
|
|
user_id="wrong_otp_user",
|
|
name="错误OTP用户",
|
|
role="admin",
|
|
password="testpassword123"
|
|
)
|
|
# 模拟已绑定 MFA
|
|
secret = pyotp.random_base32()
|
|
agent.mfa_secret = secret
|
|
agent.mfa_enabled = True
|
|
await db_session.commit()
|
|
|
|
# 使用错误的 OTP 登录
|
|
response = await _login_and_get_token(
|
|
client,
|
|
user_id="wrong_otp_user",
|
|
name="错误OTP用户",
|
|
password="testpassword123",
|
|
otp_code="000000"
|
|
)
|
|
|
|
# 应该返回业务错误
|
|
assert response["code"] != 0
|
|
|
|
|
|
# =============================================================================
|
|
# 5. 权限控制测试
|
|
# =============================================================================
|
|
class TestAdminPermission:
|
|
"""管理员权限控制测试"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_admin_requires_super_admin(self, client, db_session):
|
|
"""创建管理员需要 super_admin 角色"""
|
|
# 创建普通管理员
|
|
normal_admin = create_test_agent(user_id="normal_admin_perm", name="普通管理员")
|
|
normal_admin.role = "admin"
|
|
db_session.add(normal_admin)
|
|
await db_session.flush()
|
|
await _seed_admin_role(db_session, "normal_admin_perm", "admin")
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "normal_admin_perm", "普通管理员")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 尝试创建管理员 (无 /api 前缀)
|
|
response = await client.post(
|
|
"/admin/users",
|
|
headers=_bearer(token),
|
|
json={
|
|
"user_id": "should_fail",
|
|
"name": "应该失败",
|
|
"role": "admin"
|
|
}
|
|
)
|
|
# 应该返回非0业务码
|
|
body = response.json()
|
|
assert body["code"] != 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_admin_requires_super_admin(self, client, db_session):
|
|
"""删除管理员需要 super_admin 角色"""
|
|
# 创建普通管理员
|
|
normal_admin = create_test_agent(user_id="normal_admin_del", name="普通管理员")
|
|
normal_admin.role = "admin"
|
|
db_session.add(normal_admin)
|
|
await db_session.flush()
|
|
await _seed_admin_role(db_session, "normal_admin_del", "admin")
|
|
|
|
# 创建待删除的管理员
|
|
service = AdminUserService(db_session)
|
|
to_delete = await service.create_admin_user("target_admin", "目标管理员", "admin", "pass123")
|
|
await db_session.commit()
|
|
|
|
# 登录获取 token
|
|
login_resp = await _login_and_get_token(client, "normal_admin_del", "普通管理员")
|
|
token = login_resp["data"]["token"]
|
|
|
|
# 尝试删除管理员 (无 /api 前缀)
|
|
response = await client.delete(f"/admin/users/{to_delete.id}", headers=_bearer(token))
|
|
body = response.json()
|
|
assert body["code"] != 0
|