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
   9 import scapy.compat
  10
  11 from util import check_core_path, get_core_path
  12
  13
  14 class Hook(object):
  15     """
  16     Generic hooks before/after API/CLI calls
  17     """
  18
  19     def __init__(self, test):
  20         self.test = test
  21         self.logger = test.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                     scapy.compat.orb(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 ValueError:
  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, test):
  83         super(PollHook, self).__init__(test)
  84
  85     def on_crash(self, core_path):
  86         self.logger.error("Core file present, debug with: gdb %s %s",
  87                           self.test.vpp_bin, core_path)
  88         check_core_path(self.logger, core_path)
  89         self.logger.error("Running `file %s':", core_path)
  90         try:
  91             info = check_output(["file", core_path])
  92             self.logger.error(info)
  93         except CalledProcessError as e:
  94             self.logger.error(
  95                 "Subprocess returned with error running `file' utility on "
  96                 "core-file, "
  97                 "rc=%s",  e.returncode)
  98         except OSError as e:
  99             self.logger.error(
 100                 "Subprocess returned OS error running `file' utility on "
 101                 "core-file, "
 102                 "oserror=(%s) %s", e.errno, e.strerror)
 103         except Exception as e:
 104             self.logger.error(
 105                 "Subprocess returned unanticipated error running `file' "
 106                 "utility on core-file, "
 107                 "%s", e)
 108
 109     def poll_vpp(self):
 110         """
 111         Poll the vpp status and throw an exception if it's not running
 112         :raises VppDiedError: exception if VPP is not running anymore
 113         """
 114         if self.test.vpp_dead:
 115             # already dead, nothing to do
 116             return
 117
 118         self.test.vpp.poll()
 119         if self.test.vpp.returncode is not None:
 120             signaldict = dict(
 121                 (k, v) for v, k in reversed(sorted(signal.__dict__.items()))
 122                 if v.startswith('SIG') and not v.startswith('SIG_'))
 123
 124             if self.test.vpp.returncode in signaldict:
 125                 s = signaldict[abs(self.test.vpp.returncode)]
 126             else:
 127                 s = "unknown"
 128             msg = "VPP subprocess died unexpectedly with returncode %d [%s]." \
 129                   % (self.test.vpp.returncode, s)
 130             self.logger.critical(msg)
 131             core_path = get_core_path(self.test.tempdir)
 132             if os.path.isfile(core_path):
 133                 self.on_crash(core_path)
 134             self.test.vpp_dead = True
 135             raise VppDiedError(msg)
 136
 137     def before_api(self, api_name, api_args):
 138         """
 139         Check if VPP died before executing an API
 140
 141         :param api_name: name of the API
 142         :param api_args: tuple containing the API arguments
 143         :raises VppDiedError: exception if VPP is not running anymore
 144
 145         """
 146         super(PollHook, self).before_api(api_name, api_args)
 147         self.poll_vpp()
 148
 149     def before_cli(self, cli):
 150         """
 151         Check if VPP died before executing a CLI
 152
 153         :param cli: CLI string
 154         :raises Exception: exception if VPP is not running anymore
 155
 156         """
 157         super(PollHook, self).before_cli(cli)
 158         self.poll_vpp()
 159
 160
 161 class StepHook(PollHook):
 162     """ Hook which requires user to press ENTER before doing any API/CLI """
 163
 164     def __init__(self, test):
 165         self.skip_stack = None
 166         self.skip_num = None
 167         self.skip_count = 0
 168         super(StepHook, self).__init__(test)
 169
 170     def skip(self):
 171         if self.skip_stack is None:
 172             return False
 173         stack = traceback.extract_stack()
 174         counter = 0
 175         skip = True
 176         for e in stack:
 177             if counter > self.skip_num:
 178                 break
 179             if e[0] != self.skip_stack[counter][0]:
 180                 skip = False
 181             if e[1] != self.skip_stack[counter][1]:
 182                 skip = False
 183             counter += 1
 184         if skip:
 185             self.skip_count += 1
 186             return True
 187         else:
 188             print("%d API/CLI calls skipped in specified stack "
 189                   "frame" % self.skip_count)
 190             self.skip_count = 0
 191             self.skip_stack = None
 192             self.skip_num = None
 193             return False
 194
 195     def user_input(self):
 196         print('number\tfunction\tfile\tcode')
 197         counter = 0
 198         stack = traceback.extract_stack()
 199         for e in stack:
 200             print('%02d.\t%s\t%s:%d\t[%s]' % (counter, e[2], e[0], e[1], e[3]))
 201             counter += 1
 202         print(single_line_delim)
 203         print("You may enter a number of stack frame chosen from above")
 204         print("Calls in/below that stack frame will be not be stepped anymore")
 205         print(single_line_delim)
 206         while True:
 207             print("Enter your choice, if any, and press ENTER to continue "
 208                   "running the testcase...")
 209             choice = sys.stdin.readline().rstrip('\r\n')
 210             if choice == "":
 211                 choice = None
 212             try:
 213                 if choice is not None:
 214                     num = int(choice)
 215             except ValueError:
 216                 print("Invalid input")
 217                 continue
 218             if choice is not None and (num < 0 or num >= len(stack)):
 219                 print("Invalid choice")
 220                 continue
 221             break
 222         if choice is not None:
 223             self.skip_stack = stack
 224             self.skip_num = num
 225
 226     def before_cli(self, cli):
 227         """ Wait for ENTER before executing CLI """
 228         if self.skip():
 229             print("Skip pause before executing CLI: %s" % cli)
 230         else:
 231             print(double_line_delim)
 232             print("Test paused before executing CLI: %s" % cli)
 233             print(single_line_delim)
 234             self.user_input()
 235         super(StepHook, self).before_cli(cli)
 236
 237     def before_api(self, api_name, api_args):
 238         """ Wait for ENTER before executing API """
 239         if self.skip():
 240             print("Skip pause before executing API: %s (%s)"
 241                   % (api_name, api_args))
 242         else:
 243             print(double_line_delim)
 244             print("Test paused before executing API: %s (%s)"
 245                   % (api_name, api_args))
 246             print(single_line_delim)
 247             self.user_input()
 248         super(StepHook, self).before_api(api_name, api_args)