SilverElixir commited on
Commit
40b6d04
·
1 Parent(s): 1c088fd

Merge circuit-breaker and proxy-rotation tests into parametrized cases

Browse files
Files changed (1) hide show
  1. tests/test_bot_transport.py +21 -40
tests/test_bot_transport.py CHANGED
@@ -26,21 +26,16 @@ def test_circuit_breaker_starts_closed():
26
  assert breaker.is_down(time.monotonic()) is False
27
 
28
 
29
- def test_circuit_breaker_does_not_trip_before_threshold():
 
30
  breaker = bot._TelegramProxyCircuitBreaker(cooldown_sec=20.0, trip_threshold=3)
31
- assert breaker.note_failure() is False
32
- assert breaker.note_failure() is False
33
- assert breaker.is_down(time.monotonic()) is False
34
-
35
-
36
- def test_circuit_breaker_trips_at_threshold():
37
- breaker = bot._TelegramProxyCircuitBreaker(cooldown_sec=20.0, trip_threshold=3)
38
- breaker.note_failure()
39
- breaker.note_failure()
40
- tripped = breaker.note_failure()
41
- assert tripped is True
42
- breaker.trip()
43
- assert breaker.is_down(time.monotonic()) is True
44
 
45
 
46
  def test_circuit_breaker_success_resets_consecutive_failures():
@@ -152,40 +147,26 @@ def test_rotate_telegram_proxy_noop_with_single_candidate():
152
  bot._telegram_proxy_idx = original_idx
153
 
154
 
155
- def test_rotate_telegram_proxy_switches_and_reports_lap_not_done():
 
 
 
 
 
 
156
  original_candidates = bot._TELEGRAM_PROXY_CANDIDATES
157
  original_idx = bot._telegram_proxy_idx
158
  original_base_url = bot.TELEGRAM_API_BASE_URL
159
  original_bot = bot.bot
160
  bot._TELEGRAM_PROXY_CANDIDATES = ["https://primary.example.com", "https://fallback.example.com"]
161
- bot._telegram_proxy_idx = 0
162
- bot.TELEGRAM_API_BASE_URL = "https://primary.example.com"
163
  bot.bot = None # без реального aiogram Bot — проверяем только URL-переключение
164
  try:
165
  switched = asyncio.run(bot._rotate_telegram_proxy())
166
- assert switched is True # ещё не замкнули круг — есть смысл пробовать сразу
167
- assert bot.TELEGRAM_API_BASE_URL == "https://fallback.example.com"
168
- assert bot._telegram_proxy_idx == 1
169
- finally:
170
- bot._TELEGRAM_PROXY_CANDIDATES = original_candidates
171
- bot._telegram_proxy_idx = original_idx
172
- bot.TELEGRAM_API_BASE_URL = original_base_url
173
- bot.bot = original_bot
174
-
175
-
176
- def test_rotate_telegram_proxy_reports_lap_done_after_full_cycle():
177
- original_candidates = bot._TELEGRAM_PROXY_CANDIDATES
178
- original_idx = bot._telegram_proxy_idx
179
- original_base_url = bot.TELEGRAM_API_BASE_URL
180
- original_bot = bot.bot
181
- bot._TELEGRAM_PROXY_CANDIDATES = ["https://primary.example.com", "https://fallback.example.com"]
182
- bot._telegram_proxy_idx = 1 # уже на резервном — следующий поворот вернёт на primary (idx 0)
183
- bot.TELEGRAM_API_BASE_URL = "https://fallback.example.com"
184
- bot.bot = None
185
- try:
186
- switched = asyncio.run(bot._rotate_telegram_proxy())
187
- assert switched is False # круг замкнулся — пора паузу включать
188
- assert bot._telegram_proxy_idx == 0
189
  finally:
190
  bot._TELEGRAM_PROXY_CANDIDATES = original_candidates
191
  bot._telegram_proxy_idx = original_idx
 
26
  assert breaker.is_down(time.monotonic()) is False
27
 
28
 
29
+ @pytest.mark.parametrize(("failures", "tripped"), [(2, False), (3, True)])
30
+ def test_circuit_breaker_trip_threshold(failures, tripped):
31
  breaker = bot._TelegramProxyCircuitBreaker(cooldown_sec=20.0, trip_threshold=3)
32
+ result = False
33
+ for _ in range(failures):
34
+ result = breaker.note_failure()
35
+ assert result is tripped
36
+ if tripped:
37
+ breaker.trip()
38
+ assert breaker.is_down(time.monotonic()) is tripped
 
 
 
 
 
 
39
 
40
 
41
  def test_circuit_breaker_success_resets_consecutive_failures():
 
147
  bot._telegram_proxy_idx = original_idx
148
 
149
 
150
+ @pytest.mark.parametrize(("start_idx", "expected_idx", "expected_switched", "expected_url"), [
151
+ # Ещё не замкнули круг — есть смысл пробовать сразу.
152
+ (0, 1, True, "https://fallback.example.com"),
153
+ # Уже на резервном — поворот вернёт на primary, круг замкнулся, пора пауза.
154
+ (1, 0, False, "https://primary.example.com"),
155
+ ])
156
+ def test_rotate_telegram_proxy_advances_and_reports_lap(start_idx, expected_idx, expected_switched, expected_url):
157
  original_candidates = bot._TELEGRAM_PROXY_CANDIDATES
158
  original_idx = bot._telegram_proxy_idx
159
  original_base_url = bot.TELEGRAM_API_BASE_URL
160
  original_bot = bot.bot
161
  bot._TELEGRAM_PROXY_CANDIDATES = ["https://primary.example.com", "https://fallback.example.com"]
162
+ bot._telegram_proxy_idx = start_idx
163
+ bot.TELEGRAM_API_BASE_URL = "https://primary.example.com" if start_idx == 0 else "https://fallback.example.com"
164
  bot.bot = None # без реального aiogram Bot — проверяем только URL-переключение
165
  try:
166
  switched = asyncio.run(bot._rotate_telegram_proxy())
167
+ assert switched is expected_switched
168
+ assert bot._telegram_proxy_idx == expected_idx
169
+ assert bot.TELEGRAM_API_BASE_URL == expected_url
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
170
  finally:
171
  bot._TELEGRAM_PROXY_CANDIDATES = original_candidates
172
  bot._telegram_proxy_idx = original_idx