From f0cf3f8504f42e329273aa76c7b37cde5fd462a0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 30 Jul 2026 07:07:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(inmemory):=20=E5=A2=9E=E5=8A=A0email?= =?UTF-8?q?=E7=A9=BA=E5=80=BC=E9=98=B2=E5=BE=A1+token=E5=94=AF=E4=B8=80?= =?UTF-8?q?=E6=80=A7=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - save方法增加email空值校验，避免None.lower()崩溃 - find_by_email/find_by_username增加空值短路返回None - 增加email_verification_token和password_reset_token唯一性约束 - 新增5个测试用例覆盖防御场景 --- .../adapters/in_memory/user_repository.py | 18 +++++++ tests/unit/test_inmemory_user_repository.py | 52 +++++++++++++++++++ 2 files changed, 70 insertions(+) diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py index 778d7a312..72cc8118d 100755 --- a/packages/adapters/in_memory/user_repository.py +++ b/packages/adapters/in_memory/user_repository.py @@ -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) diff --git a/tests/unit/test_inmemory_user_repository.py b/tests/unit/test_inmemory_user_repository.py index e40e277df..6d8eb4153 100755 --- a/tests/unit/test_inmemory_user_repository.py +++ b/tests/unit/test_inmemory_user_repository.py @@ -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)