zkxw/backend/tests/test_wx_qr_api.py

"""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"