copilot-swe-agent[bot]andgenichin
8b3e7ac753
Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>
61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from trex_core.application.balance_health import BalanceHealthTracker
|
|
|
|
|
|
def test_long_outage_backoff_is_capped_and_status_stays_valid():
|
|
now = {"t": datetime(2026, 9, 12, tzinfo=UTC)}
|
|
tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300)
|
|
|
|
backoffs: list[float] = []
|
|
for _ in range(20):
|
|
lifecycle, backoff = tracker.record_failure("execution", code="http_5xx")
|
|
backoffs.append(backoff)
|
|
assert lifecycle in {None, "detected", "ongoing"}
|
|
now["t"] += timedelta(seconds=301)
|
|
|
|
assert max(backoffs) == 300.0
|
|
state = tracker.status("execution", stale_after_sec=600)
|
|
assert state["state"] == "unavailable"
|
|
assert state["consecutive_failures"] == 20
|
|
assert state["failure_code"] == "http_5xx"
|
|
|
|
|
|
def test_long_outage_over_1100_failures_never_overflows_and_stays_capped():
|
|
now = {"t": datetime(2026, 9, 12, tzinfo=UTC)}
|
|
tracker = BalanceHealthTracker(clock=lambda: now["t"], base_backoff_sec=15, max_backoff_sec=300)
|
|
|
|
last_lifecycle = None
|
|
for i in range(1_100):
|
|
lifecycle, backoff = tracker.record_failure("execution", code="http_5xx")
|
|
last_lifecycle = lifecycle
|
|
if i >= 5:
|
|
assert backoff == 300.0
|
|
else:
|
|
assert backoff <= 300.0
|
|
now["t"] += timedelta(seconds=301)
|
|
|
|
state = tracker.status("execution", stale_after_sec=600)
|
|
assert state["state"] == "unavailable"
|
|
assert state["consecutive_failures"] == 1_100
|
|
assert state["failure_code"] == "http_5xx"
|
|
assert state["next_retry_at"] is not None
|
|
assert last_lifecycle in {None, "ongoing"}
|
|
|
|
|
|
def test_restart_resets_process_local_health_and_re_emits_detected():
|
|
now = {"t": datetime(2026, 9, 12, tzinfo=UTC)}
|
|
before_restart = BalanceHealthTracker(clock=lambda: now["t"])
|
|
before_restart.record_success("execution")
|
|
lifecycle, _ = before_restart.record_failure("execution", code="timeout")
|
|
assert lifecycle == "detected"
|
|
|
|
after_restart = BalanceHealthTracker(clock=lambda: now["t"])
|
|
lifecycle, _ = after_restart.record_failure("execution", code="timeout")
|
|
assert lifecycle == "detected"
|
|
state = after_restart.status("execution", stale_after_sec=600)
|
|
assert state["last_success_at"] is None
|
|
assert state["consecutive_failures"] == 1
|