77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from app.services import mp_webview_ticket as svc
|
|
|
|
|
|
class FakeRedis:
|
|
def __init__(self):
|
|
self.data = {}
|
|
|
|
def set(self, key, value, ex=None, nx=False):
|
|
if nx and key in self.data:
|
|
return False
|
|
self.data[key] = value
|
|
return True
|
|
|
|
def get(self, key):
|
|
return self.data.get(key)
|
|
|
|
def delete(self, key):
|
|
return 1 if self.data.pop(key, None) is not None else 0
|
|
|
|
def eval(self, script, numkeys, *keys_and_args):
|
|
key = keys_and_args[0]
|
|
val = self.data.get(key)
|
|
if val is not None:
|
|
self.data.pop(key, None)
|
|
return val
|
|
|
|
|
|
def test_issue_and_consume_signed():
|
|
ticket = svc.issue_ticket("u1", "sid-1", "rj-1")
|
|
assert ticket.startswith(svc.SIGNED_PREFIX)
|
|
payload = svc.consume_ticket(ticket)
|
|
assert payload["userId"] == "u1"
|
|
assert payload["sessionId"] == "sid-1"
|
|
assert payload["refreshJti"] == "rj-1"
|
|
# 签名票在 TTL 内可再次兑换（兼容 web-view 二次加载）
|
|
assert svc.consume_ticket(ticket)["userId"] == "u1"
|
|
|
|
|
|
def test_consume_unknown_uuid_returns_none():
|
|
fake = FakeRedis()
|
|
with patch.object(svc, "get_sync_redis_client", return_value=fake):
|
|
assert svc.consume_ticket("missing") is None
|
|
|
|
|
|
def test_issue_without_redis_still_signed():
|
|
with patch.object(svc, "get_sync_redis_client", return_value=None):
|
|
ticket = svc.issue_ticket("u1", "sid-1", "rj-1")
|
|
assert ticket.startswith(svc.SIGNED_PREFIX)
|
|
payload = svc.consume_ticket(ticket)
|
|
assert payload["userId"] == "u1"
|
|
assert payload["sessionId"] == "sid-1"
|
|
|
|
|
|
def test_consume_legacy_uuid_from_redis():
|
|
fake = FakeRedis()
|
|
tid = "AABBCCDDEEFF00112233445566778899"
|
|
fake.set(
|
|
svc._key(tid),
|
|
'{"userId":"u1","sessionId":"sid-1","refreshJti":"rj-1"}',
|
|
ex=60,
|
|
nx=True,
|
|
)
|
|
with patch.object(svc, "get_sync_redis_client", return_value=fake):
|
|
payload = svc.consume_ticket(tid)
|
|
assert payload["userId"] == "u1"
|
|
assert svc.consume_ticket(tid) is None
|
|
|
|
|
|
def test_consume_uuid_without_redis_raises():
|
|
with patch.object(svc, "get_sync_redis_client", return_value=None):
|
|
with pytest.raises(RuntimeError, match="Redis"):
|
|
svc.consume_ticket("AABBCCDDEEFF00112233445566778899")
|