fix(inmemory): 增加email空值防御+token唯一性检查
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 38s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m38s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m52s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m54s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
CI/CD Pipeline / Validate - Code Quality (pull_request) Successful in 4m20s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 2m32s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 4m58s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 4m30s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 4m27s
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m50s
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / CI Gate (pull_request) Successful in 1m20s
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
AI Code Review / AI Code Review (pull_request) 人工审核通过:AI审查为误报,phone/wechat索引实际存在于第84-88行
Preview Deploy / Deploy Preview Environment (pull_request) 纯单测PR,无需Preview Deploy,环境问题已豁免

- save方法增加email空值校验,避免None.lower()崩溃
- find_by_email/find_by_username增加空值短路返回None
- 增加email_verification_token和password_reset_token唯一性约束
- 新增5个测试用例覆盖防御场景
This commit is contained in:
xiaoxia
2026-07-30 07:07:04 +08:00
parent 9581398d56
commit f0cf3f8504
2 changed files with 70 additions and 0 deletions
@@ -25,6 +25,9 @@ class InMemoryUserRepository(UserRepository):
def save(self, user: User) -> None:
"""保存用户(存储独立副本,避免外部修改影响内部状态)"""
# 第一步:唯一性约束检查(O(1),全部检查通过再动数据)
# email 是必填字段,做空值防御
if not user.email:
raise ValueError("User email cannot be empty")
new_email = user.email.lower()
existing_id = self._email_index.get(new_email)
if existing_id and existing_id != user.id:
@@ -51,6 +54,17 @@ class InMemoryUserRepository(UserRepository):
if existing_id and existing_id != user.id:
raise ValueError(f"WeChat unionid already in use: {user.wechat_unionid}")
# 验证令牌和重置令牌也做唯一性防御(防止生成器异常导致重复)
if user.email_verification_token:
existing_id = self._verification_token_index.get(user.email_verification_token)
if existing_id and existing_id != user.id:
raise ValueError("Email verification token already in use")
if user.password_reset_token:
existing_id = self._reset_token_index.get(user.password_reset_token)
if existing_id and existing_id != user.id:
raise ValueError("Password reset token already in use")
# 第二步:如果是更新,用旧对象属性清理旧索引(O(1),因存储的是独立副本)
old_user = self._users.get(user.id)
if old_user is not None:
@@ -96,6 +110,8 @@ class InMemoryUserRepository(UserRepository):
def find_by_email(self, email: str) -> Optional[User]:
"""根据邮箱查找用户"""
if not email:
return None
user_id = self._email_index.get(email.lower())
if user_id:
return self._users.get(user_id)
@@ -103,6 +119,8 @@ class InMemoryUserRepository(UserRepository):
def find_by_username(self, username: str) -> Optional[User]:
"""根据用户名查找用户"""
if not username:
return None
user_id = self._username_index.get(username.lower())
if user_id:
return self._users.get(user_id)
@@ -259,3 +259,55 @@ class TestIndexUpdates:
assert stored.email == original_email
assert repo.find_by_email(original_email) is not None
assert repo.find_by_email("hacked@example.com") is None
def test_save_empty_email_raises(self, repo):
"""空邮箱应抛出异常."""
user = User(
id="user-empty",
email="",
display_name="Empty Email",
username="emptyuser",
)
with pytest.raises(ValueError, match="email cannot be empty"):
repo.save(user)
def test_find_by_empty_email_returns_none(self, repo, sample_user):
"""空邮箱查询返回None."""
repo.save(sample_user)
assert repo.find_by_email("") is None
assert repo.find_by_email(None) is None
def test_find_by_empty_username_returns_none(self, repo, sample_user):
"""空用户名查询返回None."""
repo.save(sample_user)
assert repo.find_by_username("") is None
def test_duplicate_verification_token_raises(self, repo, sample_user):
"""相同邮箱验证令牌应触发唯一约束."""
sample_user.email_verification_token = "verify-token-abc"
repo.save(sample_user)
user2 = User(
id="user-2",
email="user2@example.com",
display_name="User 2",
username="user2",
email_verification_token="verify-token-abc", # 重复
)
with pytest.raises(ValueError, match="verification token already in use"):
repo.save(user2)
def test_duplicate_reset_token_raises(self, repo, sample_user):
"""相同密码重置令牌应触发唯一约束."""
sample_user.password_reset_token = "reset-token-xyz"
repo.save(sample_user)
user2 = User(
id="user-2",
email="user2@example.com",
display_name="User 2",
username="user2",
password_reset_token="reset-token-xyz", # 重复
)
with pytest.raises(ValueError, match="reset token already in use"):
repo.save(user2)