zkxw/backend/tests/test_mp_bind.py

"""服务号 mp-bind / mp-bind-status：成功与冲突路径（mock oauth）。"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from fastapi import HTTPException


@pytest.mark.asyncio
async def test_mp_bind_success():
    user = SimpleNamespace(
        id="u1",
        del_flag=0,
        wx_mp_openid=None,
        wx_unionid=None,
        wx_mp_bind_time=None,
        update_time=None,
    )
    db = MagicMock()
    with (
        patch(
            "app.api.mini_auth._resolve_user_id",
            new_callable=AsyncMock,
            return_value="u1",
        ),
        patch(
            "app.api.mini_auth.oauth_code_to_openid",
            return_value={"openid": "mp-oid-1", "unionid": "uni-1"},
        ),
        patch("app.api.mini_auth.user_crud.get_by_wx_mp_openid", return_value=None),
        patch("app.api.mini_auth.user_crud.get", return_value=user),
    ):
        from app.api.mini_auth import mp_bind

        resp = await mp_bind(
            request=MagicMock(),
            body={"code": "oauth-code"},
            db=db,
        )

    assert resp["code"] == 200
    assert resp["data"]["bound"] is True
    assert user.wx_mp_openid == "mp-oid-1"
    assert user.wx_unionid == "uni-1"
    assert user.wx_mp_bind_time is not None
    db.add.assert_called_once_with(user)
    db.commit.assert_called_once()


@pytest.mark.asyncio
async def test_mp_bind_conflict_when_openid_bound_to_other():
    other = SimpleNamespace(id="u-other")
    db = MagicMock()
    with (
        patch(
            "app.api.mini_auth._resolve_user_id",
            new_callable=AsyncMock,
            return_value="u1",
        ),
        patch(
            "app.api.mini_auth.oauth_code_to_openid",
            return_value={"openid": "mp-oid-taken", "unionid": None},
        ),
        patch(
            "app.api.mini_auth.user_crud.get_by_wx_mp_openid",
            return_value=other,
        ),
    ):
        from app.api.mini_auth import mp_bind

        resp = await mp_bind(
            request=MagicMock(),
            body={"code": "oauth-code"},
            db=db,
        )

    assert resp["code"] == 409
    assert "已绑定其他账号" in resp["message"]
    db.commit.assert_not_called()


@pytest.mark.asyncio
async def test_mp_bind_status_bound_true():
    user = SimpleNamespace(id="u1", wx_mp_openid="mp-oid-1")
    with (
        patch(
            "app.api.mini_auth._resolve_user_id",
            new_callable=AsyncMock,
            return_value="u1",
        ),
        patch("app.api.mini_auth.user_crud.get", return_value=user),
    ):
        from app.api.mini_auth import mp_bind_status

        resp = await mp_bind_status(request=MagicMock(), db=MagicMock())

    assert resp["code"] == 200
    assert resp["data"]["bound"] is True


@pytest.mark.asyncio
async def test_mp_bind_unauthorized():
    with patch(
        "app.api.mini_auth._resolve_user_id",
        new_callable=AsyncMock,
        side_effect=HTTPException(status_code=401, detail="未授权"),
    ):
        from app.api.mini_auth import mp_bind

        resp = await mp_bind(
            request=MagicMock(),
            body={"code": "x"},
            db=MagicMock(),
        )

    assert resp["code"] == 401
    assert resp["message"] == "未授权"