xiaoxia-saas/tests/unit/test_wechat_oauth_service.py

"""微信 OAuth 服务单元测试."""

from __future__ import annotations

import time
from unittest.mock import MagicMock, patch

import pytest

from packages.application.auth.wechat_oauth_service import (
    STATE_TTL_SECONDS,
    MemoryStateStore,
    WechatOAuthService,
    WechatUserInfo,
    get_wechat_oauth_service,
)


class TestMemoryStateStore:
    """MemoryStateStore 测试"""

    def test_put_and_verify(self):
        """存入 state 后可以验证通过"""
        store = MemoryStateStore()
        store.put("state_123")
        assert store.verify_and_consume("state_123") is True

    def test_verify_nonexistent(self):
        """不存在的 state 验证失败"""
        store = MemoryStateStore()
        assert store.verify_and_consume("nonexistent") is False

    def test_state_consumed_after_verify(self):
        """state 验证后被消费，不能重复使用"""
        store = MemoryStateStore()
        store.put("state_123")
        assert store.verify_and_consume("state_123") is True
        assert store.verify_and_consume("state_123") is False

    def test_multiple_states(self):
        """多个 state 独立管理"""
        store = MemoryStateStore()
        store.put("state_a")
        store.put("state_b")
        assert store.verify_and_consume("state_a") is True
        assert store.verify_and_consume("state_b") is True

    def test_expired_state_cleaned(self):
        """过期 state 会被清理"""
        store = MemoryStateStore(ttl_seconds=1)
        store.put("expired_state")
        time.sleep(1.1)
        assert store.verify_and_consume("expired_state") is False

    def test_custom_ttl(self):
        """自定义 TTL"""
        store = MemoryStateStore(ttl_seconds=60)
        store.put("my_state")
        # 立即验证应该通过
        assert store.verify_and_consume("my_state") is True

    def test_clean_expired_on_put(self):
        """put 时清理过期 state"""
        store = MemoryStateStore(ttl_seconds=1)
        store.put("old_state")
        time.sleep(1.1)
        # put 新 state 时会触发清理
        store.put("new_state")
        # old_state 已经过期了，验证应该失败
        assert store.verify_and_consume("old_state") is False
        # new_state 应该还在
        assert store.verify_and_consume("new_state") is True

    def test_clean_expired_on_verify(self):
        """verify 时清理过期 state"""
        store = MemoryStateStore(ttl_seconds=1)
        store.put("old_state")
        time.sleep(1.1)
        # 验证不存在的 state 也会触发清理
        store.verify_and_consume("other_state")
        # old_state 已过期，验证失败
        assert store.verify_and_consume("old_state") is False


class TestWechatUserInfo:
    """WechatUserInfo 测试"""

    def test_create_with_openid(self):
        """仅用 openid 创建"""
        info = WechatUserInfo(openid="openid_123")
        assert info.openid == "openid_123"
        assert info.unionid == ""
        assert info.nickname == ""
        assert info.avatar_url == ""

    def test_create_with_all_fields(self):
        """所有字段创建"""
        info = WechatUserInfo(
            openid="openid_123",
            unionid="unionid_456",
            nickname="测试用户",
            avatar_url="https://example.com/avatar.jpg",
        )
        assert info.openid == "openid_123"
        assert info.unionid == "unionid_456"
        assert info.nickname == "测试用户"
        assert info.avatar_url == "https://example.com/avatar.jpg"


class TestWechatOAuthServiceInit:
    """WechatOAuthService 初始化测试"""

    def test_not_configured_default(self):
        """默认参数（无环境变量）时未配置"""
        with patch.dict("os.environ", {}, clear=False):
            # 确保环境变量为空
            service = WechatOAuthService(app_id="", app_secret="", redirect_uri="")
            assert service.is_configured() is False

    def test_configured_with_params(self):
        """显式传入配置时已配置"""
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
        )
        assert service.is_configured() is True

    def test_missing_app_id_not_configured(self):
        """缺少 app_id 未配置"""
        service = WechatOAuthService(
            app_id="",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
        )
        assert service.is_configured() is False

    def test_default_state_store(self):
        """默认使用 MemoryStateStore"""
        service = WechatOAuthService(app_id="wx123", app_secret="s", redirect_uri="https://x.com")
        assert isinstance(service._state_store, MemoryStateStore)

    def test_custom_state_store(self):
        """可以自定义 state_store"""
        custom_store = MagicMock()
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="s",
            redirect_uri="https://x.com",
            state_store=custom_store,
        )
        assert service._state_store is custom_store


