shaguabijia-app-server/tests/test_wechat_login.py

"""微信登录 M1 测试:bind_ticket 令牌、wechat-login(openid 命中/未命中)、
bind-phone(建号/占用/令牌过期)。

沿用 tests/test_auth.py 风格:HTTP 走 client fixture;微信 code→openid 用 monkeypatch
拦掉(conftest 里 WECHAT_APP_ID/SECRET 是 dummy,不真连微信);短信走 SMS_MOCK(任意 6 位通过)。
"""
from __future__ import annotations

import pytest

from app.api.v1 import auth
from app.core import security
from app.integrations import wxpay


def _fake_userinfo(openid: str, nickname: str | None = "微信昵称", avatar: str | None = "http://x/a.png"):
    """返回一个可传给 monkeypatch 的假 code_to_userinfo(忽略 code,固定返回给定 openid)。"""
    def _f(code: str) -> dict:
        return {"openid": openid, "nickname": nickname, "avatar_url": avatar, "raw": {}}
    return _f


# ===== Task 1: bind_ticket 令牌 =====

def test_bind_ticket_roundtrip() -> None:
    token = security.create_bind_ticket(openid="oid1", wechat_nickname="昵", wechat_avatar_url="http://a")
    claims = security.decode_bind_ticket(token)
    assert claims["openid"] == "oid1"
    assert claims["wnk"] == "昵"
    assert claims["wav"] == "http://a"


def test_bind_ticket_wrong_type_rejected() -> None:
    # 用 access token 冒充 bind_ticket → TokenError(typ 不匹配)
    access, _ = security.create_token(user_id=1, token_type="access")
    with pytest.raises(security.TokenError):
        security.decode_bind_ticket(access)


def test_bind_ticket_expired_rejected(monkeypatch) -> None:
    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
    token = security.create_bind_ticket(openid="oid", wechat_nickname=None, wechat_avatar_url=None)
    with pytest.raises(security.TokenError):
        security.decode_bind_ticket(token)


# ===== Task 2: wechat-login =====

def test_wechat_login_new_openid_returns_bind_ticket(client, monkeypatch) -> None:
    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_new_1", "小明", "http://x/m.png"))
    r = client.post("/api/v1/auth/wechat-login", json={"code": "wxcode1", "device_id": "devA"})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "need_bind_phone"
    assert body["bind_ticket"]
    assert body["wechat_nickname"] == "小明"
    assert body["wechat_avatar_url"] == "http://x/m.png"
    assert body["token"] is None


def test_wechat_login_invalid_code_returns_400(client, monkeypatch) -> None:
    def _raise(code: str) -> dict:
        raise ValueError("微信授权失败: invalid code")
    monkeypatch.setattr(wxpay, "code_to_userinfo", _raise)
    r = client.post("/api/v1/auth/wechat-login", json={"code": "bad", "device_id": "devA"})
    assert r.status_code == 400, r.text


# ===== Task 3: bind-phone/sms =====

