From 80722d76db74f6ba0cca4e7706d2f8d39526a622 Mon Sep 17 00:00:00 2001 From: zlwu <704541015@qq.com> Date: Thu, 13 Aug 2026 18:15:53 +0800 Subject: [PATCH] feat(mini-auth): add WeChat service-account bind APIs Co-authored-by: Cursor --- backend/app/api/mini_auth.py | 47 ++++++++++++++ backend/app/crud/user.py | 10 +++ backend/tests/test_mp_bind.py | 117 ++++++++++++++++++++++++++++++++++ 3 files changed, 174 insertions(+) create mode 100644 backend/tests/test_mp_bind.py diff --git a/backend/app/api/mini_auth.py b/backend/app/api/mini_auth.py index 0e31e9e7..453295c3 100644 --- a/backend/app/api/mini_auth.py +++ b/backend/app/api/mini_auth.py @@ -20,6 +20,7 @@ from app.services.wechat_mini_auth import ( code2session, get_phone_number, ) +from app.services.wechat_mp_auth import WechatMpAuthError, oauth_code_to_openid router = APIRouter(prefix="/mini/auth", tags=["小程序鉴权"]) @@ -120,3 +121,49 @@ async def wechat_bind( db.add(user) db.commit() return success_response({"bound": True, "message": "绑定成功"}) + + +@router.get("/mp-bind-status", response_model=dict) +async def mp_bind_status(request: Request, db: Session = Depends(get_db)): + """当前用户是否已绑定微信服务号 openid。""" + try: + user_id = await _resolve_user_id(request, db) + except HTTPException: + return error_response(401, "未授权") + + user = user_crud.get(db, user_id) + bound = bool(user and (getattr(user, "wx_mp_openid", None) or "").strip()) + return success_response({"bound": bound}) + + +@router.post("/mp-bind", response_model=dict) +async def mp_bind(request: Request, body: dict, db: Session = Depends(get_db)): + """服务号网页授权 code 绑定当前用户（需 Bearer）。""" + try: + user_id = await _resolve_user_id(request, db) + except HTTPException: + return error_response(401, "未授权") + + code = (body.get("code") or "").strip() + try: + info = oauth_code_to_openid(code) + except WechatMpAuthError as exc: + return error_response(400, str(exc)) + + mp_openid = info["openid"] + other = user_crud.get_by_wx_mp_openid(db, mp_openid) + if other and other.id != user_id: + return error_response(409, "该微信服务号已绑定其他账号") + + user = user_crud.get(db, user_id) + if not user or user.del_flag != 0: + return error_response(401, "用户不存在") + + user.wx_mp_openid = mp_openid + if info.get("unionid"): + user.wx_unionid = info["unionid"] + user.wx_mp_bind_time = datetime.now() + user.update_time = datetime.now() + db.add(user) + db.commit() + return success_response({"bound": True}) diff --git a/backend/app/crud/user.py b/backend/app/crud/user.py index 2ae6211e..56475003 100644 --- a/backend/app/crud/user.py +++ b/backend/app/crud/user.py @@ -41,6 +41,16 @@ class _UserCRUDCompat(CRUDBase[SysUser]): .first() ) + def get_by_wx_mp_openid(self, db: Session, wx_mp_openid: str) -> Optional[SysUser]: + oid = (wx_mp_openid or "").strip() + if not oid: + return None + return ( + db.query(SysUser) + .filter(SysUser.wx_mp_openid == oid, SysUser.del_flag == 0) + .first() + ) + def list_active_by_phone_no(self, db: Session, phone_no: str) -> List[SysUser]: phone = (phone_no or "").strip() if not phone: diff --git a/backend/tests/test_mp_bind.py b/backend/tests/test_mp_bind.py new file mode 100644 index 00000000..8df547af --- /dev/null +++ b/backend/tests/test_mp_bind.py @@ -0,0 +1,117 @@ +"""服务号 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"] == "未授权"