"""InMemoryUserRepository 单元测试.""" from datetime import datetime, timezone import pytest from packages.adapters.in_memory.user_repository import InMemoryUserRepository from packages.domain.entities import User @pytest.fixture def repo() -> InMemoryUserRepository: return InMemoryUserRepository() @pytest.fixture def sample_user() -> User: return User( id="user-1", email="Test@Example.com", display_name="Test User", username="testuser", password_hash="hashed-pw", email_verification_token="verify-token-123", password_reset_token="reset-token-456", wechat_openid="wx-openid-abc", wechat_unionid="wx-unionid-def", phone="13800138000", created_at=datetime.now(timezone.utc), ) class TestSaveAndFindById: def test_save_and_find_by_id(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_id("user-1") assert found is not None assert found.id == "user-1" assert found.email == "Test@Example.com" def test_find_by_id_not_found(self, repo): assert repo.find_by_id("nonexistent") is None def test_save_overwrite_existing(self, repo, sample_user): repo.save(sample_user) sample_user.display_name = "Updated Name" repo.save(sample_user) found = repo.find_by_id("user-1") assert found.display_name == "Updated Name" class TestFindByEmail: def test_find_by_email_case_insensitive(self, repo, sample_user): repo.save(sample_user) # 用不同大小写查找 found = repo.find_by_email("test@example.com") assert found is not None assert found.id == "user-1" def test_find_by_email_exact_case(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_email("Test@Example.com") assert found is not None def test_find_by_email_not_found(self, repo): assert repo.find_by_email("notfound@example.com") is None class TestFindByUsername: def test_find_by_username_case_insensitive(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_username("TESTUSER") assert found is not None assert found.id == "user-1" def test_find_by_username_not_found(self, repo): assert repo.find_by_username("nobody") is None def test_find_by_username_empty(self, repo, sample_user): sample_user.username = "" repo.save(sample_user) # 空 username 不应该建立索引,但查找空字符串应该返回None found = repo.find_by_username("") assert found is None class TestFindByVerificationToken: def test_find_by_verification_token(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_verification_token("verify-token-123") assert found is not None assert found.id == "user-1" def test_find_by_verification_token_not_found(self, repo): assert repo.find_by_verification_token("bad-token") is None class TestFindByPasswordResetToken: def test_find_by_password_reset_token(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_password_reset_token("reset-token-456") assert found is not None assert found.id == "user-1" def test_find_by_password_reset_token_not_found(self, repo): assert repo.find_by_password_reset_token("bad-token") is None class TestFindByWechat: def test_find_by_wechat_openid(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_wechat_openid("wx-openid-abc") assert found is not None assert found.id == "user-1" def test_find_by_wechat_openid_not_found(self, repo): assert repo.find_by_wechat_openid("bad-openid") is None def test_find_by_wechat_unionid(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_wechat_unionid("wx-unionid-def") assert found is not None assert found.id == "user-1" def test_find_by_wechat_unionid_not_found(self, repo): assert repo.find_by_wechat_unionid("bad-unionid") is None def test_find_by_wechat_unionid_empty(self, repo, sample_user): sample_user.wechat_unionid = None repo.save(sample_user) assert repo.find_by_wechat_unionid("") is None class TestFindByPhone: def test_find_by_phone(self, repo, sample_user): repo.save(sample_user) found = repo.find_by_phone("13800138000") assert found is not None assert found.id == "user-1" def test_find_by_phone_not_found(self, repo): assert repo.find_by_phone("13900139000") is None def test_find_by_phone_empty(self, repo, sample_user): sample_user.phone = None repo.save(sample_user) assert repo.find_by_phone("") is None class TestDelete: def test_delete_existing_user(self, repo, sample_user): repo.save(sample_user) assert repo.delete("user-1") is True assert repo.find_by_id("user-1") is None def test_delete_cleans_all_indexes(self, repo, sample_user): repo.save(sample_user) repo.delete("user-1") assert repo.find_by_email("test@example.com") is None assert repo.find_by_username("testuser") is None assert repo.find_by_verification_token("verify-token-123") is None assert repo.find_by_password_reset_token("reset-token-456") is None def test_delete_nonexistent_user(self, repo): assert repo.delete("nonexistent") is False def test_delete_twice_returns_false(self, repo, sample_user): repo.save(sample_user) assert repo.delete("user-1") is True assert repo.delete("user-1") is False class TestIndexUpdates: def test_save_new_user_with_same_email_overwrites_index(self, repo, sample_user): """不同用户同邮箱,后者覆盖索引.""" repo.save(sample_user) user2 = User( id="user-2", email="test@example.com", # 同邮箱不同大小写 display_name="User 2", username="user2", ) repo.save(user2) # 邮箱索引指向最后保存的用户 found = repo.find_by_email("test@example.com") assert found.id == "user-2" # 原用户仍然可通过ID找到 assert repo.find_by_id("user-1") is not None