kplam
6c6f690788
- 移除冗余依赖包 - 新增账号禁用校验与用户管理能力 - 新增文档下载与管理页面文件展示 - 新增API文档页面与用户管理前端页面 - 重构时区处理与docker-compose部署配置 - 完善测试用例与项目文档
248 lines
8.8 KiB
Python
248 lines
8.8 KiB
Python
"""Spec Task 5：用户管理增强（PATCH 端点 + enabled 字段）全链路集成验证
|
||
|
||
内存 UserStore/SessionStore 经 conftest.auth_stores 夹具注入 app.api.deps 单例，
|
||
通过 TestClient 走真实 HTTP 链路（不跑 lifespan，无需真实 Redis/Qdrant），覆盖：
|
||
- 改角色生效：admin 创建 user → PATCH 改 admin → GET /auth/me 与重新登录均反映新角色
|
||
- 禁用用户：旧 token 立即失效（session 已清，1005）；重新登录 1005（"账号已禁用"）
|
||
- 重新启用：PATCH enabled=True → login 恢复成功
|
||
- 最后 admin 保护：唯一 admin 降级/禁用自己 → 1001（且自身状态未变）
|
||
- PATCH 校验：空 body / 非法 role → 1001；不存在用户 → 1004；非 admin 调用 → 1006
|
||
"""
|
||
|
||
from typing import Any
|
||
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.main import app
|
||
|
||
|
||
def _login(client: TestClient, username: str, password: str) -> dict[str, Any]:
|
||
"""调登录接口并返回响应体"""
|
||
return client.post("/api/v1/auth/login", json={"username": username, "password": password}).json()
|
||
|
||
|
||
def _bearer(token: str) -> dict[str, str]:
|
||
"""构造 Authorization Bearer 请求头"""
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
def _admin_login(client: TestClient) -> dict[str, str]:
|
||
"""以 auth_stores 预置的 admin（admin/admin-pass-123）登录，返回 Bearer 请求头"""
|
||
resp = _login(client, "admin", "admin-pass-123")
|
||
assert resp["code"] == 0, f"admin 登录失败: {resp}"
|
||
return _bearer(resp["data"]["token"])
|
||
|
||
|
||
def _create_user(
|
||
client: TestClient,
|
||
admin_headers: dict[str, str],
|
||
username: str,
|
||
password: str,
|
||
role: str = "user",
|
||
) -> None:
|
||
"""admin 创建用户并断言成功"""
|
||
resp = client.post(
|
||
"/api/v1/auth/users",
|
||
json={"username": username, "password": password, "role": role},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 0, f"创建用户 {username} 失败: {resp}"
|
||
|
||
|
||
class TestRoleChange:
|
||
"""改角色生效全链路"""
|
||
|
||
def test_role_change_takes_effect(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
# admin 创建 user 角色账号 alice
|
||
_create_user(client, admin_headers, "alice", "alice-pass-123", role="user")
|
||
|
||
# alice 登录，初始角色 user
|
||
alice_login = _login(client, "alice", "alice-pass-123")
|
||
assert alice_login["code"] == 0
|
||
assert alice_login["data"]["role"] == "user"
|
||
alice_headers = _bearer(alice_login["data"]["token"])
|
||
|
||
# 改角色前 GET /auth/me 反映 user 角色
|
||
me_before = client.get("/api/v1/auth/me", headers=alice_headers).json()
|
||
assert me_before["code"] == 0
|
||
assert me_before["data"]["role"] == "user"
|
||
|
||
# admin PATCH 改 alice 角色为 admin
|
||
patched = client.patch(
|
||
"/api/v1/auth/users/alice",
|
||
json={"role": "admin"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert patched["code"] == 0
|
||
assert patched["data"]["role"] == "admin"
|
||
|
||
# 同一 token 立即反映新角色（角色变更不清 session，用户记录实时读取）
|
||
me_after = client.get("/api/v1/auth/me", headers=alice_headers).json()
|
||
assert me_after["code"] == 0
|
||
assert me_after["data"]["role"] == "admin"
|
||
|
||
# 重新登录也反映新角色
|
||
relogin = _login(client, "alice", "alice-pass-123")
|
||
assert relogin["code"] == 0
|
||
assert relogin["data"]["role"] == "admin"
|
||
|
||
|
||
class TestDisableUser:
|
||
"""禁用用户：旧 session 清除 + 登录拒绝"""
|
||
|
||
def test_disable_user_clears_session_and_blocks_login(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
_create_user(client, admin_headers, "bob", "bob-pass-123")
|
||
|
||
# bob 登录拿 token
|
||
bob_login = _login(client, "bob", "bob-pass-123")
|
||
assert bob_login["code"] == 0
|
||
bob_headers = _bearer(bob_login["data"]["token"])
|
||
|
||
# admin 禁用 bob（清空其全部 session）
|
||
disabled = client.patch(
|
||
"/api/v1/auth/users/bob",
|
||
json={"enabled": False},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert disabled["code"] == 0
|
||
assert disabled["data"]["enabled"] is False
|
||
|
||
# bob 旧 token 调鉴权端点 → 1005（session 已清，凭证无效）
|
||
me = client.get("/api/v1/auth/me", headers=bob_headers).json()
|
||
assert me["code"] == 1005
|
||
|
||
# bob 重新登录 → 1005（账号已禁用）
|
||
relogin = _login(client, "bob", "bob-pass-123")
|
||
assert relogin["code"] == 1005
|
||
assert "禁用" in relogin["message"]
|
||
|
||
|
||
class TestReenableUser:
|
||
"""重新启用：login 恢复成功"""
|
||
|
||
def test_reenable_user_allows_login(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
_create_user(client, admin_headers, "carol", "carol-pass-123")
|
||
|
||
# 先禁用 carol，确认登录被拒
|
||
disabled = client.patch(
|
||
"/api/v1/auth/users/carol",
|
||
json={"enabled": False},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert disabled["code"] == 0
|
||
assert _login(client, "carol", "carol-pass-123")["code"] == 1005
|
||
|
||
# 重新启用
|
||
enabled = client.patch(
|
||
"/api/v1/auth/users/carol",
|
||
json={"enabled": True},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert enabled["code"] == 0
|
||
assert enabled["data"]["enabled"] is True
|
||
|
||
# login 恢复成功
|
||
relogin = _login(client, "carol", "carol-pass-123")
|
||
assert relogin["code"] == 0
|
||
assert relogin["data"]["username"] == "carol"
|
||
|
||
|
||
class TestLastAdminProtection:
|
||
"""最后 admin 保护：唯一 admin 不可降级/禁用"""
|
||
|
||
def test_cannot_demote_last_admin(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
# 唯一 admin（admin）尝试降级自己 role=user → 1001
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/admin",
|
||
json={"role": "user"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 1001
|
||
# admin 未被降级，仍可访问 admin 专属端点
|
||
assert client.get("/api/v1/auth/users", headers=admin_headers).json()["code"] == 0
|
||
|
||
def test_cannot_disable_last_admin(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
# 唯一 admin 尝试禁用自己 enabled=False → 1001
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/admin",
|
||
json={"enabled": False},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 1001
|
||
# admin 未被禁用，旧 token 仍可用
|
||
assert client.get("/api/v1/auth/users", headers=admin_headers).json()["code"] == 0
|
||
|
||
|
||
class TestPatchValidation:
|
||
"""PATCH 端点校验：空 body / 非法 role / 不存在用户 / 非 admin"""
|
||
|
||
def test_empty_body_rejected(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
_create_user(client, admin_headers, "dave", "dave-pass-123")
|
||
|
||
# 空 body（role 与 enabled 均缺）→ 模型校验 1001
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/dave",
|
||
json={},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 1001
|
||
|
||
def test_invalid_role_rejected(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
_create_user(client, admin_headers, "dave", "dave-pass-123")
|
||
|
||
# role 非法（非 admin/user）→ 模型校验 1001
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/dave",
|
||
json={"role": "superuser"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 1001
|
||
|
||
def test_nonexistent_user_rejected(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
|
||
# 不存在用户 → 1004
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/ghost",
|
||
json={"role": "admin"},
|
||
headers=admin_headers,
|
||
).json()
|
||
assert resp["code"] == 1004
|
||
|
||
def test_non_admin_forbidden(self, auth_stores) -> None:
|
||
client = TestClient(app)
|
||
admin_headers = _admin_login(client)
|
||
_create_user(client, admin_headers, "eve", "eve-pass-123", role="user")
|
||
|
||
# eve（user 角色）登录后调 PATCH → 1006
|
||
eve_login = _login(client, "eve", "eve-pass-123")
|
||
assert eve_login["code"] == 0
|
||
eve_headers = _bearer(eve_login["data"]["token"])
|
||
|
||
resp = client.patch(
|
||
"/api/v1/auth/users/eve",
|
||
json={"role": "admin"},
|
||
headers=eve_headers,
|
||
).json()
|
||
assert resp["code"] == 1006
|