summaryrefslogtreecommitdiffstats
path: root/test/hook.py
AgeCommit message (Expand)AuthorFilesLines
2023-11-03tests: refactor asf framework codeDave Wallace1-2/+2
2022-09-27tests: disable failing tests on ubuntu-22.04 debian-11Dave Wallace1-1/+1
2022-05-10tests: replace pycodestyle with blackKlement Sekera1-29/+37
2022-03-27tests: fix core file messageDmitry Valter1-1/+2
2022-01-31tests: support skipping to test method with STEPKlement Sekera1-4/+24
2020-12-07tests: py2 cleanup - remove subclassing of objectPaul Vinciguerra1-1/+1
2019-06-24tests: refactor VppDiedError.Paul Vinciguerra1-20/+4
2019-03-27make-test: fix ValueError raised by hook in python3Naveen Joy1-1/+1
2019-03-11VPP-1508: Use scapy.compat to manage packet level library differences.Paul Vinciguerra1-1/+4
2019-03-11Test: Fix hook.py: AttributeErrorPaul Vinciguerra1-1/+1
2019-02-28Remove unused imports from hooksNaveen Joy1-4/+0
2019-02-26make test: Add exception handling around subprocess.Paul Vinciguerra1-5/+16
2019-02-04Fix inheritance problem in test/hook.py.Paul Vinciguerra1-16/+16
2019-02-01VTL: Fix issue with ipaddress library use under python2.Paul Vinciguerra1-0/+5
2018-12-13test/hook.py. Add human-friendly annotations to log msgs.Paul Vinciguerra1-3/+19
2018-10-10Setup, teardown, DEBUG=core, FAILFAST fixesjuraj.linkes1-21/+14
2018-08-30Fix the default step when using STEP=1 while testingjuraj.linkes1-1/+1
2018-08-23CSIT-1139: Implement parallel test executionjuraj.linkes1-5/+5
2018-03-31make test: print a warning in case a core_pattern contains a filter programAndrew Yourtchenko1-0/+2
2018-03-24make test: enhance core-file informationKlement Sekera1-2/+12
2018-03-23make test: code cleanupKlement Sekera1-3/+5
2018-03-14make test: early core detection, code cleanupKlement Sekera1-0/+2
2018-02-01IPv4/6 reassemblyKlement Sekera1-9/+5
2017-11-06make test: fix DEBUG=core errorKlement Sekera1-1/+1
2017-08-10make test: detect hung testsKlement Sekera1-22/+7
2017-01-11make test: improve documentation and PEP8 complianceKlement Sekera1-2/+2
2016-12-07BFD: basic asynchronous session up/downKlement Sekera1-8/+12
2016-11-28make test: detect early vpp crashKlement Sekera1-2/+1
2016-11-02Improve debug-cli in test frameworkKlement Sekera1-1/+1
2016-10-31add vpp debugging support to test frameworkKlement Sekera1-8/+104
2016-10-26refactor test frameworkKlement Sekera1-0/+128
def after_cli(self, cli): """ Function called after CLI call """ pass class VppDiedError(Exception): pass class PollHook(Hook): """ Hook which checks if the vpp subprocess is alive """ def __init__(self, testcase): self.vpp_dead = False self.testcase = testcase self.logger = testcase.logger def spawn_gdb(self, gdb_path, core_path): gdb_cmdline = gdb_path + ' ' + self.testcase.vpp_bin + ' ' + core_path gdb = pexpect.spawn(gdb_cmdline) gdb.interact() try: gdb.terminate(True) except: pass if gdb.isalive(): raise Exception("GDB refused to die...") def on_crash(self, core_path): if self.testcase.debug_core: gdb_path = '/usr/bin/gdb' if os.path.isfile(gdb_path) and os.access(gdb_path, os.X_OK): # automatically attach gdb self.spawn_gdb(gdb_path, core_path) return else: self.logger.error( "Debugger '%s' does not exist or is not an executable.." % gdb_path) self.logger.critical('core file present, debug with: gdb ' + self.testcase.vpp_bin + ' ' + core_path) 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.vpp_dead: # already dead, nothing to do return self.testcase.vpp.poll() if self.testcase.vpp.returncode is not None: signaldict = dict( (k, v) for v, k in reversed(sorted(signal.__dict__.items())) if v.startswith('SIG') and not v.startswith('SIG_')) msg = "VPP subprocess died unexpectedly with returncode %d [%s]" % ( self.testcase.vpp.returncode, signaldict[abs(self.testcase.vpp.returncode)]) self.logger.critical(msg) core_path = self.testcase.tempdir + '/core' if os.path.isfile(core_path): self.on_crash(core_path) self.testcase.vpp_dead = True raise VppDiedError(msg) def after_api(self, api_name, api_args): """ Check if VPP died after 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).after_api(api_name, api_args) self.poll_vpp() def after_cli(self, cli): """ Check if VPP died after executing a CLI :param cli: CLI string :raises Exception: exception if VPP is not running anymore """ super(PollHook, self).after_cli(cli) self.poll_vpp() class StepHook(PollHook): """ Hook which requires user to press ENTER before doing any API/CLI """ def __init__(self, testcase): self.skip_stack = None self.skip_num = None self.skip_count = 0 super(StepHook, self).__init__(testcase) 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 can 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: choice = raw_input("Enter your choice, if any, and press ENTER to " "continue running the testcase...") if choice == "": choice = None try: if choice is not None: num = int(choice) except: 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)