lb: remove api boilerplate
[vpp.git] / 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 framework
9 from log import RED, single_line_delim, double_line_delim
10 from util import check_core_path, get_core_path
11
12
13 class Hook(object):
14     """
15     Generic hooks before/after API/CLI calls
16     """
17
18     def __init__(self, test):
19         self.test = test
20         self.logger = test.logger
21
22     def before_api(self, api_name, api_args):
23         """
24         Function called before API call
25         Emit a debug message describing the API name and arguments
26
27         @param api_name: name of the API
28         @param api_args: tuple containing the API arguments
29         """
30
31         def _friendly_format(val):
32             if not isinstance(val, str):
33                 return val
34             if len(val) == 6:
35                 return '{!s} ({!s})'.format(val, ':'.join(['{:02x}'.format(
36                     scapy.compat.orb(x)) for x in val]))
37             try:
38                 # we don't call test_type(val) because it is a packed value.
39                 return '{!s} ({!s})'.format(val, str(
40                     ipaddress.ip_address(val)))
41             except ValueError:
42                 return val
43
44         _args = ', '.join("{!s}={!r}".format(key, _friendly_format(val)) for
45                           (key, val) in api_args.items())
46         self.logger.debug("API: %s (%s)" %
47                           (api_name, _args), extra={'color': RED})
48
49     def after_api(self, api_name, api_args):
50         """
51         Function called after API call
52
53         @param api_name: name of the API
54         @param api_args: tuple containing the API arguments
55         """
56         pass
57
58     def before_cli(self, cli):
59         """
60         Function called before CLI call
61         Emit a debug message describing the CLI
62
63         @param cli: CLI string
64         """
65         self.logger.debug("CLI: %s" % (cli), extra={'color': RED})
66
67     def after_cli(self, cli):
68         """
69         Function called after CLI call
70         """
71         pass
72
73
74 class PollHook(Hook):
75     """ Hook which checks if the vpp subprocess is alive """
76
77     def __init__(self, test):
78         super(PollHook, self).__init__(test)
79
80     def on_crash(self, core_path):
81         self.logger.error("Core file present, debug with: gdb %s %s",
82                           self.test.vpp_bin, core_path)
83         check_core_path(self.logger, core_path)
84         self.logger.error("Running `file %s':", core_path)
85         try:
86             info = check_output(["file", core_path])
87             self.logger.error(info)
88         except CalledProcessError as e:
89             self.logger.error(
90                 "Subprocess returned with error running `file' utility on "
91                 "core-file, "
92                 "rc=%s",  e.returncode)
93         except OSError as e:
94             self.logger.error(
95                 "Subprocess returned OS error running `file' utility on "
96                 "core-file, "
97                 "oserror=(%s) %s", e.errno, e.strerror)
98         except Exception as e:
99             self.logger.error(
100                 "Subprocess returned unanticipated error running `file' "
101                 "utility on core-file, "
102                 "%s", e)
103
104     def poll_vpp(self):
105         """
106         Poll the vpp status and throw an exception if it's not running
107         :raises VppDiedError: exception if VPP is not running anymore
108         """
109         if self.test.vpp_dead:
110             # already dead, nothing to do
111             return
112
113         self.test.vpp.poll()
114         if self.test.vpp.returncode is not None:
115             self.test.vpp_dead = True
116             raise framework.VppDiedError(rv=self.test.vpp.returncode)
117             core_path = get_core_path(self.test.tempdir)
118             if os.path.isfile(core_path):
119                 self.on_crash(core_path)
120
121     def before_api(self, api_name, api_args):
122         """
123         Check if VPP died before executing an API
124
125         :param api_name: name of the API
126         :param api_args: tuple containing the API arguments
127         :raises VppDiedError: exception if VPP is not running anymore
128
129         """
130         super(PollHook, self).before_api(api_name, api_args)
131         self.poll_vpp()
132
133     def before_cli(self, cli):
134         """
135         Check if VPP died before executing a CLI
136
137         :param cli: CLI string
138         :raises Exception: exception if VPP is not running anymore
139
140         """
141         super(PollHook, self).before_cli(cli)
142         self.poll_vpp()
143
144
145 class StepHook(PollHook):
146     """ Hook which requires user to press ENTER before doing any API/CLI """
147
148     def __init__(self, test):
149         self.skip_stack = None
150         self.skip_num = None
151         self.skip_count = 0
152         super(StepHook, self).__init__(test)
153
154     def skip(self):
155         if self.skip_stack is None:
156             return False
157         stack = traceback.extract_stack()
158         counter = 0
159         skip = True
160         for e in stack:
161             if counter > self.skip_num:
162                 break
163             if e[0] != self.skip_stack[counter][0]:
164                 skip = False
165             if e[1] != self.skip_stack[counter][1]:
166                 skip = False
167             counter += 1
168         if skip:
169             self.skip_count += 1
170             return True
171         else:
172             print("%d API/CLI calls skipped in specified stack "
173                   "frame" % self.skip_count)
174             self.skip_count = 0
175             self.skip_stack = None
176             self.skip_num = None
177             return False
178
179     def user_input(self):
180         print('number\tfunction\tfile\tcode')
181         counter = 0
182         stack = traceback.extract_stack()
183         for e in stack:
184             print('%02d.\t%s\t%s:%d\t[%s]' % (counter, e[2], e[0], e[1], e[3]))
185             counter += 1
186         print(single_line_delim)
187         print("You may enter a number of stack frame chosen from above")
188         print("Calls in/below that stack frame will be not be stepped anymore")
189         print(single_line_delim)
190         while True:
191             print("Enter your choice, if any, and press ENTER to continue "
192                   "running the testcase...")
193             choice = sys.stdin.readline().rstrip('\r\n')
194             if choice == "":
195                 choice = None
196             try:
197                 if choice is not None:
198                     num = int(choice)
199             except ValueError:
200                 print("Invalid input")
201                 continue
202             if choice is not None and (num < 0 or num >= len(stack)):
203                 print("Invalid choice")
204                 continue
205             break
206         if choice is not None:
207             self.skip_stack = stack
208             self.skip_num = num
209
210     def before_cli(self, cli):
211         """ Wait for ENTER before executing CLI """
212         if self.skip():
213             print("Skip pause before executing CLI: %s" % cli)
214         else:
215             print(double_line_delim)
216             print("Test paused before executing CLI: %s" % cli)
217             print(single_line_delim)
218             self.user_input()
219         super(StepHook, self).before_cli(cli)
220
221     def before_api(self, api_name, api_args):
222         """ Wait for ENTER before executing API """
223         if self.skip():
224             print("Skip pause before executing API: %s (%s)"
225                   % (api_name, api_args))
226         else:
227             print(double_line_delim)
228             print("Test paused before executing API: %s (%s)"
229                   % (api_name, api_args))
230             print(single_line_delim)
231             self.user_input()
232         super(StepHook, self).before_api(api_name, api_args)