Nguyen Minh Phuc
50c2fe2a1e
ci / lint (push) Successful in 1m19s
ci / unit (push) Failing after 1m2s
ci / integration (push) Has been skipped
ci / types (push) Successful in 1m37s
ci / security (push) Failing after 38s
ci / dockerfile (push) Successful in 14s
ci / image (api) (push) Has been skipped
ci / image (reconciler) (push) Has been skipped
ci / image (worker) (push) Has been skipped
ci / bump (push) Has been skipped
Complete working build of the system learn-python/ teaches. 164 tests, mypy --strict clean, domain coverage 99%.
41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""Unit tests for retry backoff math."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from hypothesis import given
|
|
from hypothesis import strategies as st
|
|
|
|
from svcforge_core.domain.backoff import next_attempt_at
|
|
|
|
T = datetime(2026, 7, 17, 12, 0, 0, tzinfo=UTC)
|
|
|
|
|
|
def test_attempt_zero_full_jitter_is_base() -> None:
|
|
assert next_attempt_at(0, now=T, rand=lambda: 1.0) == T + timedelta(seconds=2)
|
|
|
|
|
|
def test_zero_jitter_returns_now() -> None:
|
|
assert next_attempt_at(0, now=T, rand=lambda: 0.0) == T
|
|
|
|
|
|
def test_negative_attempt_raises_value_error() -> None:
|
|
with pytest.raises(ValueError, match="attempt"):
|
|
next_attempt_at(-1, now=T)
|
|
|
|
|
|
@given(attempt=st.integers(min_value=0, max_value=64), r=st.floats(min_value=0.0, max_value=1.0))
|
|
def test_delay_is_always_within_zero_and_cap(attempt: int, r: float) -> None:
|
|
cap_s = 300.0
|
|
got = next_attempt_at(attempt, now=T, cap_s=cap_s, rand=lambda: r)
|
|
delay = (got - T).total_seconds()
|
|
assert 0.0 <= delay <= cap_s
|
|
|
|
|
|
@given(attempt=st.integers(min_value=0, max_value=63))
|
|
def test_delay_is_non_decreasing_in_attempt(attempt: int) -> None:
|
|
def ceiling_of(a: int) -> float:
|
|
return (next_attempt_at(a, now=T, rand=lambda: 1.0) - T).total_seconds()
|
|
|
|
assert ceiling_of(attempt) <= ceiling_of(attempt + 1)
|