logan
be2d6f3894
后端新增 4 个 LDAP 管理端点: - POST /ldap-config/create-ou: 创建组织单元 - POST /ldap-config/create-user: 创建单个用户 (SSHA加密) - POST /ldap-config/delete-user: 删除用户 - POST /ldap-config/batch-import-ldap: 批量导入用户 - POST /ldap-config/user-list: 查询用户列表(恢复) 前端 LdapConfigPage 增强: - LDAP 账号管理折叠区(配置存在即显示) - OU 下拉选择 + 根节点选项 + 数量统计 - 用户列表表格(全选/单选/DN显示) - 新建 OU 弹框(名称+父级DN+描述) - 添加用户弹框(用户名+密码+显示名+邮箱+目标OU) - 删除选中用户按钮(确认后批量删除) 前端 UsersPage BatchImportModal 三 Tab: - 本地导入 / 导入到 LDAP / 从 LDAP 导入 测试修复: - test_user_endpoint.py: 分页格式兼容 - test_users.py: 分页格式兼容 - test_system_settings.py: 精简后配置项匹配
233 lines
7.4 KiB
Python
233 lines
7.4 KiB
Python
"""用户管理 API 端点测试。"""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from app.core.security import hash_password, create_access_token
|
|
from app.models.user import User
|
|
|
|
|
|
def auth_header(token: str) -> dict:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestListUsers:
|
|
"""GET /api/v1/users"""
|
|
|
|
async def test_list_users_empty(self, client: AsyncClient, admin_token: str):
|
|
resp = await client.get("/api/v1/users", headers=auth_header(admin_token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# admin_user fixture 创建了 1 个
|
|
assert isinstance(data, dict) and "items" in data
|
|
|
|
async def test_list_users_filter_role(
|
|
self, client: AsyncClient, db, admin_token: str
|
|
):
|
|
# 创建一个 dba 用户
|
|
dba = User(
|
|
username="dba_test",
|
|
display_name="DBA",
|
|
role="dba",
|
|
hashed_password=hash_password("123456"),
|
|
)
|
|
db.add(dba)
|
|
await db.commit()
|
|
|
|
resp = await client.get(
|
|
"/api/v1/users?role=dba", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
items = data.get("items", data) if isinstance(data, dict) else data
|
|
assert all(u["role"] == "dba" for u in items)
|
|
|
|
async def test_list_users_filter_active(
|
|
self, client: AsyncClient, db, admin_token: str
|
|
):
|
|
dev = User(
|
|
username="inactive_dev",
|
|
display_name="离线",
|
|
role="developer",
|
|
is_active=False,
|
|
hashed_password=hash_password("123456"),
|
|
)
|
|
db.add(dev)
|
|
await db.commit()
|
|
|
|
resp = await client.get(
|
|
"/api/v1/users?is_active=false", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
items = data.get("items", data) if isinstance(data, dict) else data
|
|
assert all(u["is_active"] is False for u in items)
|
|
|
|
async def test_list_users_forbidden_non_admin(
|
|
self, client: AsyncClient, developer_token: str
|
|
):
|
|
resp = await client.get("/api/v1/users", headers=auth_header(developer_token))
|
|
assert resp.status_code == 403
|
|
|
|
async def test_list_users_unauthorized(self, client: AsyncClient):
|
|
resp = await client.get("/api/v1/users")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCreateUser:
|
|
"""POST /api/v1/users"""
|
|
|
|
async def test_create_user_success(self, client: AsyncClient, admin_token: str):
|
|
resp = await client.post(
|
|
"/api/v1/users",
|
|
json={
|
|
"username": "newuser",
|
|
"display_name": "新用户",
|
|
"email": "new@test.com",
|
|
"phone": "13800001111",
|
|
"role": "developer",
|
|
"password": "pass123",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["username"] == "newuser"
|
|
assert data["role"] == "developer"
|
|
assert "hashed_password" not in data
|
|
|
|
async def test_create_user_duplicate_username(
|
|
self, client: AsyncClient, admin_token: str, admin_user: User
|
|
):
|
|
resp = await client.post(
|
|
"/api/v1/users",
|
|
json={
|
|
"username": "admin",
|
|
"display_name": "重复",
|
|
"role": "developer",
|
|
"password": "123456",
|
|
},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "已存在" in resp.json()["detail"]
|
|
|
|
async def test_create_user_forbidden_dba(
|
|
self, client: AsyncClient, dba_token: str
|
|
):
|
|
resp = await client.post(
|
|
"/api/v1/users",
|
|
json={
|
|
"username": "hack",
|
|
"display_name": "hack",
|
|
"role": "developer",
|
|
"password": "123",
|
|
},
|
|
headers=auth_header(dba_token),
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestGetUser:
|
|
"""GET /api/v1/users/{user_id}"""
|
|
|
|
async def test_get_user_success(
|
|
self, client: AsyncClient, admin_token: str, admin_user: User
|
|
):
|
|
resp = await client.get(
|
|
f"/api/v1/users/{admin_user.id}", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["username"] == "admin"
|
|
|
|
async def test_get_user_not_found(self, client: AsyncClient, admin_token: str):
|
|
resp = await client.get(
|
|
"/api/v1/users/9999", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
async def test_get_user_forbidden(self, client: AsyncClient, developer_token: str):
|
|
resp = await client.get(
|
|
"/api/v1/users/1", headers=auth_header(developer_token)
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUpdateUser:
|
|
"""PUT /api/v1/users/{user_id}"""
|
|
|
|
async def test_update_user_display_name(
|
|
self, client: AsyncClient, admin_token: str, dba_user
|
|
):
|
|
resp = await client.put(
|
|
f"/api/v1/users/{dba_user.id}",
|
|
json={"display_name": "DBA王五"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["display_name"] == "DBA王五"
|
|
|
|
async def test_update_user_role(
|
|
self, client: AsyncClient, admin_token: str, developer_user
|
|
):
|
|
resp = await client.put(
|
|
f"/api/v1/users/{developer_user.id}",
|
|
json={"role": "dba"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["role"] == "dba"
|
|
|
|
async def test_update_user_not_found(self, client: AsyncClient, admin_token: str):
|
|
resp = await client.put(
|
|
"/api/v1/users/9999",
|
|
json={"display_name": "x"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
async def test_update_user_partial(
|
|
self, client: AsyncClient, admin_token: str, dba_user
|
|
):
|
|
resp = await client.put(
|
|
f"/api/v1/users/{dba_user.id}",
|
|
json={"email": "dba@test.com"},
|
|
headers=auth_header(admin_token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["email"] == "dba@test.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestDeleteUser:
|
|
"""DELETE /api/v1/users/{user_id}"""
|
|
|
|
async def test_delete_user_success(
|
|
self, client: AsyncClient, admin_token: str, developer_user
|
|
):
|
|
resp = await client.delete(
|
|
f"/api/v1/users/{developer_user.id}", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 200
|
|
assert "已删除" in resp.json()["message"]
|
|
|
|
async def test_delete_self_forbidden(
|
|
self, client: AsyncClient, admin_token: str, admin_user: User
|
|
):
|
|
resp = await client.delete(
|
|
f"/api/v1/users/{admin_user.id}", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "不能删除自己" in resp.json()["detail"]
|
|
|
|
async def test_delete_user_not_found(self, client: AsyncClient, admin_token: str):
|
|
resp = await client.delete(
|
|
"/api/v1/users/9999", headers=auth_header(admin_token)
|
|
)
|
|
assert resp.status_code == 404
|