xiaoxia-saas/tests/unit/test_inmemory_user_repository.py

"""InMemoryUserRepository 单元测试."""

import copy
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_raises_uniqueness_error(self, repo, sample_user):
        """不同用户同邮箱应触发唯一约束异常."""
        repo.save(sample_user)
        user2 = User(
            id="user-2",
            email="test@example.com",  # 同邮箱
            display_name="User 2",
            username="user2",
        )
        with pytest.raises(ValueError, match="Email already in use"):
            repo.save(user2)

    def test_same_user_update_email_allowed(self, repo, sample_user):
        """同一用户更新邮箱不触发唯一约束."""
        repo.save(sample_user)
        sample_user.email = "new@example.com"
        repo.save(sample_user)

        found = repo.find_by_email("new@example.com")
        assert found is not None
        assert found.id == "user-1"
        assert repo.find_by_email("test@example.com") is None

    def test_update_email_conflict_does_not_corrupt_indexes(self, repo, sample_user):
        """更新邮箱与其他用户冲突时，索引必须保持一致，旧邮箱索引不丢失."""
        repo.save(sample_user)
        # 第二个用户
        user2 = User(
            id="user-2",
            email="other@example.com",
            display_name="User 2",
            username="user2",
        )
        repo.save(user2)

        # 尝试把 user2 的邮箱改成 sample_user 的邮箱（冲突）
        user2_new = copy.deepcopy(user2)
        user2_new.email = "test@example.com"
        with pytest.raises(ValueError, match="Email already in use"):
            repo.save(user2_new)

        # 索引必须保持一致（user-1仍占test@example.com，user-2仍占other@example.com）
        assert repo.find_by_email("test@example.com").id == "user-1"
        assert repo.find_by_email("other@example.com").id == "user-2"
        assert repo.find_by_username("testuser").id == "user-1"
        assert repo.find_by_username("user2").id == "user-2"

    def test_update_username_conflict_does_not_corrupt_indexes(self, repo, sample_user):
        """更新用户名与他人冲突时，各索引保持一致不丢失."""
        repo.save(sample_user)
        user2 = User(
            id="user-2",
            email="other@example.com",
            display_name="User 2",
            username="user2",
            phone="13900000002",
        )
        repo.save(user2)

        # 尝试把 user2 用户名改成 testuser（冲突）
        user2_new = copy.deepcopy(user2)
        user2_new.username = "testuser"
        with pytest.raises(ValueError, match="Username already in use"):
            repo.save(user2_new)

        # 各索引必须保持一致
        assert repo.find_by_username("testuser").id == "user-1"
        assert repo.find_by_username("user2").id == "user-2"
        assert repo.find_by_email("other@example.com").id == "user-2"
        assert repo.find_by_phone("13900000002").id == "user-2"

    def test_save_stores_independent_copy(self, repo, sample_user):
        """save存储独立副本，外部修改对象不影响仓储内部状态."""
        repo.save(sample_user)

        # 外部修改对象属性
        original_email = sample_user.email
        sample_user.email = "hacked@example.com"
        sample_user.display_name = "Hacked"

        # 仓储中数据不应受影响
        stored = repo.find_by_id(sample_user.id)
        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)