zkxw/backend/tests/test_wx_qr_session.py

"""wx_qr_session 状态机与一次性消费。"""
import time
from unittest.mock import patch

from app.services import wx_qr_session as s

_patcher = None


def setup_function():
    global _patcher
    s._memory_clear_for_tests()
    _patcher = patch.object(s, "_redis", return_value=None)
    _patcher.start()


def teardown_function():
    global _patcher
    if _patcher:
        _patcher.stop()
        _patcher = None


def test_create_pending_and_get():
    sid, short = s.create_session(ttl_sec=60)
    assert sid and short and len(short) <= 32
    data = s.get_session(sid)
    assert data["status"] == "pending"
    assert s.resolve_session_id(short) == sid


def test_mark_scanned():
    sid, _ = s.create_session(ttl_sec=60)
    assert s.mark_scanned(sid) is True
    assert s.get_session(sid)["status"] == "scanned"


def test_mark_success_consume_once():
    sid, _ = s.create_session(ttl_sec=60)
    payload = {"token": "t", "refreshToken": "r", "expiresIn": 1, "userInfo": {}}
    assert s.mark_success(sid, payload) is True
    first = s.consume_success(sid)
    assert first["token"] == "t"
    assert s.consume_success(sid) is None
    assert s.get_session(sid)["status"] == "consumed"


def test_mark_failed():
    sid, _ = s.create_session(ttl_sec=60)
    assert s.mark_failed(sid, "未找到员工") is True
    assert s.get_session(sid)["status"] == "failed"
    assert s.get_session(sid)["message"] == "未找到员工"


def test_expire_old_session():
    sid, _ = s.create_session(ttl_sec=1)
    time.sleep(1.1)
    assert s.get_session(sid) is None or s.get_session(sid)["status"] == "expired"


def test_invalidate():
    sid, _ = s.create_session(ttl_sec=60)
    s.invalidate(sid)
    data = s.get_session(sid)
    assert data is None or data["status"] == "expired"