test/hook.py

   1 import os
   2 import sys
   3 import traceback
   4 import ipaddress
   5 from subprocess import check_output, CalledProcessError
   6
   7 import scapy.compat
   8 import asfframework
   9 from config import config
  10 from log import RED, single_line_delim, double_line_delim
  11 from util import check_core_path, get_core_path
  12
  13
  14 class Hook:
  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(
  37                     val, ":".join(["{:02x}".format(scapy.compat.orb(x)) for x in val])
  38                 )
  39             try:
  40                 # we don't call test_type(val) because it is a packed value.
  41                 return "{!s} ({!s})".format(val, str(ipaddress.ip_address(val)))
  42             except ValueError:
  43                 return val
  44
  45         _args = ", ".join(
  46             "{!s}={!r}".format(key, _friendly_format(val))
  47             for (key, val) in api_args.items()
  48         )
  49         self.logger.debug("API: %s (%s)" % (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 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(
  84             "Core file present, debug with: gdb %s %s", config.vpp, core_path
  85         )
  86         check_core_path(self.logger, core_path)
  87         self.logger.error("Running `file %s':", core_path)
  88         try:
  89             info = check_output(["file", core_path])
  90             self.logger.error(info)
  91         except CalledProcessError as e:
  92             self.logger.error(
  93                 "Subprocess returned with error running `file' utility on "
  94                 "core-file, "
  95                 "rc=%s",
  96                 e.returncode,
  97             )
  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",
 103                 e.errno,
 104                 e.strerror,
 105             )
 106         except Exception as e:
 107             self.logger.error(
 108                 "Subprocess returned unanticipated error running `file' "
 109                 "utility on core-file, "
 110                 "%s",
 111                 e,
 112             )
 113
 114     def poll_vpp(self):
 115         """
 116         Poll the vpp status and throw an exception if it's not running
 117         :raises VppDiedError: exception if VPP is not running anymore
 118         """
 119         if not hasattr(self.test, "vpp") or self.test.vpp_dead:
 120             # already dead, nothing to do
 121             return
 122
 123         self.test.vpp.poll()
 124         if self.test.vpp.returncode is not None:
 125             self.test.vpp_dead = True
 126             core_path = get_core_path(self.test.tempdir)
 127             if os.path.isfile(core_path):
 128                 self.on_crash(core_path)
 129             raise asfframework.VppDiedError(rv=self.test.vpp.returncode)
 130
 131     def before_api(self, api_name, api_args):
 132         """
 133         Check if VPP died before executing an API
 134
 135         :param api_name: name of the API
 136         :param api_args: tuple containing the API arguments
 137         :raises VppDiedError: exception if VPP is not running anymore
 138
 139         """
 140         super(PollHook, self).before_api(api_name, api_args)
 141         self.poll_vpp()
 142
 143     def before_cli(self, cli):
 144         """
 145         Check if VPP died before executing a CLI
 146
 147         :param cli: CLI string
 148         :raises Exception: exception if VPP is not running anymore
 149
 150         """
 151         super(PollHook, self).before_cli(cli)
 152         self.poll_vpp()
 153
 154
 155 class StepHook(PollHook):
 156     """Hook which requires user to press ENTER before doing any API/CLI"""
 157
 158     def __init__(self, test):
 159         self.skip_stack = None
 160         self.skip_num = None
 161         self.skip_count = 0
 162         self.break_func = None
 163         super(StepHook, self).__init__(test)
 164
 165     def skip(self):
 166         if self.break_func is not None:
 167             return self.should_skip_func_based()
 168         if self.skip_stack is not None:
 169             return self.should_skip_stack_based()
 170
 171     def should_skip_func_based(self):
 172         stack = traceback.extract_stack()
 173         for e in stack:
 174             if e[2] == self.break_func:
 175                 self.break_func = None
 176                 return False
 177         return True
 178
 179     def should_skip_stack_based(self):
 180         stack = traceback.extract_stack()
 181         counter = 0
 182         skip = True
 183         for e in stack:
 184             if counter > self.skip_num:
 185                 break
 186             if e[0] != self.skip_stack[counter][0]:
 187                 skip = False
 188             if e[1] != self.skip_stack[counter][1]:
 189                 skip = False
 190             counter += 1
 191         if skip:
 192             self.skip_count += 1
 193             return True
 194         else:
 195             print("%d API/CLI calls skipped in specified stack frame" % self.skip_count)
 196             self.skip_count = 0
 197             self.skip_stack = None
 198             self.skip_num = None
 199             return False
 200
 201     def user_input(self):
 202         print("number\tfunction\tfile\tcode")
 203         counter = 0
 204         stack = traceback.extract_stack()
 205         for e in stack:
 206             print("%02d.\t%s\t%s:%d\t[%s]" % (counter, e[2], e[0], e[1], e[3]))
 207             counter += 1
 208         print(single_line_delim)
 209         print("You may enter a number of stack frame chosen from above")
 210         print("Calls in/below that stack frame will be not be stepped anymore")
 211         print("Alternatively, enter a test function name to stop at")
 212         print(single_line_delim)
 213         while True:
 214             print(
 215                 "Enter your choice, if any, and press ENTER to continue "
 216                 "running the testcase..."
 217             )
 218             choice = sys.stdin.readline().rstrip("\r\n")
 219             if choice == "":
 220                 choice = None
 221             try:
 222                 if choice is not None:
 223                     num = int(choice)
 224             except ValueError:
 225                 if choice.startswith("test_"):
 226                     break
 227                 print("Invalid input")
 228                 continue
 229             if choice is not None and (num < 0 or num >= len(stack)):
 230                 print("Invalid choice")
 231                 continue
 232             break
 233         if choice is not None:
 234             if choice.startswith("test_"):
 235                 self.break_func = choice
 236             else:
 237                 self.break_func = None
 238                 self.skip_stack = stack
 239                 self.skip_num = num
 240
 241     def before_cli(self, cli):
 242         """Wait for ENTER before executing CLI"""
 243         if self.skip():
 244             print("Skip pause before executing CLI: %s" % cli)
 245         else:
 246             print(double_line_delim)
 247             print("Test paused before executing CLI: %s" % cli)
 248             print(single_line_delim)
 249             self.user_input()
 250         super(StepHook, self).before_cli(cli)
 251
 252     def before_api(self, api_name, api_args):
 253         """Wait for ENTER before executing API"""
 254         if self.skip():
 255             print("Skip pause before executing API: %s (%s)" % (api_name, api_args))
 256         else:
 257             print(double_line_delim)
 258             print("Test paused before executing API: %s (%s)" % (api_name, api_args))
 259             print(single_line_delim)
 260             self.user_input()
 261         super(StepHook, self).before_api(api_name, api_args)