55 lines
1.5 KiB
Python
55 lines
1.5 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 getdel(self, key):
|
|
return self.data.pop(key, None)
|
|
|
|
|
|
def test_issue_and_consume_once():
|
|
fake = FakeRedis()
|
|
with patch.object(svc, "get_sync_redis_client", return_value=fake):
|
|
ticket = svc.issue_ticket("u1", "sid-1", "rj-1")
|
|
payload = svc.consume_ticket(ticket)
|
|
assert payload["userId"] == "u1"
|
|
assert payload["sessionId"] == "sid-1"
|
|
assert payload["refreshJti"] == "rj-1"
|
|
assert svc.consume_ticket(ticket) is None
|
|
|
|
|
|
def test_consume_unknown_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_raises():
|
|
with patch.object(svc, "get_sync_redis_client", return_value=None):
|
|
with pytest.raises(RuntimeError, match="Redis"):
|
|
svc.issue_ticket("u1", "sid-1")
|
|
|
|
|
|
def test_consume_without_redis_raises():
|
|
with patch.object(svc, "get_sync_redis_client", return_value=None):
|
|
with pytest.raises(RuntimeError, match="Redis"):
|
|
svc.consume_ticket("any-ticket")
|