|
1 | 1 | import random
|
2 |
| -from time import monotonic, sleep |
| 2 | +import sys |
| 3 | +from time import perf_counter, sleep |
3 | 4 | from typing import List, NoReturn, Tuple, Type
|
4 | 5 | from unittest.mock import Mock
|
5 | 6 |
|
@@ -65,39 +66,46 @@ def create_timestamped_callable(sleep_per_call: float = 0) -> Tuple[Mock, List[f
|
65 | 66 | timestamps = []
|
66 | 67 |
|
67 | 68 | def _raise_error() -> NoReturn:
|
68 |
| - timestamps.append(monotonic()) |
| 69 | + timestamps.append(perf_counter()) |
69 | 70 | if sleep_per_call:
|
70 | 71 | sleep(sleep_per_call)
|
71 | 72 | raise RuntimeError
|
72 | 73 |
|
73 | 74 | return Mock(wraps=_raise_error), timestamps
|
74 | 75 |
|
75 | 76 |
|
76 |
| -# Use multiple of 15ms as Windows' sleep is only accurate to 15ms. |
77 | 77 | @pytest.mark.parametrize("wait_duration", [0.015, 0.045, 0.15])
|
78 | 78 | def test_retry_wait(wait_duration: float) -> None:
|
79 | 79 | function, timestamps = create_timestamped_callable()
|
80 | 80 | # Only the first retry will be scheduled before the time limit is exceeded.
|
81 | 81 | wrapped = retry(wait=wait_duration, stop_after_delay=0.01)(function)
|
82 |
| - start_time = monotonic() |
| 82 | + start_time = perf_counter() |
83 | 83 | with pytest.raises(RuntimeError):
|
84 | 84 | wrapped()
|
85 | 85 | assert len(timestamps) == 2
|
86 |
| - assert timestamps[1] - start_time >= wait_duration |
| 86 | + # Windows' timers historically had poor resolution causing these tests |
| 87 | + # to fail randomly frequently. |
| 88 | + if sys.platform != "win32": |
| 89 | + # Add a margin of 10% to permit for unavoidable variation. |
| 90 | + assert timestamps[1] - start_time >= (wait_duration * 0.9) |
87 | 91 |
|
88 | 92 |
|
89 | 93 | @pytest.mark.parametrize(
|
90 |
| - "call_duration, max_allowed_calls", [(0.01, 10), (0.04, 3), (0.15, 1)] |
| 94 | + "call_duration, max_allowed_calls", [(0.01, 11), (0.04, 3), (0.15, 1)] |
91 | 95 | )
|
92 | 96 | def test_retry_time_limit(call_duration: float, max_allowed_calls: int) -> None:
|
93 | 97 | function, timestamps = create_timestamped_callable(sleep_per_call=call_duration)
|
94 | 98 | wrapped = retry(wait=0, stop_after_delay=0.1)(function)
|
95 | 99 |
|
96 |
| - start_time = monotonic() |
| 100 | + start_time = perf_counter() |
97 | 101 | with pytest.raises(RuntimeError):
|
98 | 102 | wrapped()
|
99 | 103 | assert len(timestamps) <= max_allowed_calls
|
100 |
| - assert all(t - start_time <= 0.1 for t in timestamps) |
| 104 | + # Windows' timers historically had poor resolution causing these tests |
| 105 | + # to fail randomly frequently. |
| 106 | + if sys.platform != "win32": |
| 107 | + # Add a margin of 10% to permit for unavoidable variation. |
| 108 | + assert all(t - start_time <= (0.1 * 1.1) for t in timestamps) |
101 | 109 |
|
102 | 110 |
|
103 | 111 | def test_retry_method() -> None:
|
|
0 commit comments