shanxia2.0/backend/tests/test_wx_qr_api.py

"""web wx-qr create/status API。"""
from types import SimpleNamespace
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from app.main import app
from app.services import wx_qr_session

client = TestClient(app)


@pytest.fixture(autouse=True)
def force_memory_backend(monkeypatch):
    monkeypatch.setattr(wx_qr_session, "_redis", lambda: None)
    wx_qr_session._memory_clear_for_tests()


def test_wx_qr_create_and_poll_pending():
    with 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"},
    }
    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_web_login_tokens", 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"