import os import sys import traceback import ipaddress from subprocess import check_output, CalledProcessError import scapy.compat import asfframework from config import config from log import RED, single_line_delim, double_line_delim from util import check_core_path, get_core_path class Hook: """ Generic hooks before/after API/CLI calls """ def __init__(self, test): self.test = test self.logger = test.logger def before_api(self, api_name, api_args): """ Function called before API call Emit a debug message describing the API name and arguments @param api_name: name of the API @param api_args: tuple containing the API arguments """ def _friendly_format(val): if not isinstance(val, str): return val if len(val) == 6: return "{!s} ({!s})".format( val, ":".join(["{:02x}".format(scapy.compat.orb(x)) for x in val]) ) try: # we don't call test_type(val) because it is a packed value. return "{!s} ({!s})".format(val, str(ipaddress.ip_address(val))) except ValueError: return val _args = ", ".join( "{!s}={!r}".format(key, _friendly_format(val)) for (key, val) in api_args.items() ) self.logger.debug("API: %s (%s)" % (api_name, _args), extra={"color": RED}) def after_api(self, api_name, api_args): """ Function called after API call @param api_name: name of the API @param api_args: tuple containing the API arguments """ pass def before_cli(self, cli): """ Function called before CLI call Emit a debug message describing the CLI @param cli: CLI string """ self.logger.debug("CLI: %s" % (cli), extra={"color": RED}) def after_cli(self, cli): """ Function called after CLI call """ pass class PollHook(Hook): """Hook which checks if the vpp subprocess is alive""" def __init__(self, test): super(PollHook, self).__init__(test) def on_crash(self, core_path): self.logger.error( "Core file present, debug with: gdb %s %s", config.vpp, core_path ) check_core_path(self.logger, core_path) self.logger.error("Running `file %s':", core_path) try: info = check_output(["file", core_path]) self.logger.error(info) except CalledProcessError as e: self.logger.error( "Subprocess returned with error running `file' utility on " "core-file, " "rc=%s", e.returncode, ) except OSError as e: self.logger.error( "Subprocess returned OS error running `file' utility on " "core-file, " "oserror=(%s) %s", e.errno, e.strerror, ) except Exception as e: self.logger.error( "Subprocess returned unanticipated error running `file' " "utility on core-file, " "%s", e, ) def poll_vpp(self): """ Poll the vpp status and throw an exception if it's not running :raises VppDiedError: exception if VPP is not running anymore """ if not hasattr(self.test, "vpp") or self.test.vpp_dead: # already dead, nothing to do return self.test.vpp.poll() if self.test.vpp.returncode is not None: self.test.vpp_dead = True core_path = get_core_path(self.test.tempdir) if os.path.isfile(core_path): self.on_crash(core_path) raise asfframework.VppDiedError(rv=self.test.vpp.returncode) def before_api(self, api_name, api_args): """ Check if VPP died before executing an API :param api_name: name of the API :param api_args: tuple containing the API arguments :raises VppDiedError: exception if VPP is not running anymore """ super(PollHook, self).before_api(api_name, api_args) self.poll_vpp() def before_cli(self, cli): """ Check if VPP died before executing a CLI :param cli: CLI string :raises Exception: exception if VPP is not running anymore """ super(PollHook, self).before_cli(cli) self.poll_vpp() class StepHook(PollHook): """Hook which requires user to press ENTER before doing any API/CLI""" def __init__(self, test): self.skip_stack = None self.skip_num = None self.skip_count = 0 self.break_func = None super(StepHook, self).__init__(test) def skip(self): if self.break_func is not None: return self.should_skip_func_based() if self.skip_stack is not None: return self.should_skip_stack_based() def should_skip_func_based(self): stack = traceback.extract_stack() for e in stack: if e[2] == self.break_func: self.break_func = None return False return True def should_skip_stack_based(self): stack = traceback.extract_stack() counter = 0 skip = True for e in stack: if counter > self.skip_num: break if e[0] != self.skip_stack[counter][0]: skip = False if e[1] != self.skip_stack[counter][1]: skip = False counter += 1 if skip: self.skip_count += 1 return True else: print("%d API/CLI calls skipped in specified stack frame" % self.skip_count) self.skip_count = 0 self.skip_stack = None self.skip_num = None return False def user_input(self): print("number\tfunction\tfile\tcode") counter = 0 stack = traceback.extract_stack() for e in stack: print("%02d.\t%s\t%s:%d\t[%s]" % (counter, e[2], e[0], e[1], e[3])) counter += 1 print(single_line_delim) print("You may enter a number of stack frame chosen from above") print("Calls in/below that stack frame will be not be stepped anymore") print("Alternatively, enter a test function name to stop at") print(single_line_delim) while True: print( "Enter your choice, if any, and press ENTER to continue " "running the testcase..." ) choice = sys.stdin.readline().rstrip("\r\n") if choice == "": choice = None try: if choice is not None: num = int(choice) except ValueError: if choice.startswith("test_"): break print("Invalid input") continue if choice is not None and (num < 0 or num >= len(stack)): print("Invalid choice") continue break if choice is not None: if choice.startswith("test_"): self.break_func = choice else: self.break_func = None self.skip_stack = stack self.skip_num = num def before_cli(self, cli): """Wait for ENTER before executing CLI""" if self.skip(): print("Skip pause before executing CLI: %s" % cli) else: print(double_line_delim) print("Test paused before executing CLI: %s" % cli) print(single_line_delim) self.user_input() super(StepHook, self).before_cli(cli) def before_api(self, api_name, api_args): """Wait for ENTER before executing API""" if self.skip(): print("Skip pause before executing API: %s (%s)" % (api_name, api_args)) else: print(double_line_delim) print("Test paused before executing API: %s (%s)" % (api_name, api_args)) print(single_line_delim) self.user_input() super(StepHook, self).before_api(api_name, api_args)