xiaoxia-saas/tests/unit/test_inmemory_user_repository.py

"""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