diff --git a/backend/app/services/wechat_login_service.py b/backend/app/services/wechat_login_service.py new file mode 100644 index 000000000..6b609e13e --- /dev/null +++ b/backend/app/services/wechat_login_service.py @@ -0,0 +1,58 @@ +"""微信小程序登录业务:openid + 手机号一致性校验(仅快捷登录绑号)。""" +from __future__ import annotations + +from datetime import datetime +from typing import Optional, Tuple + +from sqlalchemy.orm import Session + +from app.crud.user import user_crud +from app.models.models import SysUser +from app.services.wechat_mini_auth import normalize_cn_mobile, phones_match + + +def resolve_wechat_login_user( + db: Session, *, openid: str, phone: str +) -> Tuple[Optional[SysUser], Optional[str], Optional[int]]: + oid = (openid or "").strip() + if not oid: + return None, "微信未返回 openid", 400 + if not normalize_cn_mobile(phone): + return None, "请授权手机号登录,或使用账号密码登录", 400 + + bound = user_crud.get_by_wx_openid(db, oid) + if bound: + if getattr(bound, "status", "1") != "1": + return None, "账号已被停用,请联系管理员", 403 + if not phones_match(getattr(bound, "phone_no", None), phone): + return ( + None, + "微信与账号不匹配,请联系管理员", + 401, + ) + return bound, None, None + + users = user_crud.list_active_by_phone_no(db, phone) + if not users: + return ( + None, + "未找到对应员工账号(请确认手机号已在系统维护)", + 401, + ) + if len(users) > 1: + return None, "手机号对应多个账号,请联系管理员", 400 + + user = users[0] + other = user_crud.get_by_wx_openid(db, oid) + if other and other.id != user.id: + return None, "微信与账号不匹配,请联系管理员", 409 + existing_oid = (getattr(user, "wx_openid", None) or "").strip() + if existing_oid and existing_oid != oid: + return None, "微信与账号不匹配,请联系管理员", 409 + + user.wx_openid = oid + user.update_time = datetime.now() + db.add(user) + db.commit() + db.refresh(user) + return user, None, None diff --git a/backend/tests/test_wechat_login_resolve.py b/backend/tests/test_wechat_login_resolve.py new file mode 100644 index 000000000..6e8b7fb57 --- /dev/null +++ b/backend/tests/test_wechat_login_resolve.py @@ -0,0 +1,88 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +from app.services.wechat_login_service import resolve_wechat_login_user + + +def test_openid_bound_phone_mismatch_rejects(): + bound = SimpleNamespace(id="u1", phone_no="13800138000", wx_openid="oid-a") + db = MagicMock() + with patch( + "app.services.wechat_login_service.user_crud.get_by_wx_openid", + return_value=bound, + ): + user, err, code = resolve_wechat_login_user( + db, openid="oid-a", phone="13900139000" + ) + assert user is None + assert code == 401 + assert "不一致" in err or "不匹配" in err + + +def test_openid_bound_phone_match_ok(): + bound = SimpleNamespace(id="u1", phone_no="13800138000", wx_openid="oid-a") + db = MagicMock() + with patch( + "app.services.wechat_login_service.user_crud.get_by_wx_openid", + return_value=bound, + ): + user, err, code = resolve_wechat_login_user( + db, openid="oid-a", phone="+8613800138000" + ) + assert user is bound + assert err is None + assert code is None + + +def test_unbound_no_user_rejects(): + db = MagicMock() + with patch( + "app.services.wechat_login_service.user_crud.get_by_wx_openid", + return_value=None, + ), patch( + "app.services.wechat_login_service.user_crud.list_active_by_phone_no", + return_value=[], + ): + user, err, code = resolve_wechat_login_user( + db, openid="oid-new", phone="13800138000" + ) + assert user is None + assert code == 401 + + +def test_unbound_multi_user_rejects(): + db = MagicMock() + u1 = SimpleNamespace(id="a", phone_no="13800138000", wx_openid=None) + u2 = SimpleNamespace(id="b", phone_no="13800138000", wx_openid=None) + with patch( + "app.services.wechat_login_service.user_crud.get_by_wx_openid", + return_value=None, + ), patch( + "app.services.wechat_login_service.user_crud.list_active_by_phone_no", + return_value=[u1, u2], + ): + user, err, code = resolve_wechat_login_user( + db, openid="oid-new", phone="13800138000" + ) + assert user is None + assert code == 400 + assert "多个" in err + + +def test_unbound_unique_binds_openid(): + db = MagicMock() + u = SimpleNamespace(id="a", phone_no="13800138000", wx_openid=None, update_time=None) + with patch( + "app.services.wechat_login_service.user_crud.get_by_wx_openid", + side_effect=[None, None], + ), patch( + "app.services.wechat_login_service.user_crud.list_active_by_phone_no", + return_value=[u], + ): + user, err, code = resolve_wechat_login_user( + db, openid="oid-new", phone="13800138000" + ) + assert err is None + assert user is u + assert u.wx_openid == "oid-new" + db.commit.assert_called()