| 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) |
| 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): |
| |
| 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 |
| |
| 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() |
| |
| 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() |
