shanxia2.0/backend/tests/test_wx_qr_session.py

"""wx_qr_session 状态机与一次性消费。"""
import time

import pytest

from app.services import wx_qr_session as s


@pytest.fixture(autouse=True)
def force_memory_backend(monkeypatch):
    """本地 .env 常有 REDIS_HOST；单测强制走进程内存，避免连真实 Redis。"""
    monkeypatch.setattr(s, "_redis", lambda: None)
    s._memory_clear_for_tests()


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)
    data = s.get_session(sid)
    assert data is None or data["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"