class TestGenerateAuthUrl:
    """generate_auth_url 测试"""

    def test_returns_url_and_state(self):
        """返回 URL 和 state"""
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
        )
        url, state = service.generate_auth_url()
        assert isinstance(url, str)
        assert isinstance(state, str)
        assert len(state) > 0
        assert "weixin.qq.com" in url

    def test_url_contains_params(self):
        """URL 包含必要参数"""
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
        )
        url, state = service.generate_auth_url(scope="snsapi_login")

        assert "appid=wx123" in url
        assert "snsapi_login" in url
        assert state in url
        assert "response_type=code" in url

    def test_state_saved_to_store(self):
        """生成的 state 存入 store"""
        mock_store = MagicMock()
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
            state_store=mock_store,
        )
        url, state = service.generate_auth_url()
        mock_store.put.assert_called_once_with(state)

    def test_mock_mode_when_not_configured(self):
        """未配置时返回 mock URL"""
        service = WechatOAuthService(app_id="", app_secret="", redirect_uri="https://example.com/callback")
        url, state = service.generate_auth_url()
        assert "/mock/wechat/auth" in url
        assert "mock" in url

    def test_different_states_each_time(self):
        """每次生成不同的 state"""
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://example.com/callback",
        )
        _, state1 = service.generate_auth_url()
        _, state2 = service.generate_auth_url()
        assert state1 != state2


class TestHandleCallback:
    """handle_callback 测试"""

    def test_missing_code_returns_error(self):
        """缺少 code 返回错误"""
        service = WechatOAuthService(app_id="wx123", app_secret="s", redirect_uri="https://x.com")
        user_info, error = service.handle_callback("", "some_state")
        assert user_info is None
        assert "缺少授权码" in error

    def test_invalid_state_returns_error(self):
        """state 无效返回错误"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = False
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="s",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )
        user_info, error = service.handle_callback("code123", "bad_state")
        assert user_info is None
        assert "state" in error

    def test_mock_mode_when_not_configured(self):
        """未配置时返回 mock 用户信息"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="",
            app_secret="",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )
        user_info, error = service.handle_callback("mock_code_12345", "valid_state")

        assert error is None
        assert user_info is not None
        assert user_info.openid.startswith("mock_")
        assert "微信测试用户" in user_info.nickname

    def test_state_consumed_after_callback(self):
        """回调处理后 state 被消费"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="",
            app_secret="",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )
        service.handle_callback("code", "valid_state")
        mock_store.verify_and_consume.assert_called_once_with("valid_state")

    def test_real_mode_success(self):
        """真实模式下成功获取用户信息"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )

        mock_token_resp = MagicMock()
        mock_token_resp.json.return_value = {
            "access_token": "access_token_123",
            "openid": "real_openid",
            "unionid": "real_unionid",
        }
        mock_user_resp = MagicMock()
        mock_user_resp.json.return_value = {
            "nickname": "真实用户",
            "headimgurl": "https://wx.qlogo.cn/avatar.jpg",
        }

        with patch("requests.get") as mock_get:
            mock_get.side_effect = [mock_token_resp, mock_user_resp]
            user_info, error = service.handle_callback("auth_code", "valid_state")

        assert error is None
        assert user_info is not None
        assert user_info.openid == "real_openid"
        assert user_info.unionid == "real_unionid"
        assert user_info.nickname == "真实用户"
        assert user_info.avatar_url == "https://wx.qlogo.cn/avatar.jpg"

    def test_real_mode_token_error(self):
        """真实模式下 access_token 接口返回错误"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )

        mock_resp = MagicMock()
        mock_resp.json.return_value = {
            "errcode": 40029,
            "errmsg": "invalid code",
        }

        with patch("requests.get", return_value=mock_resp):
            user_info, error = service.handle_callback("bad_code", "valid_state")

        assert user_info is None
        assert error is not None
        assert "微信授权失败" in error

    def test_real_mode_userinfo_error(self):
        """真实模式下用户信息接口返回错误"""
        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )

        mock_token_resp = MagicMock()
        mock_token_resp.json.return_value = {
            "access_token": "access_123",
            "openid": "open_123",
        }
        mock_user_resp = MagicMock()
        mock_user_resp.json.return_value = {
            "errcode": 40001,
            "errmsg": "invalid token",
        }

        with patch("requests.get") as mock_get:
            mock_get.side_effect = [mock_token_resp, mock_user_resp]
            user_info, error = service.handle_callback("code", "state")

        assert user_info is None
        assert "获取用户信息失败" in error

    def test_real_mode_network_error(self):
        """网络异常时返回友好错误"""
        import requests

        mock_store = MagicMock()
        mock_store.verify_and_consume.return_value = True
        service = WechatOAuthService(
            app_id="wx123",
            app_secret="secret456",
            redirect_uri="https://x.com",
            state_store=mock_store,
        )

        with patch("requests.get", side_effect=requests.ConnectionError()):
            user_info, error = service.handle_callback("code", "state")

        assert user_info is None
        assert "暂不可用" in error

    def test_empty_state_returns_error(self):
        """空 state 返回错误"""
        service = WechatOAuthService(app_id="wx123", app_secret="s", redirect_uri="https://x.com")
        user_info, error = service.handle_callback("code123", "")
        assert user_info is None
        assert "state" in error


class TestGetWechatOAuthService:
    """get_wechat_oauth_service 函数测试"""

    def test_returns_service_instance(self):
        """返回 WechatOAuthService 实例"""
        service = get_wechat_oauth_service()
        assert isinstance(service, WechatOAuthService)