Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 07bed2f114 | |||
| b02cf0f73a | |||
| f161d07615 | |||
| 84a20d353e |
@@ -64,6 +64,11 @@ CHUANGLAN_SMS_TEMPLATE_ID=1022457679
|
||||
CHUANGLAN_SMS_SIGNATURE=
|
||||
CHUANGLAN_SMS_ENDPOINT=https://smssh.253.com/msg/sms/v2/tpl/send
|
||||
CHUANGLAN_SMS_TIMEOUT_SEC=10
|
||||
# --- 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值 ---
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_ID=
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
|
||||
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
|
||||
ALIYUN_ONEKEY_TIMEOUT_SEC=15
|
||||
|
||||
# ===== 测试账号(release 包全流程联调用)=====
|
||||
# 配一个固定测试手机号,专供无 SIM 卡 / 不走一键登录时打通全流程:该号登录【免短信验证码】
|
||||
|
||||
+13
-12
@@ -33,7 +33,7 @@ from app.core.security import (
|
||||
issue_token_pair,
|
||||
)
|
||||
from app.integrations import wxpay
|
||||
from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
|
||||
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
|
||||
from app.integrations.sms import SmsError, send_code, verify_code
|
||||
from app.repositories import onboarding as onboarding_repo
|
||||
from app.repositories import phone_rebind as rebind_repo
|
||||
@@ -110,14 +110,15 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
|
||||
):
|
||||
raise HTTPException(status_code=403, detail="当前设备环境异常,暂无法登录")
|
||||
logger.info(
|
||||
"jverify_login operator=%s token_len=%d",
|
||||
"jverify_login provider=%s operator=%s token_len=%d",
|
||||
req.provider or "-",
|
||||
req.operator or "-",
|
||||
len(req.login_token),
|
||||
)
|
||||
|
||||
try:
|
||||
phone = verify_and_get_phone(req.login_token)
|
||||
except JiguangError as e:
|
||||
phone = verify_and_get_phone(req.provider, req.login_token)
|
||||
except OneClickError as e:
|
||||
risk_repo.record_behavior_event(
|
||||
db,
|
||||
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
|
||||
@@ -128,11 +129,11 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
|
||||
client_ip=_client_ip(request),
|
||||
outcome="failed",
|
||||
reason=str(e),
|
||||
details={"operator": req.operator or None},
|
||||
details={"operator": req.operator or None, "provider": req.provider or None},
|
||||
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
|
||||
)
|
||||
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
|
||||
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
|
||||
|
||||
user = user_repo.upsert_user_for_login(db, phone=phone, register_channel="jverify")
|
||||
risk_repo.record_behavior_event(
|
||||
@@ -146,7 +147,7 @@ def jverify_login(req: JverifyLoginRequest, request: Request, db: DbSession) ->
|
||||
phone=phone,
|
||||
client_ip=_client_ip(request),
|
||||
outcome="success",
|
||||
details={"operator": req.operator or None},
|
||||
details={"operator": req.operator or None, "provider": req.provider or None},
|
||||
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
|
||||
)
|
||||
if user.status != "active":
|
||||
@@ -492,10 +493,10 @@ def wechat_bind_phone_jverify(
|
||||
raise HTTPException(status_code=401, detail="授权已过期,请重新用微信登录") from e
|
||||
|
||||
try:
|
||||
phone = verify_and_get_phone(req.login_token)
|
||||
except JiguangError as e:
|
||||
logger.error("[JG] verify+decrypt failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=502, detail=f"jiguang verify failed: {e}") from e
|
||||
phone = verify_and_get_phone("jiguang", req.login_token)
|
||||
except OneClickError as e:
|
||||
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
|
||||
|
||||
return _finish_wechat_bind(
|
||||
db,
|
||||
|
||||
@@ -165,6 +165,14 @@ class Settings(BaseSettings):
|
||||
CHUANGLAN_SMS_ENDPOINT: str = "https://smssh.253.com/msg/sms/v2/tpl/send"
|
||||
CHUANGLAN_SMS_TIMEOUT_SEC: int = 10 # httpx 读/连超时秒
|
||||
|
||||
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
|
||||
# 与短信同属 dypnsapi 产品:同一阿里云账号可复用 ALIYUN_SMS_ACCESS_KEY_*,默认独立字段解耦。
|
||||
# 缺凭证 → provider=aliyun 换号抛错→502,不启动崩(见 aliyun_oneclick_configured)。
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
|
||||
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
|
||||
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15 # 阿里云 API 读/连超时秒
|
||||
|
||||
@property
|
||||
def aliyun_sms_configured(self) -> bool:
|
||||
"""阿里云短信凭证齐全(缺则 SMS_PROVIDER=aliyun 时 /sms/* 返 503,而非启动崩)。"""
|
||||
@@ -175,6 +183,13 @@ class Settings(BaseSettings):
|
||||
and self.ALIYUN_SMS_TEMPLATE_CODE
|
||||
)
|
||||
|
||||
@property
|
||||
def aliyun_oneclick_configured(self) -> bool:
|
||||
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
|
||||
return bool(
|
||||
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
|
||||
)
|
||||
|
||||
@property
|
||||
def chuanglan_sms_configured(self) -> bool:
|
||||
"""创蓝短信凭证齐全(缺则 SMS_PROVIDER=chuanglan 时 /sms/send 返 503,而非启动崩)。"""
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
|
||||
|
||||
链路:
|
||||
Android 阿里云 SDK getLoginToken → spToken(access_token)
|
||||
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
|
||||
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
|
||||
|
||||
与 jiguang 对齐:对外暴露 verify_and_get_phone(login_token)->str,失败抛 AliyunOneClickError,
|
||||
由 oneclick.py 门面统一 catch。
|
||||
|
||||
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
|
||||
单测 monkeypatch 它即可,不触真 SDK/网络。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
logger = logging.getLogger("shagua.aliyun.onekey")
|
||||
|
||||
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
|
||||
|
||||
|
||||
class AliyunOneClickError(Exception):
|
||||
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
|
||||
|
||||
|
||||
def verify_and_get_phone(login_token: str) -> str:
|
||||
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
|
||||
if not settings.aliyun_oneclick_configured:
|
||||
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
|
||||
|
||||
result = _call_get_mobile(login_token)
|
||||
if not (result["success"] and result["code"] == "OK"):
|
||||
logger.error(
|
||||
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
|
||||
result["code"], result["message"],
|
||||
)
|
||||
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
|
||||
|
||||
phone = (result["mobile"] or "").strip()
|
||||
if not (phone.isdigit() and len(phone) == 11):
|
||||
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
|
||||
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
|
||||
return phone
|
||||
|
||||
|
||||
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
|
||||
|
||||
def _get_client():
|
||||
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
|
||||
global _client
|
||||
if _client is None:
|
||||
from alibabacloud_dypnsapi20170525.client import Client
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
|
||||
cfg = open_api_models.Config(
|
||||
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
|
||||
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
|
||||
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
|
||||
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
|
||||
)
|
||||
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
|
||||
_client = Client(cfg)
|
||||
return _client
|
||||
|
||||
|
||||
def _call_get_mobile(login_token: str) -> dict:
|
||||
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。
|
||||
|
||||
⚠️ GetMobile 响应体字段名以实际 SDK 为准(code=="OK"、get_mobile_result_dto.mobile)。接真号联调时
|
||||
若字段不同,只需改本函数末尾归一化,verify_and_get_phone 及单测不动。
|
||||
"""
|
||||
try:
|
||||
from alibabacloud_dypnsapi20170525 import models as dypns_models
|
||||
req = dypns_models.GetMobileRequest(access_token=login_token)
|
||||
body = _get_client().get_mobile(req).body
|
||||
except Exception as e:
|
||||
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
|
||||
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
|
||||
dto = getattr(body, "get_mobile_result_dto", None)
|
||||
mobile = getattr(dto, "mobile", None) if dto else None
|
||||
return {
|
||||
"success": (body.code == "OK"),
|
||||
"code": body.code,
|
||||
"message": body.message,
|
||||
"mobile": mobile,
|
||||
}
|
||||
@@ -0,0 +1,41 @@
|
||||
"""一键登录换号门面:按 provider 分派到极光/阿里云, 统一异常与手机号脱敏。
|
||||
|
||||
为什么要门面:一键登录 token 与「拉授权页的那家 SDK」强绑定 —— 客户端用极光 SDK 拉的
|
||||
token 只能用极光换号, 阿里云的只能用阿里云换号。所以 provider 必须由客户端如实上报, 服务端
|
||||
按此分派、不能猜。老客户端不带 provider → 默认极光(向后兼容)。
|
||||
|
||||
用法(api 层):
|
||||
from app.integrations import oneclick
|
||||
try:
|
||||
phone = oneclick.verify_and_get_phone(req.provider, req.login_token)
|
||||
except oneclick.OneClickError as e:
|
||||
raise HTTPException(502, ...) from e
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
# 用「模块属性访问」而非 from ... import 函数:保证 monkeypatch 各家实现时门面能拿到替身。
|
||||
from app.integrations import aliyun_onekey, jiguang
|
||||
from app.integrations.jiguang import mask_phone # 复用脱敏, 从门面转出供 api 层用
|
||||
|
||||
__all__ = ["OneClickError", "mask_phone", "verify_and_get_phone"]
|
||||
|
||||
PROVIDER_JIGUANG = "jiguang"
|
||||
PROVIDER_ALIYUN = "aliyun"
|
||||
|
||||
|
||||
class OneClickError(Exception):
|
||||
"""换号失败统一异常(不区分厂商), api 层 catch → 502。"""
|
||||
|
||||
|
||||
def verify_and_get_phone(provider: str, login_token: str) -> str:
|
||||
"""按 provider 换取明文手机号。失败(任一厂商)抛 OneClickError。
|
||||
|
||||
provider 大小写不敏感;空 / 未知值兜底走极光(主家), 不因客户端传错值而拒登。
|
||||
"""
|
||||
p = (provider or PROVIDER_JIGUANG).strip().lower()
|
||||
try:
|
||||
if p == PROVIDER_ALIYUN:
|
||||
return aliyun_onekey.verify_and_get_phone(login_token)
|
||||
return jiguang.verify_and_get_phone(login_token)
|
||||
except (jiguang.JiguangError, aliyun_onekey.AliyunOneClickError) as e:
|
||||
raise OneClickError(f"[{p}] {e}") from e
|
||||
@@ -67,6 +67,11 @@ class JverifyLoginRequest(BaseModel):
|
||||
device_model: str = Field(
|
||||
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
|
||||
)
|
||||
provider: str = Field(
|
||||
"jiguang",
|
||||
description="一键登录厂商:jiguang(默认)/aliyun。决定后端用哪家换号,"
|
||||
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
|
||||
)
|
||||
|
||||
|
||||
# ===== 短信验证码 =====
|
||||
|
||||
@@ -0,0 +1,479 @@
|
||||
# 阿里云一键登录 · 后端实施计划 (M1)
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** 让 `POST /api/v1/auth/jverify-login` 支持 `provider=aliyun`,用客户端 loginToken 调阿里云 Dypnsapi `GetMobile` 换明文手机号完成登录;极光/创蓝不受影响。
|
||||
|
||||
**Architecture:** 复用已存在的 `app/integrations/oneclick.py` 门面(按 provider 分派)。新增 `aliyun_onekey.py` 换号 provider(结构仿 `chuanglan_onekey.py`,SDK 调用仿 `sms/aliyun.py` 的惰性 client)。把端点从直连 `jiguang` 改走门面并读 `req.provider`(这一步会让已存在但未接线的 `tests/test_jverify_login_endpoint.py` 转绿)。
|
||||
|
||||
**Tech Stack:** FastAPI + Pydantic v2 + pytest + `alibabacloud_dypnsapi20170525`(短信侧已引)。
|
||||
|
||||
**参考规范:** `../specs/2026-07-28-aliyun-oneclick-login-design.md`(在 android 仓)。
|
||||
|
||||
**运行测试:** `.venv/Scripts/python -m pytest tests/test_oneclick.py tests/test_aliyun_onekey.py tests/test_jverify_login_endpoint.py -q`。(conftest 用临时文件 SQLite,端点测试**无需 PG**;`SGB_TEST_SKIP_DB` 未被 conftest 使用。)
|
||||
|
||||
> **执行状态(2026-07-28):M1 已实施并验证通过(未提交,待授权)。**
|
||||
> 实施中发现原计划前提有误:`oneclick.py` 门面 / `chuanglan_onekey.py` / `test_oneclick.py` / `test_jverify_login_endpoint.py` 均为**未提交的创蓝评估 WIP,已清除,git 全程无记录**——门面并不存在。遂按「建小 facade(jiguang+aliyun,弃创蓝)」重做,顺序:配置 → aliyun_onekey → **新建** oneclick 门面 → 接线 auth.py 两处 + provider 字段 → **新建**端点测试。
|
||||
> 验证:新增 15 单测全绿(aliyun_onekey 4 / oneclick 8 / 端点 3)+ 微信绑号回归绿;全量 `pytest` 与改动前逐一对照——同样 8 个既有失败(compare/coupon/invite/withdraw_tiers/notification,均与本次无关:SQLite/顺序依赖),本次 **0 新增失败**(617→632 passed)。
|
||||
> 下方 B1–B5 为实现参考(代码与落地一致);「已存在的门面/测试转绿」等措辞按本节修正理解,实际是先建 aliyun_onekey/facade 再接线。
|
||||
|
||||
---
|
||||
|
||||
### Task B1: 端点接入 oneclick 门面 + 加 `provider` 字段
|
||||
|
||||
已存在的 `tests/test_jverify_login_endpoint.py` 已按「门面已接线」写(monkeypatch `auth.verify_and_get_phone` 为 2 参、POST 带 `provider`),但线上 `auth.py` 仍直连 jiguang。本任务把它接上,使这些测试转绿。
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/schemas/auth.py:60-69`(`JverifyLoginRequest` 加 `provider`)
|
||||
- Modify: `app/api/v1/auth.py:36`(import 改门面)、`:112-157`(`jverify_login` 调用/异常/埋点)、`:495-498`(微信 `bind-phone/jverify` 端点同一 `verify_and_get_phone`/`JiguangError`,一并改)
|
||||
- Test: `tests/test_jverify_login_endpoint.py`(已存在,勿改,作为验收);回归 `tests/test_wechat_login.py`、`tests/test_wechat_conflict.py`(都覆盖微信绑号极光换号路径)
|
||||
|
||||
> ⚠️ `auth.py` 有**两处**用 `verify_and_get_phone` + `JiguangError`:`jverify_login`(119) 和微信 `bind-phone/jverify`(495)。换 import 会同时影响两处,必须都改;否则微信绑号编译/运行报错。
|
||||
|
||||
- [ ] **Step 1: 跑现有端点测试确认基线**
|
||||
|
||||
Run: `pytest tests/test_jverify_login_endpoint.py -v`
|
||||
Expected: 3 条用例 **FAIL**(当前端点用 1 参 `verify_and_get_phone(req.login_token)`,被 monkeypatch 成 2 参后 `TypeError`)或在无 PG 时 **SKIP**。任一情况都说明「尚未接线」。
|
||||
|
||||
- [ ] **Step 2: schema 加 `provider` 字段**
|
||||
|
||||
`app/schemas/auth.py` 的 `JverifyLoginRequest`(当前 60-69 行)末尾加一个字段:
|
||||
|
||||
```python
|
||||
class JverifyLoginRequest(BaseModel):
|
||||
login_token: str = Field(..., description="客户端 loginAuth 拿到的 loginToken", min_length=1)
|
||||
operator: str = Field("", description="CM/CU/CT,用于日志,可选")
|
||||
device_id: str = Field(
|
||||
"", max_length=64,
|
||||
description="硬件级设备标识(Android ANDROID_ID),用于新手引导按 设备+账号 去重;空=按未完成处理",
|
||||
)
|
||||
device_model: str = Field(
|
||||
"", max_length=128, description="客户端设备型号快照,用于登录安全审计"
|
||||
)
|
||||
provider: str = Field(
|
||||
"jiguang",
|
||||
description="一键登录厂商:jiguang(默认)/aliyun/chuanglan。决定后端用哪家换号,"
|
||||
"必须与客户端拉授权页的 SDK 一致。老客户端不带→默认极光(向后兼容)。",
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 3: 端点改走门面**
|
||||
|
||||
`app/api/v1/auth.py` 第 36 行的 import 从 jiguang 换成门面:
|
||||
|
||||
```python
|
||||
# 旧: from app.integrations.jiguang import JiguangError, mask_phone, verify_and_get_phone
|
||||
from app.integrations.oneclick import OneClickError, mask_phone, verify_and_get_phone
|
||||
```
|
||||
|
||||
`jverify_login`(当前 102-157 行)里的换号调用与异常分支改为:
|
||||
|
||||
```python
|
||||
try:
|
||||
phone = verify_and_get_phone(req.provider, req.login_token)
|
||||
except OneClickError as e:
|
||||
risk_repo.record_behavior_event(
|
||||
db,
|
||||
event_type=risk_repo.EVENT_ONECLICK_LOGIN,
|
||||
subject_type="device",
|
||||
subject_id=subject_id,
|
||||
device_id=req.device_id or None,
|
||||
device_model=req.device_model or None,
|
||||
client_ip=_client_ip(request),
|
||||
outcome="failed",
|
||||
reason=str(e),
|
||||
details={"operator": req.operator or None, "provider": req.provider or None},
|
||||
evaluate_rule=risk_repo.RULE_ONECLICK_DAILY,
|
||||
)
|
||||
logger.error("[oneclick] verify+decrypt failed: %s", e, exc_info=True)
|
||||
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
|
||||
```
|
||||
|
||||
并把该函数里成功埋点的 `details={"operator": req.operator or None}`(约 149 行)同样补上 `"provider": req.provider or None`;日志行(约 112-116)改成打印 `req.provider`。
|
||||
|
||||
换 import 后 `JiguangError` 不再被引用,从第 36 行 import 去掉(只留 `OneClickError, mask_phone, verify_and_get_phone`)。
|
||||
|
||||
- [ ] **Step 3b: 微信 `bind-phone/jverify` 端点同步改(否则编译/运行报错)**
|
||||
|
||||
`auth.py` 第 495-498 行的第二处换号(微信登录·本机号极光绑定)也走门面。该请求体 `WechatBindPhoneJverifyRequest` **无** `provider` 字段,M1 保持极光不变,显式传字面量 `"jiguang"`:
|
||||
|
||||
```python
|
||||
try:
|
||||
phone = verify_and_get_phone("jiguang", req.login_token)
|
||||
except OneClickError as e:
|
||||
raise HTTPException(status_code=502, detail=f"oneclick verify failed: {e}") from e
|
||||
```
|
||||
|
||||
> 说明:当客户端 `WECHAT_BIND` 皮肤改用阿里云后,此端点 + `WechatBindPhoneJverifyRequest` 也要加 `provider`(否则阿里云 token 被极光换号必失败)。这属**后续增强**,不在 M1;已在 spec 待办登记。
|
||||
|
||||
- [ ] **Step 4: 跑端点测试转绿**
|
||||
|
||||
Run: `pytest tests/test_jverify_login_endpoint.py -v`(有 PG)
|
||||
Expected: 3 条全 **PASS**(默认 provider 建号、provider 原样透传、OneClickError→502)。
|
||||
|
||||
- [ ] **Step 5: 跑门面 + 微信绑号测试不回归**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`(门面本就 2 参,未动 → PASS)
|
||||
Run(有 PG): `pytest tests/test_wechat_login.py tests/test_wechat_conflict.py -v`
|
||||
Expected: 全 **PASS**(微信绑号极光换号路径改走门面后行为不变)。
|
||||
|
||||
- [ ] **Step 6: Commit**
|
||||
|
||||
```bash
|
||||
git add app/schemas/auth.py app/api/v1/auth.py
|
||||
git commit -m "feat(auth): 一键登录/微信绑号接入 oneclick 门面并支持 provider 字段"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task B2: 阿里云一键登录配置项
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/core/config.py`(极光段 68-70 之后、或阿里云短信段 149-158 附近加一段)
|
||||
- Modify: `.env.example`
|
||||
|
||||
- [ ] **Step 1: 加配置字段 + 已配置属性**
|
||||
|
||||
`app/core/config.py` 的 `Settings` 类里新增(放在阿里云短信段落之后,与 `aliyun_sms_configured` 属性相邻):
|
||||
|
||||
```python
|
||||
# ===== 阿里云号码认证·一键登录(Dypnsapi GetMobile 换号)=====
|
||||
# 与短信同属 dypnsapi 产品:同一阿里云账号时可复用 ALIYUN_SMS_ACCESS_KEY_*,
|
||||
# 默认独立字段解耦(见 spec §2 决策)。缺凭证 → provider=aliyun 换号抛错→502,不启动崩。
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_ID: str = ""
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_SECRET: str = ""
|
||||
ALIYUN_ONEKEY_ENDPOINT: str = "dypnsapi.aliyuncs.com"
|
||||
ALIYUN_ONEKEY_TIMEOUT_SEC: int = 15
|
||||
```
|
||||
|
||||
并在 `aliyun_sms_configured` 属性下方加:
|
||||
|
||||
```python
|
||||
@property
|
||||
def aliyun_oneclick_configured(self) -> bool:
|
||||
"""阿里云一键登录凭证齐全(缺则 provider=aliyun 换号抛错→502,而非启动崩)。"""
|
||||
return bool(
|
||||
self.ALIYUN_ONEKEY_ACCESS_KEY_ID and self.ALIYUN_ONEKEY_ACCESS_KEY_SECRET
|
||||
)
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 补 `.env.example`**
|
||||
|
||||
在 `.env.example` 里对应位置追加(值留空,注释说明可复用短信 AK/SK):
|
||||
|
||||
```dotenv
|
||||
# 阿里云号码认证·一键登录(Dypnsapi GetMobile)。同账号可填与 ALIYUN_SMS_ACCESS_KEY_* 相同的值。
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_ID=
|
||||
ALIYUN_ONEKEY_ACCESS_KEY_SECRET=
|
||||
ALIYUN_ONEKEY_ENDPOINT=dypnsapi.aliyuncs.com
|
||||
ALIYUN_ONEKEY_TIMEOUT_SEC=15
|
||||
```
|
||||
|
||||
- [ ] **Step 3: 冒烟导入不报错**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 python -c "from app.core.config import settings; print(settings.aliyun_oneclick_configured)"`
|
||||
Expected: 打印 `False`(默认空凭证)。
|
||||
|
||||
- [ ] **Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add app/core/config.py .env.example
|
||||
git commit -m "feat(config): 增加阿里云一键登录换号配置项"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task B3: 实现 `aliyun_onekey` 换号 provider(TDD)
|
||||
|
||||
**Files:**
|
||||
- Create: `app/integrations/aliyun_onekey.py`
|
||||
- Test: `tests/test_aliyun_onekey.py`
|
||||
|
||||
- [ ] **Step 1: 写失败测试**
|
||||
|
||||
Create `tests/test_aliyun_onekey.py`:
|
||||
|
||||
```python
|
||||
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
|
||||
|
||||
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.config import settings
|
||||
from app.integrations import aliyun_onekey
|
||||
|
||||
|
||||
def _configure(monkeypatch):
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
|
||||
|
||||
|
||||
def test_get_phone_success(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
|
||||
)
|
||||
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
|
||||
|
||||
|
||||
def test_api_failure_raises(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
|
||||
)
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
|
||||
|
||||
def test_non_phone_result_raises(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
|
||||
)
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
|
||||
|
||||
def test_not_configured_raises(monkeypatch):
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑测试确认失败**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
|
||||
Expected: FAIL,`ModuleNotFoundError: app.integrations.aliyun_onekey`。
|
||||
|
||||
- [ ] **Step 3: 实现 provider**
|
||||
|
||||
Create `app/integrations/aliyun_onekey.py`:
|
||||
|
||||
```python
|
||||
"""阿里云号码认证·一键登录服务端换号(Dypnsapi GetMobile)。
|
||||
|
||||
链路:
|
||||
Android 阿里云 SDK getLoginToken → spToken(access_token)
|
||||
→ 本服务调 Dypnsapi GetMobile(AccessToken=spToken)
|
||||
→ 阿里云直接返回明文手机号(无需 RSA/AES,比极光/创蓝少一步解密)
|
||||
|
||||
与 jiguang/chuanglan 对齐:对外暴露 verify_and_get_phone(login_token)->str,
|
||||
失败抛 AliyunOneClickError,由 oneclick.py 门面统一 catch。
|
||||
|
||||
SDK 交互隔离在 _call_get_mobile 薄封装(惰性 import + 惰性建 client,仿 sms/aliyun.py),
|
||||
单测 monkeypatch 它即可。凭证复用/独立见 config.ALIYUN_ONEKEY_*。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
logger = logging.getLogger("shagua.aliyun.onekey")
|
||||
|
||||
_client = None # 惰性构建的 dypnsapi client(模块级缓存)
|
||||
|
||||
|
||||
class AliyunOneClickError(Exception):
|
||||
"""阿里云取号失败的统一异常,由 oneclick 门面 catch 翻成 4xx/5xx。"""
|
||||
|
||||
|
||||
def verify_and_get_phone(login_token: str) -> str:
|
||||
"""对外唯一函数:loginToken(access_token) → 明文手机号。失败抛 AliyunOneClickError。"""
|
||||
if not settings.aliyun_oneclick_configured:
|
||||
raise AliyunOneClickError("ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET not configured")
|
||||
|
||||
result = _call_get_mobile(login_token)
|
||||
if not (result["success"] and result["code"] == "OK"):
|
||||
logger.error(
|
||||
"[ALIYUN-ONEKEY] get_mobile failed code=%s msg=%s",
|
||||
result["code"], result["message"],
|
||||
)
|
||||
raise AliyunOneClickError(f"aliyun get_mobile failed code={result['code']}")
|
||||
|
||||
phone = (result["mobile"] or "").strip()
|
||||
if not (phone.isdigit() and len(phone) == 11):
|
||||
logger.error("[ALIYUN-ONEKEY] unexpected mobile format: %r", phone)
|
||||
raise AliyunOneClickError("aliyun get_mobile returned non-phone")
|
||||
return phone
|
||||
|
||||
|
||||
# ==================== SDK 接缝(单测 monkeypatch 这个)====================
|
||||
|
||||
def _get_client():
|
||||
"""惰性构建 dypnsapi client(仿 sms/aliyun.py:jiguang-only 部署不加载 alibabacloud)。"""
|
||||
global _client
|
||||
if _client is None:
|
||||
from alibabacloud_dypnsapi20170525.client import Client
|
||||
from alibabacloud_tea_openapi import models as open_api_models
|
||||
|
||||
cfg = open_api_models.Config(
|
||||
access_key_id=settings.ALIYUN_ONEKEY_ACCESS_KEY_ID,
|
||||
access_key_secret=settings.ALIYUN_ONEKEY_ACCESS_KEY_SECRET,
|
||||
read_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000, # SDK 单位 ms
|
||||
connect_timeout=settings.ALIYUN_ONEKEY_TIMEOUT_SEC * 1000,
|
||||
)
|
||||
cfg.endpoint = settings.ALIYUN_ONEKEY_ENDPOINT
|
||||
_client = Client(cfg)
|
||||
return _client
|
||||
|
||||
|
||||
def _call_get_mobile(login_token: str) -> dict:
|
||||
"""调 GetMobile。返回归一化 {success, code, message, mobile};import/建 client/调用任一失败抛 AliyunOneClickError。"""
|
||||
try:
|
||||
from alibabacloud_dypnsapi20170525 import models as dypns_models
|
||||
req = dypns_models.GetMobileRequest(access_token=login_token)
|
||||
body = _get_client().get_mobile(req).body
|
||||
except Exception as e:
|
||||
logger.exception("[ALIYUN-ONEKEY] get_mobile 调用异常")
|
||||
raise AliyunOneClickError("aliyun get_mobile 调用异常") from e
|
||||
dto = getattr(body, "get_mobile_result_dto", None)
|
||||
mobile = getattr(dto, "mobile", None) if dto else None
|
||||
return {
|
||||
"success": (body.code == "OK"),
|
||||
"code": body.code,
|
||||
"message": body.message,
|
||||
"mobile": mobile,
|
||||
}
|
||||
```
|
||||
|
||||
> 注:`GetMobile` 响应体字段名以实际 SDK 为准(`code`=="OK"、`get_mobile_result_dto.mobile`)。接真号联调时若字段名不同(如 `mobile` 在别的 dto 下),只需改 `_call_get_mobile` 的最后归一化,`verify_and_get_phone` 及测试不动。
|
||||
|
||||
- [ ] **Step 4: 跑测试转绿**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py -v`
|
||||
Expected: 4 条全 **PASS**。
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add app/integrations/aliyun_onekey.py tests/test_aliyun_onekey.py
|
||||
git commit -m "feat(auth): 新增阿里云一键登录换号 provider(Dypnsapi GetMobile)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task B4: 门面注册 aliyun 分派(TDD)
|
||||
|
||||
**Files:**
|
||||
- Modify: `app/integrations/oneclick.py`
|
||||
- Test: `tests/test_oneclick.py`(加两条)
|
||||
|
||||
- [ ] **Step 1: 加失败测试**
|
||||
|
||||
在 `tests/test_oneclick.py` 顶部 import 加 `aliyun_onekey`:
|
||||
|
||||
```python
|
||||
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang, oneclick
|
||||
```
|
||||
|
||||
文件末尾追加:
|
||||
|
||||
```python
|
||||
def test_dispatch_aliyun(monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_al(t):
|
||||
seen["al"] = t
|
||||
return "13855555555"
|
||||
|
||||
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
|
||||
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13855555555"
|
||||
assert seen["al"] == "tok"
|
||||
|
||||
|
||||
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
|
||||
def boom(_t):
|
||||
raise aliyun_onekey.AliyunOneClickError("aliyun down")
|
||||
|
||||
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
|
||||
with pytest.raises(oneclick.OneClickError):
|
||||
oneclick.verify_and_get_phone("aliyun", "tok")
|
||||
```
|
||||
|
||||
- [ ] **Step 2: 跑确认失败**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py::test_dispatch_aliyun -v`
|
||||
Expected: FAIL(`aliyun` 未注册 → 走默认极光 → `13855555555` 不匹配 / `seen` 无 `al`)。
|
||||
|
||||
- [ ] **Step 3: 注册 aliyun 分派**
|
||||
|
||||
`app/integrations/oneclick.py` 三处改动:
|
||||
|
||||
```python
|
||||
# import(第 17 行)加入 aliyun_onekey:
|
||||
from app.integrations import aliyun_onekey, chuanglan_onekey, jiguang
|
||||
|
||||
# 常量段加:
|
||||
PROVIDER_ALIYUN = "aliyun"
|
||||
|
||||
# verify_and_get_phone 分派与异常 catch:
|
||||
def verify_and_get_phone(provider: str, login_token: str) -> str:
|
||||
p = (provider or PROVIDER_JIGUANG).strip().lower()
|
||||
try:
|
||||
if p == PROVIDER_CHUANGLAN:
|
||||
return chuanglan_onekey.verify_and_get_phone(login_token)
|
||||
if p == PROVIDER_ALIYUN:
|
||||
return aliyun_onekey.verify_and_get_phone(login_token)
|
||||
return jiguang.verify_and_get_phone(login_token)
|
||||
except (
|
||||
jiguang.JiguangError,
|
||||
chuanglan_onekey.ChuanglanError,
|
||||
aliyun_onekey.AliyunOneClickError,
|
||||
) as e:
|
||||
raise OneClickError(f"[{p}] {e}") from e
|
||||
```
|
||||
|
||||
- [ ] **Step 4: 跑门面测试全绿**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py -v`
|
||||
Expected: 全 **PASS**(含新增两条 + 原有创蓝/极光分派不回归)。
|
||||
|
||||
- [ ] **Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add app/integrations/oneclick.py tests/test_oneclick.py
|
||||
git commit -m "feat(auth): oneclick 门面注册阿里云 provider 分派"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task B5: 全量回归 + 真号联调冒烟
|
||||
|
||||
- [ ] **Step 1: 全量单测**
|
||||
|
||||
Run: `SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py tests/test_aliyun_onekey.py -v` 及(有 PG)`pytest tests/test_jverify_login_endpoint.py -v`
|
||||
Expected: 全 PASS。
|
||||
|
||||
- [ ] **Step 2: 填真实凭证冒烟(联调期)**
|
||||
|
||||
在 `.env` 填 `ALIYUN_ONEKEY_ACCESS_KEY_ID/SECRET`(阿里云控制台号码认证方案对应账号 AK/SK),起服务,用客户端真机取到的 loginToken:
|
||||
|
||||
Run:
|
||||
```bash
|
||||
curl -s -X POST http://127.0.0.1:8770/api/v1/auth/jverify-login \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d '{"login_token":"<真机取到的token>","provider":"aliyun","device_id":"smoke"}'
|
||||
```
|
||||
Expected: 200 + `user.phone` 为真实手机号 + `access_token` 非空。失败看日志 `shagua.aliyun.onekey`(`GetMobile failed code=...`)。
|
||||
|
||||
- [ ] **Step 3: 无凭证降级冒烟**
|
||||
|
||||
清空 `ALIYUN_ONEKEY_*` 重启,`provider=aliyun` 请求应 **502**(`OneClickError [aliyun] ... not configured`),且极光 `provider` 省略仍 200。
|
||||
|
||||
---
|
||||
|
||||
## 自检清单(实现者跑完对照)
|
||||
|
||||
- [ ] `provider` 缺省 = jiguang,老客户端不改也能登(B1)。
|
||||
- [ ] `provider=aliyun` 端到端换号 200(B5 Step2)。
|
||||
- [ ] 任一厂商换号失败 → 502,不建号(B1 tests)。
|
||||
- [ ] 缺阿里云凭证不 crash 启动,仅该 provider 502(B2 property + B5 Step3)。
|
||||
- [ ] 极光/创蓝分派与端点原行为不回归(B4 Step4 / B1 Step5)。
|
||||
@@ -0,0 +1,51 @@
|
||||
"""阿里云一键登录换号 provider 单测:monkeypatch SDK 接缝 _call_get_mobile,不触真 SDK/网络。
|
||||
|
||||
SGB_TEST_SKIP_DB=1 pytest tests/test_aliyun_onekey.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.core.config import settings
|
||||
from app.integrations import aliyun_onekey
|
||||
|
||||
|
||||
def _configure(monkeypatch):
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "ak")
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "sk")
|
||||
|
||||
|
||||
def test_get_phone_success(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "13800138000"},
|
||||
)
|
||||
assert aliyun_onekey.verify_and_get_phone("tok") == "13800138000"
|
||||
|
||||
|
||||
def test_api_failure_raises(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": False, "code": "MobileNumberIllegal", "message": "x", "mobile": None},
|
||||
)
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
|
||||
|
||||
def test_non_phone_result_raises(monkeypatch):
|
||||
_configure(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
aliyun_onekey, "_call_get_mobile",
|
||||
lambda t: {"success": True, "code": "OK", "message": "OK", "mobile": "not-a-phone"},
|
||||
)
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
|
||||
|
||||
def test_not_configured_raises(monkeypatch):
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_ID", "")
|
||||
monkeypatch.setattr(settings, "ALIYUN_ONEKEY_ACCESS_KEY_SECRET", "")
|
||||
with pytest.raises(aliyun_onekey.AliyunOneClickError):
|
||||
aliyun_onekey.verify_and_get_phone("tok")
|
||||
@@ -0,0 +1,50 @@
|
||||
"""jverify-login 端点:provider 分派 + 错误映射(端到端:换号 → 建号 → 签 JWT)。
|
||||
|
||||
需要 DB(建号)。本机无 PG 时 client fixture 会自动 skip(SGB_TEST_SKIP_DB=1);
|
||||
PG 环境(CI 或本地起 Docker)完整跑。换号一步 monkeypatch 掉, 只验端点编排:
|
||||
provider 是否原样传给门面、成功建号、OneClickError → 502。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from app.api.v1 import auth
|
||||
from app.integrations.oneclick import OneClickError
|
||||
|
||||
|
||||
def test_jverify_login_default_provider_builds_account(client, monkeypatch):
|
||||
"""不带 provider 的老客户端 → 门面按默认(极光)换号, 建号登录成功。"""
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: "13500000000")
|
||||
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok-old"})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["user"]["phone"] == "13500000000"
|
||||
assert body["access_token"]
|
||||
|
||||
|
||||
def test_jverify_login_passes_provider_to_facade(client, monkeypatch):
|
||||
"""provider=aliyun 必须原样传给门面(它决定用哪家换号)。"""
|
||||
captured: dict = {}
|
||||
|
||||
def fake(provider, token):
|
||||
captured["provider"] = provider
|
||||
captured["token"] = token
|
||||
return "13500000001"
|
||||
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", fake)
|
||||
r = client.post(
|
||||
"/api/v1/auth/jverify-login",
|
||||
json={"login_token": "tok-al", "provider": "aliyun"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert captured["provider"] == "aliyun"
|
||||
assert captured["token"] == "tok-al"
|
||||
|
||||
|
||||
def test_jverify_login_oneclick_error_maps_to_502(client, monkeypatch):
|
||||
"""任一厂商换号失败(OneClickError) → 502, 不建号。"""
|
||||
|
||||
def boom(provider, token):
|
||||
raise OneClickError("verify failed")
|
||||
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", boom)
|
||||
r = client.post("/api/v1/auth/jverify-login", json={"login_token": "tok"})
|
||||
assert r.status_code == 502
|
||||
@@ -0,0 +1,74 @@
|
||||
"""一键登录换号门面 oneclick 单测:按 provider 分派到极光/阿里云, 统一异常。
|
||||
|
||||
纯函数(monkeypatch 掉两家的 verify_and_get_phone), 不碰 DB。本机可跑:
|
||||
SGB_TEST_SKIP_DB=1 pytest tests/test_oneclick.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.integrations import aliyun_onekey, jiguang, oneclick
|
||||
|
||||
|
||||
def test_dispatch_defaults_to_jiguang_when_blank(monkeypatch):
|
||||
"""provider 为空 = 老客户端 → 走极光(向后兼容)。"""
|
||||
seen = {}
|
||||
|
||||
def fake_jg(t):
|
||||
seen["jg"] = t
|
||||
return "13800000000"
|
||||
|
||||
monkeypatch.setattr(jiguang, "verify_and_get_phone", fake_jg)
|
||||
assert oneclick.verify_and_get_phone("", "tok") == "13800000000"
|
||||
assert seen["jg"] == "tok"
|
||||
|
||||
|
||||
def test_dispatch_jiguang_explicit(monkeypatch):
|
||||
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13811111111")
|
||||
assert oneclick.verify_and_get_phone("jiguang", "tok") == "13811111111"
|
||||
|
||||
|
||||
def test_dispatch_aliyun(monkeypatch):
|
||||
seen = {}
|
||||
|
||||
def fake_al(t):
|
||||
seen["al"] = t
|
||||
return "13822222222"
|
||||
|
||||
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", fake_al)
|
||||
assert oneclick.verify_and_get_phone("aliyun", "tok") == "13822222222"
|
||||
assert seen["al"] == "tok"
|
||||
|
||||
|
||||
def test_provider_is_case_insensitive(monkeypatch):
|
||||
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", lambda t: "13844444444")
|
||||
assert oneclick.verify_and_get_phone("Aliyun", "tok") == "13844444444"
|
||||
|
||||
|
||||
def test_unknown_provider_falls_back_to_jiguang(monkeypatch):
|
||||
"""未知 provider 兜底走极光(主家), 不因客户端传错值而拒登。"""
|
||||
monkeypatch.setattr(jiguang, "verify_and_get_phone", lambda t: "13833333333")
|
||||
assert oneclick.verify_and_get_phone("weird-value", "tok") == "13833333333"
|
||||
|
||||
|
||||
def test_jiguang_error_wrapped_as_oneclick_error(monkeypatch):
|
||||
def boom(_t):
|
||||
raise jiguang.JiguangError("jg down")
|
||||
|
||||
monkeypatch.setattr(jiguang, "verify_and_get_phone", boom)
|
||||
with pytest.raises(oneclick.OneClickError):
|
||||
oneclick.verify_and_get_phone("jiguang", "tok")
|
||||
|
||||
|
||||
def test_aliyun_error_wrapped_as_oneclick_error(monkeypatch):
|
||||
def boom(_t):
|
||||
raise aliyun_onekey.AliyunOneClickError("aliyun down")
|
||||
|
||||
monkeypatch.setattr(aliyun_onekey, "verify_and_get_phone", boom)
|
||||
with pytest.raises(oneclick.OneClickError):
|
||||
oneclick.verify_and_get_phone("aliyun", "tok")
|
||||
|
||||
|
||||
def test_mask_phone_reexported():
|
||||
"""auth 层只依赖 oneclick, 脱敏函数从门面转出。"""
|
||||
assert oneclick.mask_phone("13800138000") == "138****00"
|
||||
@@ -147,8 +147,8 @@ def test_wechat_bind_jverify_creates_account(client, monkeypatch) -> None:
|
||||
"""本机号(极光)绑定路径:verify_and_get_phone 拦掉,未占用 → 建号登入。"""
|
||||
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_5", "极光用户", None))
|
||||
phone = "13900139005"
|
||||
# 极光 loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部 from ...jiguang import verify_and_get_phone)
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", lambda token: phone)
|
||||
# loginToken→手机号 在 auth 模块命名空间打桩(auth.py 顶部改 from ...oneclick import verify_and_get_phone,2 参 provider/token)
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", lambda provider, token: phone)
|
||||
|
||||
ticket = client.post(
|
||||
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
|
||||
@@ -181,11 +181,11 @@ def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) ->
|
||||
|
||||
|
||||
def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
|
||||
"""极光核验失败(JiguangError)→ 502。"""
|
||||
"""换号失败(OneClickError)→ 502。"""
|
||||
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))
|
||||
|
||||
def _raise(token: str) -> str:
|
||||
raise auth.JiguangError("mock jg failure")
|
||||
def _raise(provider: str, token: str) -> str:
|
||||
raise auth.OneClickError("mock oneclick failure")
|
||||
|
||||
monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
|
||||
ticket = client.post(
|
||||
|
||||
Reference in New Issue
Block a user