tests/topotests/munet/testing/util.py

   1 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
   2 # SPDX-License-Identifier: GPL-2.0-or-later
   3 #
   4 # April 22 2022, Christian Hopps <chopps@gmail.com>
   5 #
   6 # Copyright (c) 2022, LabN Consulting, L.L.C
   7 #
   8 """Utility functions useful when using munet testing functionailty in pytest."""
   9 import asyncio
  10 import datetime
  11 import functools
  12 import logging
  13 import sys
  14 import time
  15
  16 from ..base import BaseMunet
  17 from ..cli import async_cli
  18
  19
  20 # =================
  21 # Utility Functions
  22 # =================
  23
  24
  25 async def async_pause_test(desc=""):
  26     isatty = sys.stdout.isatty()
  27     if not isatty:
  28         desc = f" for {desc}" if desc else ""
  29         logging.info("NO PAUSE on non-tty terminal%s", desc)
  30         return
  31
  32     while True:
  33         if desc:
  34             print(f"\n== PAUSING: {desc} ==")
  35         try:
  36             user = input('PAUSED, "cli" for CLI, "pdb" to debug, "Enter" to continue: ')
  37         except EOFError:
  38             print("^D...continuing")
  39             break
  40         user = user.strip()
  41         if user == "cli":
  42             await async_cli(BaseMunet.g_unet)
  43         elif user == "pdb":
  44             breakpoint()  # pylint: disable=W1515
  45         elif user:
  46             print(f'Unrecognized input: "{user}"')
  47         else:
  48             break
  49
  50
  51 def pause_test(desc=""):
  52     asyncio.run(async_pause_test(desc))
  53
  54
  55 def retry(retry_timeout, initial_wait=0, expected=True):
  56     """decorator: retry while functions return is not None or raises an exception.
  57
  58     * `retry_timeout`: Retry for at least this many seconds; after waiting
  59                        initial_wait seconds
  60     * `initial_wait`: Sleeps for this many seconds before first executing function
  61     * `expected`: if False then the return logic is inverted, except for exceptions,
  62                   (i.e., a non None ends the retry loop, and returns that value)
  63     """
  64
  65     def _retry(func):
  66         @functools.wraps(func)
  67         def func_retry(*args, **kwargs):
  68             retry_sleep = 2
  69
  70             # Allow the wrapped function's args to override the fixtures
  71             _retry_timeout = kwargs.pop("retry_timeout", retry_timeout)
  72             _expected = kwargs.pop("expected", expected)
  73             _initial_wait = kwargs.pop("initial_wait", initial_wait)
  74             retry_until = datetime.datetime.now() + datetime.timedelta(
  75                 seconds=_retry_timeout + _initial_wait
  76             )
  77
  78             if initial_wait > 0:
  79                 logging.info("Waiting for [%s]s as initial delay", initial_wait)
  80                 time.sleep(initial_wait)
  81
  82             while True:
  83                 seconds_left = (retry_until - datetime.datetime.now()).total_seconds()
  84                 try:
  85                     ret = func(*args, **kwargs)
  86                     if _expected and ret is None:
  87                         logging.debug("Function succeeds")
  88                         return ret
  89                     logging.debug("Function returned %s", ret)
  90                 except Exception as error:
  91                     logging.info("Function raised exception: %s", str(error))
  92                     ret = error
  93
  94                 if seconds_left < 0:
  95                     logging.info("Retry timeout of %ds reached", _retry_timeout)
  96                     if isinstance(ret, Exception):
  97                         raise ret
  98                     return ret
  99
 100                 logging.info(
 101                     "Sleeping %ds until next retry with %.1f retry time left",
 102                     retry_sleep,
 103                     seconds_left,
 104                 )
 105                 time.sleep(retry_sleep)
 106
 107         func_retry._original = func  # pylint: disable=W0212
 108         return func_retry
 109
 110     return _retry