logan
305bbb153c
新增测试文件: - test_user_endpoint.py (18 tests): 用户CRUD/列表过滤/权限控制/自删保护 - test_pool_endpoint.py (12 tests): 连接池状态/重置/概览/mock外部依赖 - test_schema_drift_endpoint.py (13 tests): Schema快照捕获/漂移检测/历史查询 覆盖率提升: - user端点: 0→18 tests (CRUD+权限+重复用户名+自删保护) - pool端点: 0→12 tests (stats/reset/overview+角色访问控制) - schema_drift端点: 0→13 tests (capture/check/history+dict序列化兼容) 修复: run.sh启动脚本同步
231 lines
7.3 KiB
Python
231 lines
7.3 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, list)
|
|
|
|
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()
|
|
assert all(u["role"] == "dba" for u in data)
|
|
|
|
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()
|
|
assert all(u["is_active"] is False for u in data)
|
|
|
|
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
|