xiaoxia-saas/tests/unit/test_wechat_oauth_service.py

"""
微信 OAuth 服务单元测试（第二十波）

覆盖：
- MemoryStateStore (put / verify_and_consume / 过期清理)
- WechatOAuthService.is_configured
- WechatOAuthService.generate_auth_url (正常模式 + mock模式)
- WechatOAuthService.handle_callback (正常 / 缺code / state无效 / mock模式 / access_token失败 / userinfo失败 / 网络异常)
"""

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,
)

# ============================================================
# MemoryStateStore
# ============================================================


class TestMemoryStateStore:
    """MemoryStateStore 内存 state 存储"""

    def test_put_and_verify(self):
        """放入并验证成功"""
        store = MemoryStateStore()
        store.put("state-1")
        assert store.verify_and_consume("state-1") is True

    def test_verify_consumes_once(self):
        """state 是一次性的，验证后即消费"""
        store = MemoryStateStore()
        store.put("state-1")
        assert store.verify_and_consume("state-1") is True
        assert store.verify_and_consume("state-1") is False

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

    def test_expired_state_is_cleaned(self):
        """过期的 state 会被清理"""
        store = MemoryStateStore(ttl_seconds=1)  # 1秒过期
        store.put("state-1")
        time.sleep(1.1)
        assert store.verify_and_consume("state-1") is False

    def test_put_cleans_expired(self):
        """put 时会清理过期的"""
        store = MemoryStateStore(ttl_seconds=1)
        store.put("state-1")
        time.sleep(1.1)
        store.put("state-2")
        # state-1 应该被清理掉了
        assert len(store._states) == 1
        assert "state-2" in store._states

    def test_default_ttl(self):
        """默认 TTL 是 10 分钟"""
        store = MemoryStateStore()
        assert store._ttl == STATE_TTL_SECONDS


# ============================================================
# WechatOAuthService - is_configured
# ============================================================


class TestIsConfigured:
    """is_configured 配置检查"""

    def test_fully_configured(self):
        """三项都配置了"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        assert svc.is_configured() is True

    def test_missing_app_id(self):
        """缺 app_id"""
        svc = WechatOAuthService(app_id="", app_secret="secret", redirect_uri="https://example.com/cb")
        assert svc.is_configured() is False

    def test_missing_app_secret(self):
        """缺 app_secret"""
        svc = WechatOAuthService(app_id="wx123", app_secret="", redirect_uri="https://example.com/cb")
        assert svc.is_configured() is False

    def test_missing_redirect_uri(self):
        """缺 redirect_uri"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="")
        assert svc.is_configured() is False

    def test_none_configured(self):
        """全没配置"""
        svc = WechatOAuthService(app_id="", app_secret="", redirect_uri="")
        assert svc.is_configured() is False


# ============================================================
# WechatOAuthService - generate_auth_url
# ============================================================


class TestGenerateAuthUrl:
    """generate_auth_url 生成授权链接"""

    def test_configured_mode(self):
        """配置完整时生成正式微信授权链接"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        url, state = svc.generate_auth_url()

        assert "open.weixin.qq.com" in url
        assert "appid=wx123" in url
        assert "redirect_uri=" in url
        assert "response_type=code" in url
        assert "scope=snsapi_login" in url
        assert f"state={state}" in url
        assert "#wechat_redirect" in url
        assert state  # state 非空

    def test_mock_mode(self):
        """未配置时返回 mock URL"""
        svc = WechatOAuthService(app_id="", app_secret="", redirect_uri="")
        url, state = svc.generate_auth_url()

        assert "/mock/wechat/auth" in url
        assert "app_id=mock" in url
        assert f"state={state}" in url
        assert state

    def test_custom_scope(self):
        """自定义 scope"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        url, _ = svc.generate_auth_url(scope="snsapi_userinfo")
        assert "scope=snsapi_userinfo" in url

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

    def test_state_stored_in_store(self):
        """生成的 state 会存入 store，可被 callback 验证"""
        store = MemoryStateStore()
        svc = WechatOAuthService(
            app_id="wx123",
            app_secret="secret",
            redirect_uri="https://example.com/cb",
            state_store=store,
        )
        _, state = svc.generate_auth_url()
        assert store.verify_and_consume(state) is True


# ============================================================
# WechatOAuthService - handle_callback
# ============================================================


