62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
"""web wx-qr create/status API。"""
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.services import wx_qr_session
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def setup_function():
|
|
wx_qr_session._memory_clear_for_tests()
|
|
|
|
|
|
def test_wx_qr_create_and_poll_pending():
|
|
with patch("app.services.wx_qr_session._redis", return_value=None), patch(
|
|
"app.api.public.get_unlimited_wxacode",
|
|
return_value=b"\x89PNG\r\n\x1a\n",
|
|
):
|
|
r = client.post("/api/web/auth/wx-qr/create")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body.get("code") == 200
|
|
data = body["data"]
|
|
sid = data["sessionId"]
|
|
assert data["qrImageBase64"].startswith("data:image/")
|
|
st = client.get("/api/web/auth/wx-qr/status", params={"sessionId": sid})
|
|
assert st.status_code == 200
|
|
assert st.json()["data"]["status"] == "pending"
|
|
|
|
|
|
def test_web_qr_confirm_success_sets_session():
|
|
fake_user = SimpleNamespace(
|
|
id="u1", phone_no="13800138000", wx_openid=None, wx_unionid=None, status="1"
|
|
)
|
|
tokens = {
|
|
"token": "access",
|
|
"refreshToken": "refresh",
|
|
"expiresIn": 3600,
|
|
"userInfo": {"id": "u1"},
|
|
}
|
|
with patch("app.services.wx_qr_session._redis", return_value=None):
|
|
wx_qr_session._memory_clear_for_tests()
|
|
sid, _short = wx_qr_session.create_session(ttl_sec=60)
|
|
with patch("app.api.mini_auth.code2session", return_value={"openid": "oid1", "unionid": ""}), \
|
|
patch("app.api.mini_auth.get_phone_number", return_value="13800138000"), \
|
|
patch("app.api.mini_auth.resolve_wechat_login_user", return_value=(fake_user, None, None)), \
|
|
patch("app.api.mini_auth.issue_tokens_for_user", new=AsyncMock(return_value=tokens)):
|
|
r = client.post(
|
|
"/api/mini/auth/web-qr-confirm",
|
|
json={"sessionId": sid, "loginCode": "lc", "phoneCode": "pc"},
|
|
)
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body.get("code") == 200
|
|
assert body["data"].get("ok") is True
|
|
assert "token" not in (body.get("data") or {})
|
|
consumed = wx_qr_session.consume_success(sid)
|
|
assert consumed and consumed["token"] == "access"
|