xiaoxia-saas/tests/unit/test_wechat_sync_use_case.py

"""微信同步登录 UseCase 单元测试."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from packages.application.auth.wechat_sync_use_case import (
    WechatSyncRequest,
    WechatSyncResponse,
    WechatSyncUseCase,
)
from packages.domain.entities import User


@pytest.fixture
def mock_user_repo():
    return MagicMock()


@pytest.fixture
def mock_session_store():
    return MagicMock()


@pytest.fixture
def sample_user():
    user = User(
        id="user_001",
        email="test@wechat.local",
        username="wx_test123",
        display_name="微信用户",
        password_hash="hashed",
        email_verified=True,
    )
    user.wechat_openid = "openid_123"
    user.wechat_unionid = "unionid_456"
    user.last_login_at = None
    user.last_login_ip = None
    return user


class TestWechatSyncRequest:
    """WechatSyncRequest 测试"""

    def test_openid_stripped(self):
        """openid 被 strip"""
        req = WechatSyncRequest(openid="  openid_123  ")
        assert req.openid == "openid_123"

    def test_unionid_stripped(self):
        """unionid 被 strip"""
        req = WechatSyncRequest(openid="o1", unionid="  unionid_456  ")
        assert req.unionid == "unionid_456"

    def test_default_nickname(self):
        """默认昵称"""
        req = WechatSyncRequest(openid="o1")
        assert req.nickname == "微信用户"

    def test_default_source(self):
        """默认来源"""
        req = WechatSyncRequest(openid="o1")
        assert req.source == "miniapp"

    def test_empty_unionid(self):
        """不传 unionid 默认为空字符串"""
        req = WechatSyncRequest(openid="o1")
        assert req.unionid == ""


class TestWechatSyncResponse:
    """WechatSyncResponse 测试"""

    def test_to_dict_contains_fields(self):
        """to_dict 包含所有必要字段"""
        resp = WechatSyncResponse(
            access_token="access_123",
            refresh_token="refresh_456",
            user_id="user_001",
            nickname="测试用户",
            avatar_url="https://example.com/avatar.jpg",
            is_new_user=False,
            expires_in=1800,
        )
        data = resp.to_dict()

        assert data["access_token"] == "access_123"
        assert data["token"] == "access_123"  # 兼容字段
        assert data["refresh_token"] == "refresh_456"
        assert data["user_id"] == "user_001"
        assert data["is_new_user"] is False
        assert data["expires_in"] == 1800
        assert "user" in data
        assert "user_info" in data
        assert data["user"]["id"] == "user_001"
        assert data["user"]["nickname"] == "测试用户"
        assert data["user"]["display_name"] == "测试用户"


class TestWechatSyncUseCaseLoginExisting:
    """已有用户登录测试"""

    def test_login_by_openid(self, mock_user_repo, mock_session_store, sample_user):
        """通过 openid 登录已有用户"""
        mock_user_repo.find_by_wechat_openid.return_value = sample_user
        mock_user_repo.find_by_wechat_unionid.return_value = None
        mock_user_repo.save.return_value = sample_user

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="openid_123", nickname="测试")
        response, error = use_case.execute(request)

        assert error is None
        assert response is not None
        assert response.user_id == "user_001"
        assert response.is_new_user is False
        mock_user_repo.find_by_wechat_openid.assert_called_once_with("openid_123")
        mock_session_store.save_session.assert_called_once()

    def test_login_by_unionid(self, mock_user_repo, mock_session_store, sample_user):
        """openid 没找到，通过 unionid 找到并绑定 openid"""
        sample_user.wechat_openid = None  # 没有当前 openid
        mock_user_repo.find_by_wechat_openid.return_value = None
        mock_user_repo.find_by_wechat_unionid.return_value = sample_user
        mock_user_repo.save.return_value = sample_user

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(
            openid="new_openid",
            unionid="unionid_456",
            nickname="测试",
        )
        response, error = use_case.execute(request)

        assert error is None
        assert response is not None
        assert response.is_new_user is False
        # 应该保存了新的 openid
        assert sample_user.wechat_openid == "new_openid"
        mock_user_repo.save.assert_called()

    def test_updates_last_login(self, mock_user_repo, mock_session_store, sample_user):
        """登录时更新最后登录信息"""
        mock_user_repo.find_by_wechat_openid.return_value = sample_user
        mock_user_repo.save.return_value = sample_user

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="openid_123")
        use_case.execute(request)

        assert sample_user.last_login_at is not None
        assert sample_user.last_login_ip == "bff_gateway"

    def test_returns_tokens(self, mock_user_repo, mock_session_store, sample_user):
        """返回 access_token 和 refresh_token"""
        mock_user_repo.find_by_wechat_openid.return_value = sample_user
        mock_user_repo.save.return_value = sample_user

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="openid_123")
        response, _ = use_case.execute(request)

        assert response.access_token is not None
        assert len(response.access_token) > 0
        assert response.refresh_token is not None
        assert len(response.refresh_token) > 0
        assert response.expires_in > 0


class TestWechatSyncUseCaseNewUser:
    """新用户注册测试"""

    def test_create_new_user(self, mock_user_repo, mock_session_store):
        """openid 和 unionid 都没找到，创建新用户"""
        mock_user_repo.find_by_wechat_openid.return_value = None
        mock_user_repo.find_by_wechat_unionid.return_value = None
        mock_user_repo.find_by_username.return_value = None  # username 不重复

        saved_user = None

        def capture_save(user):
            nonlocal saved_user
            saved_user = user

        mock_user_repo.save.side_effect = capture_save

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(
            openid="new_openid_789",
            unionid="new_union_789",
            nickname="新用户",
            avatar_url="https://example.com/avatar.jpg",
        )
        response, error = use_case.execute(request)

        assert error is None
        assert response is not None
        assert response.is_new_user is True
        assert saved_user is not None
        assert saved_user.wechat_openid == "new_openid_789"
        assert saved_user.wechat_unionid == "new_union_789"
        assert saved_user.email.endswith("@wechat.local")
        assert saved_user.username.startswith("wx_")
        assert saved_user.email_verified is True

    def test_new_user_email_based_on_openid(self, mock_user_repo, mock_session_store):
        """新用户邮箱基于 openid 生成"""
        mock_user_repo.find_by_wechat_openid.return_value = None
        mock_user_repo.find_by_wechat_unionid.return_value = None
        mock_user_repo.find_by_username.return_value = None

        saved_user = None

        def capture_save(user):
            nonlocal saved_user
            saved_user = user

        mock_user_repo.save.side_effect = capture_save

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="abcdef1234567890")
        use_case.execute(request)

        assert "abcdef1234567890" in saved_user.email or "abcdef1234567890"[:20] in saved_user.email
        assert saved_user.email.endswith("@wechat.local")

    def test_username_conflict_adds_suffix(self, mock_user_repo, mock_session_store):
        """用户名冲突时加后缀"""
        call_count = [0]

        def mock_find_by_username(username):
            # 前两次返回存在（模拟冲突），第三次返回 None（可用）
            call_count[0] += 1
            if call_count[0] <= 2:
                return MagicMock()
            return None

        mock_user_repo.find_by_wechat_openid.return_value = None
        mock_user_repo.find_by_wechat_unionid.return_value = None
        mock_user_repo.find_by_username.side_effect = mock_find_by_username

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="test_openid")
        response, error = use_case.execute(request)

        assert error is None
        assert response is not None
        assert response.is_new_user is True
        # find_by_username 被调用了多次（找不冲突的用户名）
        assert mock_user_repo.find_by_username.call_count >= 2

    def test_new_user_has_password_hash(self, mock_user_repo, mock_session_store):
        """新用户有随机密码哈希（不能是空的）"""
        mock_user_repo.find_by_wechat_openid.return_value = None
        mock_user_repo.find_by_wechat_unionid.return_value = None
        mock_user_repo.find_by_username.return_value = None

        saved_user = None

        def capture_save(user):
            nonlocal saved_user
            saved_user = user

        mock_user_repo.save.side_effect = capture_save

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="new_openid")
        use_case.execute(request)

        assert saved_user.password_hash is not None
        assert len(saved_user.password_hash) > 0


class TestWechatSyncUseCaseErrors:
    """错误场景测试"""

    def test_empty_openid(self, mock_user_repo, mock_session_store):
        """空 openid 返回错误"""
        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="")
        response, error = use_case.execute(request)

        assert response is None
        assert "openid is required" in error

    def test_exception_returns_error(self, mock_user_repo, mock_session_store):
        """异常时返回友好错误"""
        mock_user_repo.find_by_wechat_openid.side_effect = Exception("DB error")

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="openid_123")
        response, error = use_case.execute(request)

        assert response is None
        assert "Internal error" in error


class TestWechatSyncSession:
    """Session 相关测试"""

    def test_session_saved(self, mock_user_repo, mock_session_store, sample_user):
        """登录时保存 session"""
        mock_user_repo.find_by_wechat_openid.return_value = sample_user
        mock_user_repo.save.return_value = sample_user

        use_case = WechatSyncUseCase(
            mock_user_repo,
            session_store=mock_session_store,
            jwt_secret_key="test-secret-key-for-jwt-12345",
        )
        request = WechatSyncRequest(openid="openid_123", source="miniapp")
        use_case.execute(request)

        mock_session_store.save_session.assert_called_once()
        call_kwargs = mock_session_store.save_session.call_args[1]
        assert call_kwargs["user_id"] == "user_001"
        assert "wechat_miniapp" in call_kwargs["device_info"]
        assert call_kwargs["expires_in_seconds"] == 30 * 24 * 3600