1
- """
2
- dvs_common contains common infrastructure for writing tests for the
3
- virtual switch.
4
- """
1
+ """Common infrastructure for writing VS tests."""
5
2
6
3
import collections
7
4
import time
8
5
6
+ from typing import Any , Callable , Tuple
7
+
9
8
_PollingConfig = collections .namedtuple ('PollingConfig' , 'polling_interval timeout strict' )
10
9
10
+
11
11
class PollingConfig (_PollingConfig ):
12
- """
13
- PollingConfig provides parameters that are used to control the behavior
14
- for polling functions.
15
-
16
- Params:
17
- polling_interval (int): How often to poll, in seconds.
18
- timeout (int): The maximum amount of time to wait, in seconds.
19
- strict (bool): If the strict flag is set, reaching the timeout
20
- will cause tests to fail (e.g. assert False)
12
+ """PollingConfig provides parameters that are used to control polling behavior.
13
+
14
+ Attributes:
15
+ polling_interval (int): How often to poll, in seconds.
16
+ timeout (int): The maximum amount of time to wait, in seconds.
17
+ strict (bool): If the strict flag is set, reaching the timeout will cause tests to fail.
21
18
"""
22
19
23
- pass
24
20
25
- def wait_for_result (polling_function , polling_config ):
26
- """
27
- wait_for_result will periodically run `polling_function`
28
- using the parameters described in `polling_config` and return the
29
- output of the polling function.
30
-
31
- Args:
32
- polling_config (PollingConfig): The parameters to use to poll
33
- the db.
34
- polling_function (Callable[[], (bool, Any)]): The function being
35
- polled. The function takes no arguments and must return a
36
- status which indicates if the function was succesful or
37
- not, as well as some return value.
38
-
39
- Returns:
40
- (bool, Any): If the polling function succeeds, then this method
41
- will return True and the output of the polling function. If it
42
- does not succeed within the provided timeout, it will return False
43
- and whatever the output of the polling function was on the final
44
- attempt.
21
+ def wait_for_result (
22
+ polling_function : Callable [[], Tuple [bool , Any ]],
23
+ polling_config : PollingConfig ,
24
+ ) -> Tuple [bool , Any ]:
25
+ """Run `polling_function` periodically using the specified `polling_config`.
26
+
27
+ Args:
28
+ polling_function: The function being polled. The function cannot take any arguments and
29
+ must return a status which indicates if the function was succesful or not, as well as
30
+ some return value.
31
+ polling_config: The parameters to use to poll the polling function.
32
+
33
+ Returns:
34
+ If the polling function succeeds, then this method will return True and the output of the
35
+ polling function.
36
+
37
+ If it does not succeed within the provided timeout, it will return False and whatever the
38
+ output of the polling function was on the final attempt.
45
39
"""
46
40
if polling_config .polling_interval == 0 :
47
41
iterations = 1
@@ -57,6 +51,6 @@ def wait_for_result(polling_function, polling_config):
57
51
time .sleep (polling_config .polling_interval )
58
52
59
53
if polling_config .strict :
60
- assert False , "Operation timed out after {}s" . format ( polling_config .timeout )
54
+ assert False , f "Operation timed out after { polling_config .timeout } seconds"
61
55
62
56
return (False , result )
0 commit comments