SMANAGE/gvsds/tests/test_redis_lock.py

"""模块：redis_lock.py 单元测试

无 Redis 环境下验证降级行为：acquire 永远返回 True，
release 为 no-op，上下文管理器与装饰器均正常工作。
有 Redis 时由集成测试覆盖。
"""
import unittest
from unittest import mock

from django.test import SimpleTestCase

from gvsds import redis_lock


class HasRedisDetectionTestCase(SimpleTestCase):
    """_HasRedis 探测逻辑。"""

    def testHasRedis_CachesResult(self):
        # 重置缓存
        redis_lock._REDIS_OK = None
        result1 = redis_lock._HasRedis()
        result2 = redis_lock._HasRedis()
        self.assertEqual(result1, result2)
        # 结果应被缓存（_REDIS_OK 非 None）
        self.assertIsNotNone(redis_lock._REDIS_OK)

    def testHasRedis_NoRedis_ReturnsFalse(self):
        redis_lock._REDIS_OK = None
        with mock.patch.object(redis_lock, '_is_redis_backend', return_value=False):
            self.assertFalse(redis_lock._HasRedis())

    def testHasRedis_WithRedis_ReturnsTrue(self):
        redis_lock._REDIS_OK = None
        with mock.patch.object(redis_lock, '_is_redis_backend', return_value=True):
            self.assertTrue(redis_lock._HasRedis())


class RedisLockNoRedisTestCase(SimpleTestCase):
    """无 Redis 时降级为 no-op。"""

    def setUp(self):
        # 强制无 Redis 模式
        self._orig = redis_lock._REDIS_OK
        redis_lock._REDIS_OK = False

    def tearDown(self):
        redis_lock._REDIS_OK = self._orig

    def testAcquire_NoRedis_ReturnsTrue(self):
        lock = redis_lock.RedisLock('test-key')
        self.assertTrue(lock.acquire())
        self.assertTrue(lock.acquired)

    def testRelease_NoRedis_NoOp(self):
        lock = redis_lock.RedisLock('test-key')
        lock.acquire()
        # 无 Redis 时 release 为 no-op，不应抛异常
        lock.release()

    def testContextManager_NoRedis_AlwaysAcquired(self):
        with redis_lock.RedisLock('ctx-key') as acquired:
            self.assertTrue(acquired)

    def testContextManager_NoRedis_NoExceptionOnExit(self):
        # 即使块内抛异常，__exit__ 也不应传播
        with self.assertRaises(ValueError):
            with redis_lock.RedisLock('ctx-key') as acquired:
                self.assertTrue(acquired)
                raise ValueError('boom')


class RedisLockedDecoratorTestCase(SimpleTestCase):
    """redis_locked 装饰器。"""

    def setUp(self):
        self._orig = redis_lock._REDIS_OK
        redis_lock._REDIS_OK = False

    def tearDown(self):
        redis_lock._REDIS_OK = self._orig

    def testDecorator_StaticKey_ExecutesFunc(self):
        @redis_lock.redis_locked('static-key', timeout=10)
        def Task():
            return 'done'

        self.assertEqual(Task(), 'done')

    def testDecorator_KeyArgByIndex(self):
        @redis_lock.redis_locked('order', timeout=10, key_arg=0)
        def PayOrder(order_uuid):
            return f'paid:{order_uuid}'

        self.assertEqual(PayOrder('abc-123'), 'paid:abc-123')

    def testDecorator_KeyArgByName(self):
        @redis_lock.redis_locked('order', timeout=10, key_arg='order_uuid')
        def PayOrder(order_uuid=None):
            return f'paid:{order_uuid}'

        self.assertEqual(PayOrder(order_uuid='xyz'), 'paid:xyz')

    def testDecorator_KeyArgMissing_FallsBackToStaticKey(self):
        @redis_lock.redis_locked('order', timeout=10, key_arg='missing')
        def PayOrder(order_uuid):
            return f'paid:{order_uuid}'

        # missing 参数不存在 → 使用静态 key 'order'
        self.assertEqual(PayOrder('abc'), 'paid:abc')

    def testDecorator_PreservesFuncMetadata(self):
        @redis_lock.redis_locked('meta-key')
        def ImportantTask():
            """Important docstring."""

        self.assertEqual(ImportantTask.__name__, 'ImportantTask')
        self.assertEqual(ImportantTask.__doc__, 'Important docstring.')


class RedisLockWithRedisTestCase(SimpleTestCase):
    """有 Redis 时通过 mock cache 验证核心逻辑。"""

    def setUp(self):
        self._orig = redis_lock._REDIS_OK
        redis_lock._REDIS_OK = True

    def tearDown(self):
        redis_lock._REDIS_OK = self._orig

    def testAcquire_CacheAddSuccess_ReturnsTrue(self):
        lock = redis_lock.RedisLock('win-key', timeout=5)
        with mock.patch('gvsds.redis_lock.cache') as mock_cache:
            mock_cache.add.return_value = True
            self.assertTrue(lock.acquire())
            mock_cache.add.assert_called_once()
            args, kwargs = mock_cache.add.call_args
            self.assertIn('win-key', args[0])
            self.assertEqual(kwargs['timeout'], 5)

    def testAcquire_CacheAddFail_NoWait_ReturnsFalse(self):
        lock = redis_lock.RedisLock('lose-key', wait=False)
        with mock.patch('gvsds.redis_lock.cache') as mock_cache:
            mock_cache.add.return_value = False
            self.assertFalse(lock.acquire())

    def testRelease_TokenMatch_Deletes(self):
        lock = redis_lock.RedisLock('rel-key')
        lock._acquired = True
        with mock.patch('gvsds.redis_lock.cache') as mock_cache:
            mock_cache.get.return_value = lock.token
            lock.release()
            mock_cache.delete.assert_called_once()

    def testRelease_TokenMismatch_DoesNotDelete(self):
        lock = redis_lock.RedisLock('rel-key')
        lock._acquired = True
        with mock.patch('gvsds.redis_lock.cache') as mock_cache:
            mock_cache.get.return_value = 'other-token'
            lock.release()
            mock_cache.delete.assert_not_called()

    def testRelease_NotAcquired_NoOp(self):
        lock = redis_lock.RedisLock('rel-key')
        with mock.patch('gvsds.redis_lock.cache') as mock_cache:
            lock.release()
            mock_cache.get.assert_not_called()