test/hook.py

   1 import signal
   2 import os
   3 import sys
   4 import traceback
   5 from log import RED, single_line_delim, double_line_delim
   6 import ipaddress
   7 from subprocess import check_output, CalledProcessError
   8 from util import check_core_path, get_core_path
   9 try:
  10     text_type = unicode
  11 except NameError:
  12     text_type = str
  13
  14
  15 class Hook(object):
  16     """
  17     Generic hooks before/after API/CLI calls
  18     """
  19
  20     def __init__(self, test):
  21         self.test = test
  22         self.logger = test.logger
  23
  24     def before_api(self, api_name, api_args):
  25         """
  26         Function called before API call
  27         Emit a debug message describing the API name and arguments
  28
  29         @param api_name: name of the API
  30         @param api_args: tuple containing the API arguments
  31         """
  32
  33         def _friendly_format(val):
  34             if not isinstance(val, str):
  35                 return val
  36             if len(val) == 6:
  37                 return '{!s} ({!s})'.format(val, ':'.join(['{:02x}'.format(
  38                     ord(x)) for x in val]))
  39             try:
  40                 # we don't call test_type(val) because it is a packed value.
  41                 return '{!s} ({!s})'.format(val, str(
  42                     ipaddress.ip_address(val)))
  43             except ipaddress.AddressValueError:
  44                 return val
  45
  46         _args = ', '.join("{!s}={!r}".format(key, _friendly_format(val)) for
  47                           (key, val) in api_args.items())
  48         self.logger.debug("API: %s (%s)" %
  49                           (api_name, _args), extra={'color': RED})
  50
  51     def after_api(self, api_name, api_args):
  52         """
  53         Function called after API call
  54
  55         @param api_name: name of the API
  56         @param api_args: tuple containing the API arguments
  57         """
  58         pass
  59
  60     def before_cli(self, cli):
  61         """
  62         Function called before CLI call
  63         Emit a debug message describing the CLI
  64
  65         @param cli: CLI string
  66         """
  67         self.logger.debug("CLI: %s" % (cli), extra={'color': RED})
  68
  69     def after_cli(self, cli):
  70         """
  71         Function called after CLI call
  72         """
  73         pass
  74
  75
  76 class VppDiedError(Exception):
  77     pass
  78
  79
  80 class PollHook(Hook):
  81     """ Hook which checks if the vpp subprocess is alive """
  82
  83     def __init__(self, test):
  84         super(PollHook, self).__init__(test)
  85
  86     def on_crash(self, core_path):
  87         self.logger.error("Core file present, debug with: gdb %s %s",
  88                           self.testcase.vpp_bin, core_path)
  89         check_core_path(self.logger, core_path)
  90         self.logger.error("Running `file %s':", core_path)
  91         try:
  92             info = check_output(["file", core_path])
  93             self.logger.error(info)
  94         except CalledProcessError as e:
  95             self.logger.error(
  96                 "Subprocess returned with error running `file' utility on "
  97                 "core-file, "
  98                 "rc=%s",  e.returncode)
  99         except OSError as e:
 100             self.logger.error(
 101                 "Subprocess returned OS error running `file' utility on "
 102                 "core-file, "
 103                 "oserror=(%s) %s", e.errno, e.strerror)
 104         except Exception as e:
 105             self.logger.error(
 106                 "Subprocess returned unanticipated error running `file' "
 107                 "utility on core-file, "
 108                 "%s", e)
 109
 110     def poll_vpp(self):
 111         """
 112         Poll the vpp status and throw an exception if it's not running
 113         :raises VppDiedError: exception if VPP is not running anymore
 114         """
 115         if self.test.vpp_dead:
 116             # already dead, nothing to do
 117             return
 118
 119         self.test.vpp.poll()
 120         if self.test.vpp.returncode is not None:
 121             signaldict = dict(
 122                 (k, v) for v, k in reversed(sorted(signal.__dict__.items()))
 123                 if v.startswith('SIG') and not v.startswith('SIG_'))
 124
 125             if self.test.vpp.returncode in signaldict:
 126                 s = signaldict[abs(self.test.vpp.returncode)]
 127             else:
 128                 s = "unknown"
 129             msg = "VPP subprocess died unexpectedly with returncode %d [%s]." \
 130                   % (self.test.vpp.returncode, s)
 131             self.logger.critical(msg)
 132             core_path = get_core_path(self.test.tempdir)
 133             if os.path.isfile(core_path):
 134                 self.on_crash(core_path)
 135             self.test.vpp_dead = True
 136             raise VppDiedError(msg)
 137
 138     def before_api(self, api_name, api_args):
 139         """
 140         Check if VPP died before executing an API
 141
 142         :param api_name: name of the API
 143         :param api_args: tuple containing the API arguments
 144         :raises VppDiedError: exception if VPP is not running anymore
 145
 146         """
 147         super(PollHook, self).before_api(api_name, api_args)
 148         self.poll_vpp()
 149
 150     def before_cli(self, cli):
 151         """
 152         Check if VPP died before executing a CLI
 153
 154         :param cli: CLI string
 155         :raises Exception: exception if VPP is not running anymore
 156
 157         """
 158         super(PollHook, self).before_cli(cli)
 159         self.poll_vpp()
 160
 161
 162 class StepHook(PollHook):
 163     """ Hook which requires user to press ENTER before doing any API/CLI """
 164
 165     def __init__(self, test):
 166         self.skip_stack = None
 167         self.skip_num = None
 168         self.skip_count = 0
 169         super(StepHook, self).__init__(test)
 170
 171     def skip(self):
 172         if self.skip_stack is None:
 173             return False
 174         stack = traceback.extract_stack()
 175         counter = 0
 176         skip = True
 177         for e in stack:
 178             if counter > self.skip_num:
 179                 break
 180             if e[0] != self.skip_stack[counter][0]:
 181                 skip = False
 182             if e[1] != self.skip_stack[counter][1]:
 183                 skip = False
 184             counter += 1
 185         if skip:
 186             self.skip_count += 1
 187             return True
 188         else:
 189             print("%d API/CLI calls skipped in specified stack "
 190                   "frame" % self.skip_count)
 191             self.skip_count = 0
 192             self.skip_stack = None
 193             self.skip_num = None
 194             return False
 195
 196     def user_input(self):
 197         print('number\tfunction\tfile\tcode')
 198         counter = 0
 199         stack = traceback.extract_stack()
 200         for e in stack:
 201             print('%02d.\t%s\t%s:%d\t[%s]' % (counter, e[2], e[0], e[1], e[3]))
 202             counter += 1
 203         print(single_line_delim)
 204         print("You may enter a number of stack frame chosen from above")
 205         print("Calls in/below that stack frame will be not be stepped anymore")
 206         print(single_line_delim)
 207         while True:
 208             print("Enter your choice, if any, and press ENTER to continue "
 209                   "running the testcase...")
 210             choice = sys.stdin.readline().rstrip('\r\n')
 211             if choice == "":
 212                 choice = None
 213             try:
 214                 if choice is not None:
 215                     num = int(choice)
 216             except ValueError:
 217                 print("Invalid input")
 218                 continue
 219             if choice is not None and (num < 0 or num >= len(stack)):
 220                 print("Invalid choice")
 221                 continue
 222             break
 223         if choice is not None:
 224             self.skip_stack = stack
 225             self.skip_num = num
 226
 227     def before_cli(self, cli):
 228         """ Wait for ENTER before executing CLI """
 229         if self.skip():
 230             print("Skip pause before executing CLI: %s" % cli)
 231         else:
 232             print(double_line_delim)
 233             print("Test paused before executing CLI: %s" % cli)
 234             print(single_line_delim)
 235             self.user_input()
 236         super(StepHook, self).before_cli(cli)
 237
 238     def before_api(self, api_name, api_args):
 239         """ Wait for ENTER before executing API """
 240         if self.skip():
 241             print("Skip pause before executing API: %s (%s)"
 242                   % (api_name, api_args))
 243         else:
 244             print(double_line_delim)
 245             print("Test paused before executing API: %s (%s)"
 246                   % (api_name, api_args))
 247             print(single_line_delim)
 248             self.user_input()
 249         super(StepHook, self).before_api(api_name, api_args)