File size: 4,316 Bytes
f4a3981 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | import asyncio
import time
import unittest
from unittest.mock import AsyncMock, patch
from app.utils import cache
from app.utils.cache import CircuitBreaker, get_cache, set_cache
class CircuitBreakerLogicTests(unittest.TestCase):
"""Pure state machine, no I/O."""
def test_closed_initially_allows_traffic(self):
b = CircuitBreaker(failure_threshold=3, cooldown_seconds=30)
self.assertTrue(b.allow())
def test_opens_after_consecutive_failures(self):
b = CircuitBreaker(failure_threshold=3, cooldown_seconds=30)
for _ in range(3):
b.record(success=False, duration=2.5)
self.assertFalse(b.allow(), "circuit must open after threshold failures")
def test_success_resets_the_failure_count(self):
b = CircuitBreaker(failure_threshold=3, cooldown_seconds=30)
b.record(success=False, duration=2.5)
b.record(success=False, duration=2.5)
b.record(success=True, duration=0.05) # healthy again
b.record(success=False, duration=2.5)
self.assertTrue(b.allow(), "one success must clear prior failures")
def test_slow_op_counts_as_a_failure(self):
# A 2.5s op that eventually succeeds is still a latency failure here.
b = CircuitBreaker(failure_threshold=2, cooldown_seconds=30,
slow_threshold=1.0)
b.record(success=True, duration=2.5)
b.record(success=True, duration=2.5)
self.assertFalse(b.allow())
def test_half_open_after_cooldown(self):
b = CircuitBreaker(failure_threshold=1, cooldown_seconds=0.05)
b.record(success=False, duration=2.5)
self.assertFalse(b.allow())
time.sleep(0.06)
self.assertTrue(b.allow(), "cooldown expiry must allow a probe")
class CacheCircuitTests(unittest.IsolatedAsyncioTestCase):
"""The cache must fail FAST to memory when Redis is slow, not per-call."""
async def test_slow_redis_does_not_stall_every_call(self):
cache._redis_client = None
cache._circuit.reset()
class SlowRedis:
async def get(self, key):
await asyncio.sleep(2.0)
return None
async def setex(self, key, ttl, value):
await asyncio.sleep(2.0)
with patch.object(cache, "get_redis", AsyncMock(return_value=SlowRedis())):
t0 = time.perf_counter()
for _ in range(5):
await get_cache("some:key")
first = time.perf_counter() - t0
# Circuit opens after threshold -> later calls skip Redis entirely.
for _ in range(5):
await get_cache("some:key")
later = time.perf_counter() - t0 - first
self.assertLess(
later, 1.0,
f"open circuit still paid Redis latency ({later:.2f}s for 5 calls)",
)
async def test_falls_back_to_memory_when_circuit_open(self):
cache._circuit.reset()
# Trip the breaker at the default threshold of 3.
for _ in range(3):
cache._circuit.record(success=False, duration=2.5)
self.assertFalse(cache._circuit.allow(), "circuit should be open")
with patch.object(cache, "get_redis",
AsyncMock(side_effect=AssertionError("must not be called"))):
await set_cache("mem:key", "value", 60)
got = await get_cache("mem:key")
self.assertEqual(got, "value", "memory fallback must still work")
async def test_healthy_redis_is_used_normally(self):
cache._redis_client = "sentinel"
cache._circuit.reset()
class FastRedis:
def __init__(self):
self.store = {}
async def get(self, key):
return self.store.get(key)
async def setex(self, key, ttl, value):
self.store[key] = value
fast = FastRedis()
with patch.object(cache, "get_redis", AsyncMock(return_value=fast)):
await set_cache("fast:key", "v1", 60)
got = await get_cache("fast:key")
self.assertEqual(got, "v1")
self.assertTrue(cache._circuit.allow())
async def tearDown(self):
cache._redis_client = None
cache._circuit.reset()
if __name__ == "__main__":
unittest.main() |