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