shanxia2.0/backend/tests/test_mp_webview_ticket.py

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")