vpp_config: Rework for Python2/3 compatibility. - vpp

diff options

author	Paul Vinciguerra <pvinci@vinciconsulting.com>	2018-12-19 02:05:25 -0800
committer	Dave Barach <openvpp@barachs.net>	2019-01-10 21:26:08 +0000
commit	339bc6b51dd1ea7d783b211c485881ccb495fd0d (patch)
tree	d910b6bf0dcc49b675ac406cffd3e576e9090061 /extras/vpp_config/vpplib/VppPCIUtil.py
parent	6ad7231c00287b7c1241c6e1dbbfda86f15798b4 (diff)

vpp_config: Rework for Python2/3 compatibility.

On ubuntu: $cd <basedir>/extras/vpp_config $./scripts/clean.sh $./scripts/cp-data.sh $sudo apt-get install python3-pip python3-setuptools $python3 -m pip install . $vpp-config Changes: * Convert to print() function. * raw_input changes. * floor division changes. * replace vpp-config.py with a setuptools 'vpp-config' entry_point. * replace netaddr with ipaddress from the standard library and backport. * .decode() subprocess.Popen's stdout because in python3 they are bytes. Change-Id: Id98894ee54e0c31a0ba0304134b159caef415705 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>

Diffstat (limited to 'extras/vpp_config/vpplib/VppPCIUtil.py')

-rw-r--r--

extras/vpp_config/vpplib/VppPCIUtil.py

1 files changed, 9 insertions, 7 deletions

import os import sys import traceback import ipaddress from subprocess import check_output, CalledProcessError import scapy.compat import framework from log import RED, single_line_delim, double_line_delim from util import check_core_path, get_core_path class Hook(object): """ 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", self.test.vpp_bin, 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 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 raise framework.VppDiedError(rv=self.test.vpp.returncode) core_path = get_core_path(self.test.tempdir) if os.path.isfile(core_path): self.on_crash(core_path) 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 super(StepHook, self).__init__(test) def skip(self): if self.skip_stack is None: return False 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(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: 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: 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)


context:
space:
mode: