59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
"""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"
|