Skip to content

Commit

Permalink
Add UT
Browse files Browse the repository at this point in the history
Signed-off-by: bingwang <[email protected]>
  • Loading branch information
bingwang-ms committed Oct 21, 2021
1 parent a804473 commit a501f56
Show file tree
Hide file tree
Showing 10 changed files with 1,660 additions and 56 deletions.
320 changes: 266 additions & 54 deletions tests/conftest.py

Large diffs are not rendered by default.

Empty file added tests/dvslib/__init__.py
Empty file.
543 changes: 543 additions & 0 deletions tests/dvslib/dvs_acl.py

Large diffs are not rendered by default.

55 changes: 55 additions & 0 deletions tests/dvslib/dvs_common.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
"""Common infrastructure for writing VS tests."""

import collections
import time


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


class PollingConfig(_PollingConfig):
"""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.
"""


def wait_for_result(
polling_function,
polling_config,
):
"""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
else:
iterations = int(polling_config.timeout // polling_config.polling_interval) + 1

for _ in range(iterations):
status, result = polling_function()

if status:
return (True, result)

time.sleep(polling_config.polling_interval)

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

return (False, result)
Loading

0 comments on commit a501f56

Please sign in to comment.