xiaoxia-saas/tests/unit/test_wechat_bind_routes_1719.py

"""#1719：微信绑定/解绑路由层测试（直接驱动路由函数）。

覆盖：
- GET  /wechat/bind/url：调 oauth 生成链接、记日志
- POST /wechat/bind：oauth 失败→400；绑定成功→success+user.wechat_bound=True；
  use case 返回冲突→对应状态码透传
- DELETE /wechat/bind：成功→success=True；use case 报错→状态码透传
- /auth/me 返回 wechat_bound 字段
"""

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from app.api.routes import auth as auth_route  # noqa: E402
from fastapi import HTTPException  # noqa: E402


def _auth_user(user_id="u-1", openid=None):
    user = SimpleNamespace(
        id=user_id,
        wechat_openid=openid,
        email="user@example.com",
        email_verified=True,
        username="user",
        display_name="用户",
        phone="",
        phone_verified=False,
        profile_completed=True,
    )
    return SimpleNamespace(user=user, session_id="s-1", token_type="user_auth")


def _patched_bind(result, error, status):
    """构造打了补丁的 wechat_bind_use_case 模块"""
    mod = SimpleNamespace(
        WechatBindRequest=lambda **kw: SimpleNamespace(**kw),
        WechatBindUseCase=MagicMock(),
        WechatUnbindUseCase=MagicMock(),
    )
    fake_bind_uc = MagicMock()
    fake_bind_uc.bind.return_value = (result, error, status)
    mod.WechatBindUseCase.return_value = fake_bind_uc
    return mod


def test_get_bind_url_returns_url_and_state():
    fake_oauth = MagicMock()
    fake_oauth.generate_auth_url.return_value = ("https://open.weixin.qq.com/qrconnect?xxx", "state-bind-1")

    import packages.application.auth.wechat_oauth_service as oauth_mod

    orig = oauth_mod.get_wechat_oauth_service
    oauth_mod.get_wechat_oauth_service = MagicMock(return_value=fake_oauth)
    try:
        resp = asyncio.run(auth_route.get_wechat_bind_url(current_user=_auth_user()))
    finally:
        oauth_mod.get_wechat_oauth_service = orig

    assert resp.auth_url.startswith("https://open.weixin.qq.com")
    assert resp.state == "state-bind-1"


def test_bind_oauth_error_returns_400():
    fake_oauth = MagicMock()
    fake_oauth.handle_callback.return_value = (None, "无效的 state 参数")

    import packages.application.auth.wechat_oauth_service as oauth_mod

    orig = oauth_mod.get_wechat_oauth_service
    oauth_mod.get_wechat_oauth_service = MagicMock(return_value=fake_oauth)
    try:
        with pytest.raises(HTTPException) as exc:
            asyncio.run(
                auth_route.wechat_bind(
                    SimpleNamespace(code="c-1", state="s-1"),
                    current_user=_auth_user(),
                    user_repository=MagicMock(),
                )
            )
    finally:
        oauth_mod.get_wechat_oauth_service = orig

    assert exc.value.status_code == 400
    assert "state" in exc.value.detail


def test_bind_success_returns_user_with_wechat_bound():
    fake_oauth = MagicMock()
    fake_oauth.handle_callback.return_value = (
        SimpleNamespace(openid="wx-openid-1", unionid="wx-union-1"),
        None,
    )

    bound_user = SimpleNamespace(
        id="u-1",
        wechat_openid="wx-openid-1",
        email="user@example.com",
        email_verified=True,
        username="user",
        display_name="用户",
        phone="",
        phone_verified=False,
        profile_completed=True,
    )

    import packages.application.auth.wechat_oauth_service as oauth_mod
    from packages.application.auth import wechat_bind_use_case as bind_mod

    orig_oauth = oauth_mod.get_wechat_oauth_service
    fake_bind_uc = MagicMock()
    fake_bind_uc.bind.return_value = (SimpleNamespace(user=bound_user), None, 200)
    orig_bind = bind_mod.WechatBindUseCase
    bind_mod.WechatBindUseCase = MagicMock(return_value=fake_bind_uc)
    oauth_mod.get_wechat_oauth_service = MagicMock(return_value=fake_oauth)
    try:
        resp = asyncio.run(
            auth_route.wechat_bind(
                SimpleNamespace(code="c-1", state="s-1"),
                current_user=_auth_user(),
                user_repository=MagicMock(),
            )
        )
    finally:
        oauth_mod.get_wechat_oauth_service = orig_oauth
        bind_mod.WechatBindUseCase = orig_bind

    assert resp.success is True
    assert resp.user.wechat_bound is True
    assert resp.user.user_id == "u-1"
    # 绑定请求应带上当前用户 id 与微信 openid
    call_kwargs = fake_bind_uc.bind.call_args[0][0]
    assert call_kwargs.user_id == "u-1"
    assert call_kwargs.openid == "wx-openid-1"


