118 lines
3.2 KiB
Python
118 lines
3.2 KiB
Python
"""服务号 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"] == "未授权"
|