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