Skip to content

Commit

Permalink
[vs] Incorporate python3 best practices into DVSLib (sonic-net#1357)
Browse files Browse the repository at this point in the history
- Use fstrings instead of `format`
- Add type hints
- Simplify some of the comments

Signed-off-by: Danny Allen <daall@microsoft.com>
  • Loading branch information
daall authored Jul 23, 2020
1 parent d5a18a0 commit a9479e6
Show file tree
Hide file tree
Showing 2 changed files with 228 additions and 274 deletions.
64 changes: 29 additions & 35 deletions tests/dvslib/dvs_common.py
Original file line number Diff line number Diff line change
@@ -1,47 +1,41 @@
"""
dvs_common contains common infrastructure for writing tests for the
virtual switch.
"""
"""Common infrastructure for writing VS tests."""

import collections
import time

from typing import Any, Callable, Tuple

_PollingConfig = collections.namedtuple('PollingConfig', 'polling_interval timeout strict')


class PollingConfig(_PollingConfig):
"""
PollingConfig provides parameters that are used to control the behavior
for polling functions.
Params:
polling_interval (int): How often to poll, in seconds.
timeout (int): The maximum amount of time to wait, in seconds.
strict (bool): If the strict flag is set, reaching the timeout
will cause tests to fail (e.g. assert False)
"""PollingConfig provides parameters that are used to control polling behavior.
Attributes:
polling_interval (int): How often to poll, in seconds.
timeout (int): The maximum amount of time to wait, in seconds.
strict (bool): If the strict flag is set, reaching the timeout will cause tests to fail.
"""

pass

def wait_for_result(polling_function, polling_config):
"""
wait_for_result will periodically run `polling_function`
using the parameters described in `polling_config` and return the
output of the polling function.
Args:
polling_config (PollingConfig): The parameters to use to poll
the db.
polling_function (Callable[[], (bool, Any)]): The function being
polled. The function takes no arguments and must return a
status which indicates if the function was succesful or
not, as well as some return value.
Returns:
(bool, Any): If the polling function succeeds, then this method
will return True and the output of the polling function. If it
does not succeed within the provided timeout, it will return False
and whatever the output of the polling function was on the final
attempt.
def wait_for_result(
polling_function: Callable[[], Tuple[bool, Any]],
polling_config: PollingConfig,
) -> Tuple[bool, Any]:
"""Run `polling_function` periodically using the specified `polling_config`.
Args:
polling_function: The function being polled. The function cannot take any arguments and
must return a status which indicates if the function was succesful or not, as well as
some return value.
polling_config: The parameters to use to poll the polling function.
Returns:
If the polling function succeeds, then this method will return True and the output of the
polling function.
If it does not succeed within the provided timeout, it will return False and whatever the
output of the polling function was on the final attempt.
"""
if polling_config.polling_interval == 0:
iterations = 1
Expand All @@ -57,6 +51,6 @@ def wait_for_result(polling_function, polling_config):
time.sleep(polling_config.polling_interval)

if polling_config.strict:
assert False, "Operation timed out after {}s".format(polling_config.timeout)
assert False, f"Operation timed out after {polling_config.timeout} seconds"

return (False, result)
Loading

0 comments on commit a9479e6

Please sign in to comment.