class TestHandleCallback:
    """handle_callback 处理微信回调"""

    def test_missing_code(self):
        """缺少授权码"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        user_info, err = svc.handle_callback("", "some-state")
        assert user_info is None
        assert "缺少授权码" in err

    def test_invalid_state(self):
        """state 无效或已过期"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        user_info, err = svc.handle_callback("code123", "invalid-state")
        assert user_info is None
        assert "state" in err

    def test_empty_state(self):
        """空 state"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        user_info, err = svc.handle_callback("code123", "")
        assert user_info is None
        assert "state" in err

    def test_mock_mode_success(self):
        """mock 模式下返回模拟用户信息"""
        svc = WechatOAuthService(app_id="", app_secret="", redirect_uri="")
        # 先生成一个有效的 state
        _, state = svc.generate_auth_url()

        user_info, err = svc.handle_callback("mock_code_123456", state)

        assert err is None
        assert user_info is not None
        assert user_info.openid.startswith("mock_")
        assert user_info.unionid.startswith("mock_union_")
        assert user_info.nickname == "微信测试用户"

    def test_configured_mode_success(self):
        """配置完整时正常调用微信 API"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        _, state = svc.generate_auth_url()

        with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get:
            # access_token 响应
            token_resp = MagicMock()
            token_resp.json.return_value = {
                "access_token": "at_123",
                "openid": "openid_abc",
                "unionid": "unionid_xyz",
                "expires_in": 7200,
            }
            # userinfo 响应
            user_resp = MagicMock()
            user_resp.json.return_value = {
                "openid": "openid_abc",
                "nickname": "测试用户",
                "headimgurl": "https://wx.qq.com/avatar.jpg",
                "sex": 1,
            }
            mock_get.side_effect = [token_resp, user_resp]

            user_info, err = svc.handle_callback("code_abc", state)

        assert err is None
        assert user_info is not None
        assert user_info.openid == "openid_abc"
        assert user_info.unionid == "unionid_xyz"
        assert user_info.nickname == "测试用户"
        assert user_info.avatar_url == "https://wx.qq.com/avatar.jpg"
        # 应该调用了两次 get
        assert mock_get.call_count == 2

    def test_access_token_failed(self):
        """access_token 接口返回错误"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        _, state = svc.generate_auth_url()

        with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get:
            err_resp = MagicMock()
            err_resp.json.return_value = {
                "errcode": 40029,
                "errmsg": "invalid code",
            }
            mock_get.return_value = err_resp

            user_info, err = svc.handle_callback("bad_code", state)

        assert user_info is None
        assert "微信授权失败" in err
        assert "invalid code" in err

    def test_userinfo_failed(self):
        """userinfo 接口返回错误"""
        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        _, state = svc.generate_auth_url()

        with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get:
            token_resp = MagicMock()
            token_resp.json.return_value = {
                "access_token": "at_123",
                "openid": "openid_abc",
            }
            err_resp = MagicMock()
            err_resp.json.return_value = {
                "errcode": 40001,
                "errmsg": "invalid credential",
            }
            mock_get.side_effect = [token_resp, err_resp]

            user_info, err = svc.handle_callback("code_abc", state)

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

    def test_network_error(self):
        """网络异常"""
        import requests

        svc = WechatOAuthService(app_id="wx123", app_secret="secret", redirect_uri="https://example.com/cb")
        _, state = svc.generate_auth_url()

        with patch("packages.application.auth.wechat_oauth_service.requests.get") as mock_get:
            mock_get.side_effect = requests.ConnectionError("timeout")

            user_info, err = svc.handle_callback("code_abc", state)

        assert user_info is None
        assert "微信服务暂不可用" in err

    def test_state_one_time_use(self):
        """state 一次性使用，重复使用会失败"""
        svc = WechatOAuthService(app_id="", app_secret="", redirect_uri="")
        _, state = svc.generate_auth_url()

        # 第一次成功
        user_info1, err1 = svc.handle_callback("code1", state)
        assert err1 is None
        assert user_info1 is not None

        # 第二次用同一个 state 失败
        user_info2, err2 = svc.handle_callback("code2", state)
        assert user_info2 is None
        assert "state" in err2


# ============================================================
# WechatUserInfo
# ============================================================


class TestWechatUserInfo:
    """WechatUserInfo 数据类"""

    def test_minimal_fields(self):
        info = WechatUserInfo(openid="abc")
        assert info.openid == "abc"
        assert info.unionid == ""
        assert info.nickname == ""
        assert info.avatar_url == ""

    def test_full_fields(self):
        info = WechatUserInfo(
            openid="abc",
            unionid="def",
            nickname="测试",
            avatar_url="https://example.com/avatar.jpg",
        )
        assert info.openid == "abc"
        assert info.unionid == "def"
        assert info.nickname == "测试"
        assert info.avatar_url == "https://example.com/avatar.jpg"


# ============================================================
# get_wechat_oauth_service
# ============================================================


class TestGetWechatOAuthService:
    """工厂函数"""

    def test_returns_service_instance(self):
        svc = get_wechat_oauth_service()
        assert isinstance(svc, WechatOAuthService)