|
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 | +@pytest.mark.skipif( |
| 78 | + sys.platform == "win32", reason="Too flaky on Windows due to poor timer resolution" |
| 79 | +) |
77 | 80 | @pytest.mark.parametrize("wait_duration", [0.015, 0.045, 0.15]) |
78 | 81 | def test_retry_wait(wait_duration: float) -> None: |
79 | 82 | function, timestamps = create_timestamped_callable() |
80 | 83 | # Only the first retry will be scheduled before the time limit is exceeded. |
81 | 84 | wrapped = retry(wait=wait_duration, stop_after_delay=0.01)(function) |
82 | | - start_time = monotonic() |
| 85 | + start_time = perf_counter() |
83 | 86 | with pytest.raises(RuntimeError): |
84 | 87 | wrapped() |
85 | 88 | assert len(timestamps) == 2 |
86 | | - assert timestamps[1] - start_time >= wait_duration |
| 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 |
|
| 93 | +@pytest.mark.skipif( |
| 94 | + sys.platform == "win32", reason="Too flaky on Windows due to poor timer resolution" |
| 95 | +) |
89 | 96 | @pytest.mark.parametrize( |
90 | | - "call_duration, max_allowed_calls", [(0.01, 10), (0.04, 3), (0.15, 1)] |
| 97 | + "call_duration, max_allowed_calls", [(0.01, 11), (0.04, 3), (0.15, 1)] |
91 | 98 | ) |
92 | 99 | def test_retry_time_limit(call_duration: float, max_allowed_calls: int) -> None: |
93 | 100 | function, timestamps = create_timestamped_callable(sleep_per_call=call_duration) |
94 | 101 | wrapped = retry(wait=0, stop_after_delay=0.1)(function) |
95 | 102 |
|
96 | | - start_time = monotonic() |
| 103 | + start_time = perf_counter() |
97 | 104 | with pytest.raises(RuntimeError): |
98 | 105 | wrapped() |
99 | 106 | assert len(timestamps) <= max_allowed_calls |
100 | | - assert all(t - start_time <= 0.1 for t in timestamps) |
| 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