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.test.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.test.vpp_dead:
 105             # already dead, nothing to do
 106             return
 107
 108         self.test.vpp.poll()
 109         if self.test.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.test.vpp.returncode in signaldict:
 115                 s = signaldict[abs(self.test.vpp.returncode)]
 116             else:
 117                 s = "unknown"
 118             msg = "VPP subprocess died unexpectedly with returncode %d [%s]." \
 119                   % (self.test.vpp.returncode, s)
 120             self.logger.critical(msg)
 121             core_path = get_core_path(self.test.tempdir)
 122             if os.path.isfile(core_path):
 123                 self.on_crash(core_path)
 124             self.test.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, test):
 155         self.skip_stack = None
 156         self.skip_num = None
 157         self.skip_count = 0
 158         super(StepHook, self).__init__(test)
 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)