def test_bind_conflict_propagates_409():
    fake_oauth = MagicMock()
    fake_oauth.handle_callback.return_value = (
        SimpleNamespace(openid="wx-openid-1", unionid=""),
        None,
    )

    import packages.application.auth.wechat_oauth_service as oauth_mod
    from packages.application.auth import wechat_bind_use_case as bind_mod

    orig_oauth = oauth_mod.get_wechat_oauth_service
    fake_bind_uc = MagicMock()
    fake_bind_uc.bind.return_value = (None, "该微信已绑定其他账号，请先在原账号解绑", 409)
    orig_bind = bind_mod.WechatBindUseCase
    bind_mod.WechatBindUseCase = MagicMock(return_value=fake_bind_uc)
    oauth_mod.get_wechat_oauth_service = MagicMock(return_value=fake_oauth)
    try:
        with pytest.raises(HTTPException) as exc:
            asyncio.run(
                auth_route.wechat_bind(
                    SimpleNamespace(code="c-1", state="s-1"),
                    current_user=_auth_user(),
                    user_repository=MagicMock(),
                )
            )
    finally:
        oauth_mod.get_wechat_oauth_service = orig_oauth
        bind_mod.WechatBindUseCase = orig_bind

    assert exc.value.status_code == 409
    assert "已绑定其他账号" in exc.value.detail


def test_unbind_success_returns_success_true():
    unbound_user = SimpleNamespace(
        id="u-1",
        wechat_openid=None,
        email="user@example.com",
        email_verified=True,
        username="user",
        display_name="用户",
        phone="",
        phone_verified=False,
    )

    from packages.application.auth import wechat_bind_use_case as bind_mod

    fake_uc = MagicMock()
    fake_uc.unbind.return_value = (SimpleNamespace(user=unbound_user), None, 200)
    orig = bind_mod.WechatUnbindUseCase
    bind_mod.WechatUnbindUseCase = MagicMock(return_value=fake_uc)
    try:
        resp = asyncio.run(
            auth_route.wechat_unbind(current_user=_auth_user(openid="wx-old"), user_repository=MagicMock())
        )
    finally:
        bind_mod.WechatUnbindUseCase = orig

    assert resp.success is True
    fake_uc.unbind.assert_called_once_with("u-1")


def test_unbind_rejected_no_other_login_propagates_400():
    from packages.application.auth import wechat_bind_use_case as bind_mod

    fake_uc = MagicMock()
    fake_uc.unbind.return_value = (None, "账号需要至少一种其他登录方式（已验证手机或真实邮箱）后才能解绑微信", 400)
    orig = bind_mod.WechatUnbindUseCase
    bind_mod.WechatUnbindUseCase = MagicMock(return_value=fake_uc)
    try:
        with pytest.raises(HTTPException) as exc:
            asyncio.run(auth_route.wechat_unbind(current_user=_auth_user(openid="wx-old"), user_repository=MagicMock()))
    finally:
        bind_mod.WechatUnbindUseCase = orig

    assert exc.value.status_code == 400
    assert "登录方式" in exc.value.detail


def test_me_includes_wechat_bound_flag():
    # 已绑定用户
    resp = asyncio.run(auth_route.get_current_user_info(authenticated_user=_auth_user(openid="wx-openid-1")))
    assert resp.wechat_bound is True

    # 未绑定用户
    resp2 = asyncio.run(auth_route.get_current_user_info(authenticated_user=_auth_user(openid=None)))
    assert resp2.wechat_bound is False