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