def test_wechat_bind_sms_creates_account_then_openid_logs_in(client, monkeypatch) -> None:
    """未占用 → 建微信账号(channel=wechat,昵称头像取微信);再次同 openid 登录 → 直接登入同一账号。"""
    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_flow_2", "阿花", "http://x/h.png"))
    phone = "13900139002"

    # 1) 微信登录 → 未命中 → 拿 ticket
    r = client.post("/api/v1/auth/wechat-login", json={"code": "c1", "device_id": "devB"})
    ticket = r.json()["bind_ticket"]
    assert ticket

    # 2) 短信绑号(SMS_MOCK:任意 6 位通过)→ 建号 + 登入
    r = client.post(
        "/api/v1/auth/wechat/bind-phone/sms",
        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devB"},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "logged_in"
    user = body["token"]["user"]
    assert user["phone"] == phone
    assert user["register_channel"] == "wechat"
    assert user["nickname"] == "阿花"
    assert user["avatar_url"] == "http://x/h.png"
    uid = user["id"]

    # 3) 再次微信登录(同 openid)→ 命中 → 直接登入同一账号
    r = client.post("/api/v1/auth/wechat-login", json={"code": "c2", "device_id": "devB"})
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "logged_in"
    assert body["token"]["user"]["id"] == uid


def test_wechat_bind_sms_phone_occupied(client, monkeypatch) -> None:
    """手机号已被其他账号占用 → 返回 phone_occupied + 原账号信息(不建号)。"""
    phone = "13900139003"
    # 先用普通短信登录占用该手机号(register_channel=sms)
    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
    occupied_nickname = r.json()["user"]["nickname"]

    # 微信登录(新 openid)→ 未命中 → ticket
    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_occ_3"))
    ticket = client.post(
        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devC"}
    ).json()["bind_ticket"]

    # 绑同一手机号 → 占用
    r = client.post(
        "/api/v1/auth/wechat/bind-phone/sms",
        json={"bind_ticket": ticket, "phone": phone, "code": "123456", "device_id": "devC"},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "phone_occupied"
    assert body["token"] is None
    assert body["occupied_account"]["nickname"] == occupied_nickname
    assert body["occupied_account"]["avatar_url"] is None  # 短信注册账号无头像 → 序列化为 null
    assert body["occupied_account"]["created_at"]


def test_wechat_bind_sms_expired_ticket_returns_401(client, monkeypatch) -> None:
    """过期 bind_ticket → 401。"""
    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
    expired = security.create_bind_ticket(openid="openid_exp", wechat_nickname="x", wechat_avatar_url=None)
    r = client.post(
        "/api/v1/auth/wechat/bind-phone/sms",
        json={"bind_ticket": expired, "phone": "13900139009", "code": "123456", "device_id": "devD"},
    )
    assert r.status_code == 401, r.text


# ===== Task 4: bind-phone/jverify =====

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)

    ticket = client.post(
        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
    ).json()["bind_ticket"]

    r = client.post(
        "/api/v1/auth/wechat/bind-phone/jverify",
        json={"bind_ticket": ticket, "login_token": "jgtoken", "device_id": "devE"},
    )
    assert r.status_code == 200, r.text
    body = r.json()
    assert body["status"] == "logged_in"
    user = body["token"]["user"]
    assert user["phone"] == phone
    assert user["register_channel"] == "wechat"
    assert user["nickname"] == "极光用户"
    # 微信 userinfo 隐私脱敏 avatar=None → 头像为空(客户端兜底默认头像)
    assert user["avatar_url"] is None


def test_wechat_bind_jverify_expired_ticket_returns_401(client, monkeypatch) -> None:
    """过期 bind_ticket → 401(极光绑号路径,decode 先于极光核验)。"""
    monkeypatch.setattr(security.settings, "WECHAT_BIND_TICKET_EXPIRE_MINUTES", -1)
    expired = security.create_bind_ticket(openid="openid_jv_exp", wechat_nickname="x", wechat_avatar_url=None)
    r = client.post(
        "/api/v1/auth/wechat/bind-phone/jverify",
        json={"bind_ticket": expired, "login_token": "jgtoken", "device_id": "devE"},
    )
    assert r.status_code == 401, r.text


def test_wechat_bind_jverify_jiguang_error_returns_502(client, monkeypatch) -> None:
    """极光核验失败(JiguangError)→ 502。"""
    monkeypatch.setattr(wxpay, "code_to_userinfo", _fake_userinfo("openid_jv_err"))

    def _raise(token: str) -> str:
        raise auth.JiguangError("mock jg failure")

    monkeypatch.setattr(auth, "verify_and_get_phone", _raise)
    ticket = client.post(
        "/api/v1/auth/wechat-login", json={"code": "c", "device_id": "devE"}
    ).json()["bind_ticket"]
    r = client.post(
        "/api/v1/auth/wechat/bind-phone/jverify",
        json={"bind_ticket": ticket, "login_token": "badtoken", "device_id": "devE"},
    )
    assert r.status_code == 502, r.text