Total Complexity | 3 |
Total Lines | 47 |
Duplicated Lines | 0 % |
Changes | 0 |
1 | """Test kytos.core.retry module.""" |
||
2 | # pylint: disable=no-self-use |
||
3 | |||
4 | from unittest.mock import MagicMock, patch |
||
5 | |||
6 | from tenacity import retry_if_exception_type, stop_after_attempt |
||
7 | |||
8 | from kytos.core.retry import before_sleep, for_all_methods, retries |
||
9 | |||
10 | |||
11 | @patch("kytos.core.retry.LOG") |
||
12 | def test_before_sleep(mock_log): |
||
13 | """Test before sleep.""" |
||
14 | state = MagicMock() |
||
15 | state.fn.__name__ = "some_name" |
||
16 | state.seconds_since_start = 1 |
||
17 | before_sleep(state) |
||
18 | assert mock_log.warning.call_count == 1 |
||
19 | |||
20 | |||
21 | def test_for_all_methods_retries(): |
||
22 | """test for_all_methods retries.""" |
||
23 | |||
24 | mock = MagicMock() |
||
25 | max_retries = 3 |
||
26 | |||
27 | @for_all_methods( |
||
28 | retries, |
||
29 | retry=retry_if_exception_type((ValueError,)), |
||
30 | stop=stop_after_attempt(max_retries), |
||
31 | reraise=True, |
||
32 | ) |
||
33 | class SomeClass: |
||
34 | """SomeClass.""" |
||
35 | |||
36 | def some_method(self, mock) -> None: |
||
37 | """some_method.""" |
||
38 | mock() |
||
39 | raise ValueError("some error") |
||
40 | |||
41 | some_class = SomeClass() |
||
42 | try: |
||
43 | some_class.some_method(mock) |
||
44 | except ValueError: |
||
45 | pass |
||
46 | assert mock.call_count == max_retries |
||
47 |