kplam
6c6f690788
- 移除冗余依赖包 - 新增账号禁用校验与用户管理能力 - 新增文档下载与管理页面文件展示 - 新增API文档页面与用户管理前端页面 - 重构时区处理与docker-compose部署配置 - 完善测试用例与项目文档
408 lines
15 KiB
Python
408 lines
15 KiB
Python
"""UserStore / bootstrap_admin 单元测试（FakeRedis 与内存降级，不连真实 Redis）
|
||
|
||
覆盖：
|
||
- 哈希：同密码同 salt 一致；不同 salt 不同
|
||
- create：重名 UserExistsError；非法用户名/弱密码 ValueError；记录不含明文密码
|
||
- get/list/delete/verify_password/set_password/count_admins
|
||
- bootstrap_admin：空库创建 admin 并打印/返回密码；非空库不重复创建
|
||
- 内存降级（构造传 None）全功能可用
|
||
- Redis 异常包装为 UserStoreError
|
||
"""
|
||
|
||
import fnmatch
|
||
import json
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from app.core.users import (
|
||
UserExistsError,
|
||
UserNotFoundError,
|
||
UserStore,
|
||
UserStoreError,
|
||
bootstrap_admin,
|
||
)
|
||
|
||
|
||
class FakeRedis:
|
||
"""最小内存版 redis.asyncio 客户端：get/set/setex/delete/scan_iter/keys + TTL 记录
|
||
|
||
fail=True 时所有操作抛 ConnectionError，用于验证异常包装语义。
|
||
"""
|
||
|
||
def __init__(self, fail: bool = False) -> None:
|
||
self.fail = fail
|
||
self.store: dict[str, str] = {}
|
||
self.ttls: dict[str, int] = {}
|
||
|
||
def _check(self) -> None:
|
||
if self.fail:
|
||
raise ConnectionError("redis down")
|
||
|
||
async def get(self, key: str) -> str | None:
|
||
self._check()
|
||
return self.store.get(key)
|
||
|
||
async def set(self, key: str, value: str) -> bool:
|
||
self._check()
|
||
self.store[key] = value
|
||
return True
|
||
|
||
async def setex(self, key: str, ttl: int, value: str) -> bool:
|
||
self._check()
|
||
self.store[key] = value
|
||
self.ttls[key] = ttl
|
||
return True
|
||
|
||
async def delete(self, *keys: str) -> int:
|
||
self._check()
|
||
deleted = 0
|
||
for key in keys:
|
||
if key in self.store:
|
||
del self.store[key]
|
||
deleted += 1
|
||
return deleted
|
||
|
||
async def scan_iter(self, match: str = "*"):
|
||
self._check()
|
||
for key in list(self.store):
|
||
if fnmatch.fnmatch(key, match):
|
||
yield key
|
||
|
||
async def keys(self, pattern: str = "*") -> list[str]:
|
||
self._check()
|
||
return [key for key in self.store if fnmatch.fnmatch(key, pattern)]
|
||
|
||
|
||
class FakeLogger:
|
||
"""捕获 warning 调用的假 logger（structlog 风格：event + kwargs）"""
|
||
|
||
def __init__(self) -> None:
|
||
self.warnings: list[tuple[str, dict]] = []
|
||
|
||
def warning(self, event: str, **kwargs) -> None:
|
||
self.warnings.append((event, kwargs))
|
||
|
||
|
||
class TestHashPassword:
|
||
"""PBKDF2 哈希确定性"""
|
||
|
||
def test_same_password_same_salt_consistent(self):
|
||
store = UserStore(None)
|
||
salt = b"0123456789abcdef"
|
||
assert store.hash_password("password123", salt) == store.hash_password("password123", salt)
|
||
|
||
def test_different_salt_different_hash(self):
|
||
store = UserStore(None)
|
||
assert store.hash_password("password123", b"a" * 16) != store.hash_password("password123", b"b" * 16)
|
||
|
||
|
||
class TestCreate:
|
||
"""创建用户：校验、重名、无明文"""
|
||
|
||
async def test_create_success(self):
|
||
store = UserStore(FakeRedis())
|
||
record = await store.create("alice", "password123")
|
||
assert record.username == "alice"
|
||
assert record.role == "user"
|
||
assert record.must_change_password is False
|
||
# created_at 为可解析的 ISO8601
|
||
datetime.fromisoformat(record.created_at)
|
||
|
||
async def test_record_contains_no_plaintext(self):
|
||
redis = FakeRedis()
|
||
store = UserStore(redis)
|
||
record = await store.create("alice", "password123")
|
||
assert record.password_hash != "password123"
|
||
assert "password123" not in record.password_hash
|
||
# salt 为 16 字节 hex
|
||
assert len(bytes.fromhex(record.salt)) == 16
|
||
# 落库 JSON 同样不含明文密码
|
||
raw = redis.store["user:alice"]
|
||
assert "password123" not in raw
|
||
assert json.loads(raw)["username"] == "alice"
|
||
|
||
async def test_duplicate_raises_user_exists(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
with pytest.raises(UserExistsError):
|
||
await store.create("alice", "another-password")
|
||
|
||
@pytest.mark.parametrize("username", ["a", "x" * 33, "bad name", "bad!name", "中文名", ""])
|
||
async def test_invalid_username_raises_value_error(self, username: str):
|
||
with pytest.raises(ValueError):
|
||
await UserStore(FakeRedis()).create(username, "password123")
|
||
|
||
@pytest.mark.parametrize("username", ["ab", "a" * 32, "A-Z_0-9"])
|
||
async def test_valid_username_boundary(self, username: str):
|
||
record = await UserStore(FakeRedis()).create(username, "password123")
|
||
assert record.username == username
|
||
|
||
@pytest.mark.parametrize("password", ["", "short", "1234567"])
|
||
async def test_short_password_raises_value_error(self, password: str):
|
||
with pytest.raises(ValueError):
|
||
await UserStore(FakeRedis()).create("alice", password)
|
||
|
||
async def test_exact_min_password_length_ok(self):
|
||
record = await UserStore(FakeRedis()).create("alice", "12345678")
|
||
assert record.username == "alice"
|
||
|
||
|
||
class TestGetListDelete:
|
||
"""查询 / 列表 / 删除"""
|
||
|
||
async def test_get_roundtrip_and_miss(self):
|
||
store = UserStore(FakeRedis())
|
||
created = await store.create("alice", "password123")
|
||
assert await store.get("alice") == created
|
||
assert await store.get("nobody") is None
|
||
|
||
async def test_list_scans_only_user_keys(self):
|
||
redis = FakeRedis()
|
||
store = UserStore(redis)
|
||
await store.create("alice", "password123")
|
||
await store.create("bob", "password456", role="admin")
|
||
redis.store["other:key"] = "not-a-user" # 非 user: 前缀不参与
|
||
names = {r.username for r in await store.list()}
|
||
assert names == {"alice", "bob"}
|
||
|
||
async def test_delete_idempotent(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
assert await store.delete("alice") is True
|
||
assert await store.get("alice") is None
|
||
assert await store.delete("alice") is False
|
||
|
||
|
||
class TestVerifyPassword:
|
||
"""密码校验"""
|
||
|
||
async def test_success_returns_record(self):
|
||
store = UserStore(FakeRedis())
|
||
created = await store.create("alice", "password123")
|
||
assert await store.verify_password("alice", "password123") == created
|
||
|
||
async def test_wrong_password_returns_none(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
assert await store.verify_password("alice", "wrong-password") is None
|
||
|
||
async def test_unknown_user_returns_none(self):
|
||
assert await UserStore(FakeRedis()).verify_password("nobody", "password123") is None
|
||
|
||
|
||
class TestSetPassword:
|
||
"""重置密码"""
|
||
|
||
async def test_reset_clears_flag_and_invalidates_old(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "old-password", must_change_password=True)
|
||
assert await store.set_password("alice", "new-password") is True
|
||
record = await store.get("alice")
|
||
assert record is not None
|
||
assert record.must_change_password is False
|
||
assert await store.verify_password("alice", "old-password") is None
|
||
assert await store.verify_password("alice", "new-password") is not None
|
||
|
||
async def test_unknown_user_returns_false(self):
|
||
assert await UserStore(FakeRedis()).set_password("nobody", "password123") is False
|
||
|
||
async def test_short_password_raises_value_error(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
with pytest.raises(ValueError):
|
||
await store.set_password("alice", "short")
|
||
|
||
|
||
class TestCountAdmins:
|
||
async def test_counts_only_admin_role(self):
|
||
store = UserStore(FakeRedis())
|
||
assert await store.count_admins() == 0
|
||
await store.create("admin1", "password123", role="admin")
|
||
await store.create("admin2", "password123", role="admin")
|
||
await store.create("user1", "password123")
|
||
assert await store.count_admins() == 2
|
||
await store.delete("admin1")
|
||
assert await store.count_admins() == 1
|
||
|
||
|
||
class TestEnabledField:
|
||
"""enabled 字段：默认 True、create 传参、get/list 回读、存量兼容"""
|
||
|
||
async def test_default_enabled_is_true(self):
|
||
record = await UserStore(FakeRedis()).create("alice", "password123")
|
||
assert record.enabled is True
|
||
|
||
async def test_create_with_enabled_false(self):
|
||
redis = FakeRedis()
|
||
store = UserStore(redis)
|
||
record = await store.create("alice", "password123", enabled=False)
|
||
assert record.enabled is False
|
||
# 持久化后回读仍为 False
|
||
assert (await store.get("alice")).enabled is False
|
||
# 落库 JSON 含 enabled 字段
|
||
assert json.loads(redis.store["user:alice"])["enabled"] is False
|
||
|
||
async def test_get_list_roundtrip_enabled(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123", enabled=False)
|
||
await store.create("bob", "password123", role="admin")
|
||
assert (await store.get("alice")).enabled is False
|
||
assert (await store.get("bob")).enabled is True
|
||
records = {r.username: r for r in await store.list()}
|
||
assert records["alice"].enabled is False
|
||
assert records["bob"].enabled is True
|
||
|
||
async def test_legacy_record_without_enabled_defaults_true(self):
|
||
"""存量记录无 enabled 字段时按 True 兼容（get/list）"""
|
||
redis = FakeRedis()
|
||
store = UserStore(redis)
|
||
# 直接写入无 enabled 字段的存量记录
|
||
redis.store["user:legacy"] = json.dumps(
|
||
{
|
||
"username": "legacy",
|
||
"role": "user",
|
||
"password_hash": "hash",
|
||
"salt": "00" * 16,
|
||
"must_change_password": False,
|
||
"created_at": "2024-01-01T00:00:00+00:00",
|
||
}
|
||
)
|
||
record = await store.get("legacy")
|
||
assert record is not None
|
||
assert record.enabled is True
|
||
records = await store.list()
|
||
assert records[0].enabled is True
|
||
|
||
|
||
class TestUpdateUser:
|
||
"""update_user：角色/启用状态更新与校验"""
|
||
|
||
async def test_update_role(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
updated = await store.update_user("alice", role="admin")
|
||
assert updated.role == "admin"
|
||
assert updated.enabled is True # 未改动
|
||
# 持久化
|
||
record = await store.get("alice")
|
||
assert record is not None
|
||
assert record.role == "admin"
|
||
|
||
async def test_update_enabled(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123", role="admin")
|
||
updated = await store.update_user("alice", enabled=False)
|
||
assert updated.enabled is False
|
||
assert updated.role == "admin" # 未改动
|
||
|
||
async def test_update_both(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
updated = await store.update_user("alice", role="admin", enabled=False)
|
||
assert updated.role == "admin"
|
||
assert updated.enabled is False
|
||
|
||
async def test_user_not_found_raises(self):
|
||
with pytest.raises(UserNotFoundError):
|
||
await UserStore(FakeRedis()).update_user("nobody", role="admin")
|
||
|
||
async def test_invalid_role_raises_value_error(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
with pytest.raises(ValueError):
|
||
await store.update_user("alice", role="superuser")
|
||
|
||
async def test_no_fields_noop(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123", role="admin")
|
||
updated = await store.update_user("alice")
|
||
assert updated.role == "admin"
|
||
assert updated.enabled is True
|
||
|
||
|
||
class TestBootstrapAdmin:
|
||
"""空库引导创建默认管理员"""
|
||
|
||
async def test_empty_store_creates_admin_and_returns_password(self):
|
||
store = UserStore(FakeRedis())
|
||
logger = FakeLogger()
|
||
password = await bootstrap_admin(store, logger) # type: ignore[arg-type]
|
||
assert isinstance(password, str) and len(password) > 0
|
||
record = await store.get("admin")
|
||
assert record is not None
|
||
assert record.role == "admin"
|
||
assert record.must_change_password is True
|
||
# 返回的明文密码可直接通过校验
|
||
assert await store.verify_password("admin", password) is not None
|
||
# 明文密码仅在日志中打印一次
|
||
assert len(logger.warnings) == 1
|
||
assert logger.warnings[0][1].get("password") == password
|
||
|
||
async def test_non_empty_store_returns_none_without_changes(self):
|
||
store = UserStore(FakeRedis())
|
||
await store.create("alice", "password123")
|
||
logger = FakeLogger()
|
||
assert await bootstrap_admin(store, logger) is None # type: ignore[arg-type]
|
||
assert [r.username for r in await store.list()] == ["alice"]
|
||
assert logger.warnings == []
|
||
|
||
async def test_second_call_returns_none(self):
|
||
store = UserStore(FakeRedis())
|
||
logger = FakeLogger()
|
||
assert await bootstrap_admin(store, logger) is not None # type: ignore[arg-type]
|
||
assert await bootstrap_admin(store, logger) is None # type: ignore[arg-type]
|
||
assert len(await store.list()) == 1
|
||
|
||
|
||
class TestMemoryFallback:
|
||
"""构造传 None：纯内存 dict 降级，全功能可用"""
|
||
|
||
async def test_full_flow_without_redis(self):
|
||
store = UserStore(None)
|
||
await store.create("alice", "password123", role="admin")
|
||
await store.create("bob", "password456")
|
||
assert await store.count_admins() == 1
|
||
assert {r.username for r in await store.list()} == {"alice", "bob"}
|
||
assert await store.verify_password("alice", "password123") is not None
|
||
assert await store.verify_password("alice", "wrong-password") is None
|
||
assert await store.set_password("bob", "new-password") is True
|
||
assert await store.verify_password("bob", "new-password") is not None
|
||
assert await store.delete("bob") is True
|
||
assert await store.get("bob") is None
|
||
|
||
async def test_bootstrap_admin_memory(self):
|
||
store = UserStore(None)
|
||
password = await bootstrap_admin(store, FakeLogger()) # type: ignore[arg-type]
|
||
assert password is not None
|
||
record = await store.get("admin")
|
||
assert record is not None and record.role == "admin"
|
||
|
||
|
||
class TestRedisErrors:
|
||
"""Redis 读写异常统一包装为 UserStoreError（不静默）"""
|
||
|
||
async def test_create_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).create("alice", "password123")
|
||
|
||
async def test_get_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).get("alice")
|
||
|
||
async def test_list_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).list()
|
||
|
||
async def test_delete_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).delete("alice")
|
||
|
||
async def test_set_password_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).set_password("alice", "password123")
|
||
|
||
async def test_count_admins_raises_store_error(self):
|
||
with pytest.raises(UserStoreError):
|
||
await UserStore(FakeRedis(fail=True)).count_admins()
|