guke
79 lines
3.2 KiB
Python
79 lines
3.2 KiB
Python
"""微信登录 M2 测试:conflict_ticket 令牌、继续绑定(attach/只登入)、换绑(建号+软删+30天限)。
|
|
|
|
沿用 tests/test_wechat_login.py 风格:HTTP 走 client;微信 code→openid 用 monkeypatch;
|
|
短信走 SMS_MOCK(任意 6 位过)。数据变更用"再走一遍 wechat-login 看 openid 落在哪个账号"做行为断言。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.api.v1 import auth # noqa: F401 (后续测试打桩 verify_and_get_phone 用)
|
|
from app.core import security
|
|
from app.integrations import wxpay
|
|
from app.models.phone_rebind_log import PhoneRebindLog
|
|
|
|
|
|
def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
|
|
def _f(code: str) -> dict:
|
|
return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
|
|
return _f
|
|
|
|
|
|
def _sms_occupy(client, phone: str) -> int:
|
|
"""用普通短信登录占用一个手机号(register_channel=sms),返回该账号 id。"""
|
|
assert client.post("/api/v1/auth/sms/send", json={"phone": phone}).status_code == 200
|
|
r = client.post("/api/v1/auth/sms/login", json={"phone": phone, "code": "123456"})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["user"]["id"]
|
|
|
|
|
|
def _occupy_via_conflict(client, monkeypatch, openid: str, phone: str, device_id: str) -> dict:
|
|
"""微信登录(新 openid)→ 绑同一手机号 → 返回 phone_occupied 的响应体(含 conflict_ticket)。"""
|
|
monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo(openid))
|
|
ticket = client.post(
|
|
"/api/v1/auth/wechat-login", json={"code": "c", "device_id": device_id}
|
|
).json()["bind_ticket"]
|
|
r = client.post(
|
|
"/api/v1/auth/wechat/bind-phone/sms",
|
|
json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": device_id},
|
|
)
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["status"] == "phone_occupied"
|
|
return body
|
|
|
|
|
|
# ===== Task 1: 模型可导入(建表由 conftest 的 create_all 完成) =====
|
|
|
|
def test_phone_rebind_log_model_importable() -> None:
|
|
assert PhoneRebindLog.__tablename__ == "phone_rebind_log"
|
|
|
|
|
|
# ===== Task 2: conflict_ticket 令牌 =====
|
|
|
|
def test_conflict_ticket_roundtrip() -> None:
|
|
token = security.create_conflict_ticket(
|
|
openid="oid1", wechat_nickname="昵", wechat_avatar_url="http://a", phone="13900139000"
|
|
)
|
|
claims = security.decode_conflict_ticket(token)
|
|
assert claims["openid"] == "oid1"
|
|
assert claims["wnk"] == "昵"
|
|
assert claims["wav"] == "http://a"
|
|
assert claims["phone"] == "13900139000"
|
|
|
|
|
|
def test_conflict_ticket_wrong_type_rejected() -> None:
|
|
# bind_ticket 冒充 conflict_ticket → TokenError(typ 不匹配)
|
|
bind = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
|
|
with pytest.raises(security.TokenError):
|
|
security.decode_conflict_ticket(bind)
|
|
|
|
|
|
def test_conflict_ticket_expired_rejected(monkeypatch) -> None:
|
|
monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
|
|
token = security.create_conflict_ticket(
|
|
openid="oid", wechat_nickname=None, wechat_avatar_url=None, phone="13900139000"
|
|
)
|
|
with pytest.raises(security.TokenError):
|
|
security.decode_conflict_ticket(token)
|