Coverage for tests / test_circuit_breaker.py: 14%
142 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-11-26 11:34 +0800
1"""Tests for circuit breaker functionality."""
3import pytest
5from ss_utils_circuit_breaker import CircuitBreaker, CircuitBreakerError, CircuitState
8class TestCircuitBreaker:
9 """Tests for CircuitBreaker class."""
11 async def test_initial_state_is_closed(self, breaker: CircuitBreaker) -> None:
12 """Test that circuit breaker starts in closed state."""
13 assert breaker.state == CircuitState.CLOSED
14 assert breaker.failure_count == 0
16 async def test_successful_calls_keep_circuit_closed(self, breaker: CircuitBreaker) -> None:
17 """Test that successful calls don't affect circuit state."""
19 @breaker
20 async def successful_func() -> str:
21 return "success"
23 for _ in range(10):
24 result = await successful_func()
25 assert result == "success"
26 assert breaker.state == CircuitState.CLOSED
27 assert breaker.failure_count == 0
29 async def test_circuit_opens_after_threshold_failures(self, breaker: CircuitBreaker) -> None:
30 """Test that circuit opens after reaching failure threshold."""
32 @breaker
33 async def failing_func() -> None:
34 raise ValueError("Simulated failure")
36 # Trigger failures up to threshold
37 for i in range(breaker.failure_threshold):
38 with pytest.raises(ValueError):
39 await failing_func()
40 assert breaker.failure_count == i + 1
42 # Circuit should now be open
43 assert breaker.state == CircuitState.OPEN
45 # Next call should fail immediately with CircuitBreakerError
46 with pytest.raises(CircuitBreakerError) as exc_info:
47 await failing_func()
48 assert exc_info.value.breaker_name == breaker.name
50 async def test_circuit_breaker_error_contains_name(self, breaker: CircuitBreaker) -> None:
51 """Test that CircuitBreakerError includes breaker name."""
52 breaker.open()
54 @breaker
55 async def some_func() -> None:
56 pass
58 with pytest.raises(CircuitBreakerError) as exc_info:
59 await some_func()
61 assert breaker.name in str(exc_info.value)
62 assert exc_info.value.breaker_name == breaker.name
64 async def test_open_circuit_prevents_calls(self, breaker: CircuitBreaker) -> None:
65 """Test that open circuit prevents function execution."""
66 call_count = 0
68 @breaker
69 async def tracked_func() -> str:
70 nonlocal call_count
71 call_count += 1
72 return "called"
74 # Manually open the circuit
75 breaker.open()
77 # Attempt to call should fail immediately
78 with pytest.raises(CircuitBreakerError):
79 await tracked_func()
81 # Function should not have been called
82 assert call_count == 0
84 async def test_half_open_allows_test_call(self, breaker: CircuitBreaker) -> None:
85 """Test that half-open state allows one test call."""
86 breaker.open()
87 breaker.half_open()
89 @breaker
90 async def success_func() -> str:
91 return "success"
93 result = await success_func()
94 assert result == "success"
95 assert breaker.state == CircuitState.CLOSED
97 async def test_half_open_failure_reopens_circuit(self, breaker: CircuitBreaker) -> None:
98 """Test that failure in half-open state reopens circuit."""
99 breaker.open()
100 breaker.half_open()
102 @breaker
103 async def failing_func() -> None:
104 raise ValueError("Still failing")
106 with pytest.raises(ValueError):
107 await failing_func()
109 assert breaker.state == CircuitState.OPEN
111 async def test_reset_closes_circuit(self, breaker: CircuitBreaker) -> None:
112 """Test that reset returns circuit to closed state."""
113 # Open the circuit
114 breaker.open()
115 assert breaker.state == CircuitState.OPEN
117 # Reset should close it
118 breaker.reset()
119 assert breaker.state == CircuitState.CLOSED
120 assert breaker.failure_count == 0
122 async def test_close_resets_failure_count(self, breaker: CircuitBreaker) -> None:
123 """Test that closing circuit resets failure count."""
125 @breaker
126 async def failing_func() -> None:
127 raise ValueError("Failure")
129 # Accumulate some failures (but not enough to open)
130 for _ in range(breaker.failure_threshold - 1):
131 with pytest.raises(ValueError):
132 await failing_func()
134 assert breaker.failure_count == breaker.failure_threshold - 1
136 # Manually close
137 breaker.close()
138 assert breaker.failure_count == 0
140 async def test_success_resets_failure_count(self, breaker: CircuitBreaker) -> None:
141 """Test that successful call resets failure count."""
143 @breaker
144 async def maybe_fail(should_fail: bool) -> str:
145 if should_fail:
146 raise ValueError("Failure")
147 return "success"
149 # Accumulate some failures
150 for _ in range(breaker.failure_threshold - 1):
151 with pytest.raises(ValueError):
152 await maybe_fail(should_fail=True)
154 assert breaker.failure_count == breaker.failure_threshold - 1
156 # Successful call should reset counter
157 result = await maybe_fail(should_fail=False)
158 assert result == "success"
159 assert breaker.failure_count == 0
161 async def test_expected_exception_filtering(self) -> None:
162 """Test that only expected exceptions trigger circuit breaker."""
163 breaker = CircuitBreaker(
164 name="filtered_breaker",
165 failure_threshold=2,
166 expected_exception=ValueError,
167 )
169 @breaker
170 async def multi_error(error_type: str) -> None:
171 if error_type == "value":
172 raise ValueError("Value error")
173 elif error_type == "type":
174 raise TypeError("Type error")
176 # ValueError should count as failure
177 with pytest.raises(ValueError):
178 await multi_error("value")
179 assert breaker.failure_count == 1
181 # TypeError should NOT count as failure (not in expected_exception)
182 with pytest.raises(TypeError):
183 await multi_error("type")
184 assert breaker.failure_count == 1 # Still 1, not 2
186 async def test_multiple_expected_exceptions(self) -> None:
187 """Test circuit breaker with multiple expected exception types."""
188 breaker = CircuitBreaker(
189 name="multi_exception_breaker",
190 failure_threshold=3,
191 expected_exception=(ValueError, ConnectionError),
192 )
194 @breaker
195 async def multi_fail(error_type: str) -> None:
196 if error_type == "value":
197 raise ValueError("Value error")
198 elif error_type == "connection":
199 raise ConnectionError("Connection error")
201 with pytest.raises(ValueError):
202 await multi_fail("value")
203 assert breaker.failure_count == 1
205 with pytest.raises(ConnectionError):
206 await multi_fail("connection")
207 assert breaker.failure_count == 2
209 async def test_decorated_function_preserves_return_value(self, breaker: CircuitBreaker) -> None:
210 """Test that decorated function returns correct value."""
212 @breaker
213 async def get_data() -> dict:
214 return {"key": "value", "number": 42}
216 result = await get_data()
217 assert result == {"key": "value", "number": 42}
219 async def test_decorated_function_preserves_arguments(self, breaker: CircuitBreaker) -> None:
220 """Test that decorated function receives correct arguments."""
222 @breaker
223 async def add(a: int, b: int, *, multiplier: int = 1) -> int:
224 return (a + b) * multiplier
226 result = await add(2, 3, multiplier=2)
227 assert result == 10
230class TestCircuitBreakerState:
231 """Tests for CircuitState enum."""
233 def test_state_values(self) -> None:
234 """Test that states have correct string values."""
235 assert CircuitState.CLOSED.value == "closed"
236 assert CircuitState.OPEN.value == "open"
237 assert CircuitState.HALF_OPEN.value == "half_open"
239 def test_state_is_string_enum(self) -> None:
240 """Test that CircuitState is a string enum."""
241 assert CircuitState.CLOSED == "closed"
242 assert CircuitState.OPEN == "open"
243 assert CircuitState.HALF_OPEN == "half_open"