5 from log import RED, single_line_delim, double_line_delim
7 from subprocess import check_output, CalledProcessError
8 from util import check_core_path, get_core_path
17 Generic hooks before/after API/CLI calls
20 def __init__(self, test):
22 self.logger = test.logger
24 def before_api(self, api_name, api_args):
26 Function called before API call
27 Emit a debug message describing the API name and arguments
29 @param api_name: name of the API
30 @param api_args: tuple containing the API arguments
33 def _friendly_format(val):
34 if not isinstance(val, str):
37 return '{!s} ({!s})'.format(val, ':'.join(['{:02x}'.format(
38 ord(x)) for x in val]))
40 # we don't call test_type(val) because it is a packed value.
41 return '{!s} ({!s})'.format(val, str(
42 ipaddress.ip_address(val)))
43 except ipaddress.AddressValueError:
46 _args = ', '.join("{!s}={!r}".format(key, _friendly_format(val)) for
47 (key, val) in api_args.items())
48 self.logger.debug("API: %s (%s)" %
49 (api_name, _args), extra={'color': RED})
51 def after_api(self, api_name, api_args):
53 Function called after API call
55 @param api_name: name of the API
56 @param api_args: tuple containing the API arguments
60 def before_cli(self, cli):
62 Function called before CLI call
63 Emit a debug message describing the CLI
65 @param cli: CLI string
67 self.logger.debug("CLI: %s" % (cli), extra={'color': RED})
69 def after_cli(self, cli):
71 Function called after CLI call
76 class VppDiedError(Exception):
81 """ Hook which checks if the vpp subprocess is alive """
83 def __init__(self, test):
84 super(PollHook, self).__init__(test)
86 def on_crash(self, core_path):
87 self.logger.error("Core file present, debug with: gdb %s %s" %
88 (self.test.vpp_bin, core_path))
89 check_core_path(self.logger, core_path)
90 self.logger.error("Running `file %s':" % core_path)
92 info = check_output(["file", core_path])
93 self.logger.error(info)
94 except CalledProcessError as e:
96 "Could not run `file' utility on core-file, "
97 "rc=%s" % e.returncode)
101 Poll the vpp status and throw an exception if it's not running
102 :raises VppDiedError: exception if VPP is not running anymore
104 if self.test.vpp_dead:
105 # already dead, nothing to do
109 if self.test.vpp.returncode is not None:
111 (k, v) for v, k in reversed(sorted(signal.__dict__.items()))
112 if v.startswith('SIG') and not v.startswith('SIG_'))
114 if self.test.vpp.returncode in signaldict:
115 s = signaldict[abs(self.test.vpp.returncode)]
118 msg = "VPP subprocess died unexpectedly with returncode %d [%s]." \
119 % (self.test.vpp.returncode, s)
120 self.logger.critical(msg)
121 core_path = get_core_path(self.test.tempdir)
122 if os.path.isfile(core_path):
123 self.on_crash(core_path)
124 self.test.vpp_dead = True
125 raise VppDiedError(msg)
127 def before_api(self, api_name, api_args):
129 Check if VPP died before executing an API
131 :param api_name: name of the API
132 :param api_args: tuple containing the API arguments
133 :raises VppDiedError: exception if VPP is not running anymore
136 super(PollHook, self).before_api(api_name, api_args)
139 def before_cli(self, cli):
141 Check if VPP died before executing a CLI
143 :param cli: CLI string
144 :raises Exception: exception if VPP is not running anymore
147 super(PollHook, self).before_cli(cli)
151 class StepHook(PollHook):
152 """ Hook which requires user to press ENTER before doing any API/CLI """
154 def __init__(self, test):
155 self.skip_stack = None
158 super(StepHook, self).__init__(test)
161 if self.skip_stack is None:
163 stack = traceback.extract_stack()
167 if counter > self.skip_num:
169 if e[0] != self.skip_stack[counter][0]:
171 if e[1] != self.skip_stack[counter][1]:
178 print("%d API/CLI calls skipped in specified stack "
179 "frame" % self.skip_count)
181 self.skip_stack = None
185 def user_input(self):
186 print('number\tfunction\tfile\tcode')
188 stack = traceback.extract_stack()
190 print('%02d.\t%s\t%s:%d\t[%s]' % (counter, e[2], e[0], e[1], e[3]))
192 print(single_line_delim)
193 print("You may enter a number of stack frame chosen from above")
194 print("Calls in/below that stack frame will be not be stepped anymore")
195 print(single_line_delim)
197 print("Enter your choice, if any, and press ENTER to continue "
198 "running the testcase...")
199 choice = sys.stdin.readline().rstrip('\r\n')
203 if choice is not None:
206 print("Invalid input")
208 if choice is not None and (num < 0 or num >= len(stack)):
209 print("Invalid choice")
212 if choice is not None:
213 self.skip_stack = stack
216 def before_cli(self, cli):
217 """ Wait for ENTER before executing CLI """
219 print("Skip pause before executing CLI: %s" % cli)
221 print(double_line_delim)
222 print("Test paused before executing CLI: %s" % cli)
223 print(single_line_delim)
225 super(StepHook, self).before_cli(cli)
227 def before_api(self, api_name, api_args):
228 """ Wait for ENTER before executing API """
230 print("Skip pause before executing API: %s (%s)"
231 % (api_name, api_args))
233 print(double_line_delim)
234 print("Test paused before executing API: %s (%s)"
235 % (api_name, api_args))
236 print(single_line_delim)
238 super(StepHook, self).before_api(api_name, api_args)