aboutsummaryrefslogtreecommitdiffstats
path: root/src/vnet/l2/l2.api
blob: baf830fae3b5529a78af684ffbce2bbd5b778eea (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
/* Hey Emacs use -*- mode: C -*- */
/*
 * Copyright (c) 2016 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/** \brief Reply to l2_xconnect_dump 
    @param context - sender context which was passed in the request
    @param rx_sw_if_index - Receive interface index
    @param tx_sw_if_index - Transmit interface index
 */
define l2_xconnect_details
{
  u32 context;
  u32 rx_sw_if_index;
  u32 tx_sw_if_index;
};

/** \brief Dump L2 XConnects
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
*/
define l2_xconnect_dump
{
  u32 client_index;
  u32 context;
};

/** \brief l2 fib table details structure
    @param bd_id - the l2 fib / bridge domain table id
    @param mac - the entry's mac address
    @param sw_if_index - index of the interface
    @param static_mac - the entry is statically configured.
    @param filter_mac - the entry is a mac filter entry.
    @param bvi_mac - the mac address is a bridge virtual interface
*/
define l2_fib_table_details
{
  u32 context;
  u32 bd_id;
  u64 mac;
  u32 sw_if_index;
  u8 static_mac;
  u8 filter_mac;
  u8 bvi_mac;
};

/** \brief Dump l2 fib (aka bridge domain) table
    @param client_index - opaque cookie to identify the sender
    @param bd_id - the l2 fib / bridge domain table identifier
*/
define l2_fib_table_dump
{
  u32 client_index;
  u32 context;
  u32 bd_id;
};

/** \brief L2 fib clear table request, clear all mac entries in the l2 fib
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
*/
autoreply define l2_fib_clear_table
{
  u32 client_index;
  u32 context;
};

/** \brief L2 FIB flush all  entries
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
*/
autoreply define l2fib_flush_all
{
  u32 client_index;
  u32 context;
};

/** \brief L2 FIB flush bridge domain entries
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the entry's bridge domain id
*/
autoreply define l2fib_flush_bd
{
  u32 client_index;
  u32 context;
  u32 bd_id;
};

/** \brief L2 FIB flush interface entries
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the entry's bridge domain id
*/
autoreply define l2fib_flush_int
{
  u32 client_index;
  u32 context;
  u32 sw_if_index;
};

/** \brief L2 FIB add entry request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param mac - the entry's mac address
    @param bd_id - the entry's bridge domain id
    @param sw_if_index - the interface
    @param is_add - If non zero add the entry, else delete it
    @param static_mac - 
    @param filter_mac -
*/
autoreply define l2fib_add_del
{
  u32 client_index;
  u32 context;
  u64 mac;
  u32 bd_id;
  u32 sw_if_index;
  u8 is_add;
  u8 static_mac;
  u8 filter_mac;
  u8 bvi_mac;
};

/** \brief Register to recive L2 MAC events for leanred and aged MAC
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param learn_limit - MAC learn limit, 0 => default to 1000
    @param scan_delay - event scan delay in 10 msec unit, 0 => default to 100 msec
    @param max_macs_in_event - in units of 10 mac entries, 0 => default to 100 entries
    @param enable_disable - 1 => register for MAC events, 0 => cancel registration
    @param pid - sender's pid
*/
autoreply define want_l2_macs_events
{
  u32 client_index;
  u32 context;
  u32 learn_limit;
  u8  scan_delay;
  u8  max_macs_in_event;
  u8  enable_disable;
  u32 pid;
};

/** \brief Entry for learned or aged MAC in L2 MAC Events
    @param sw_if_index - sw_if_index in the domain
    @param mac_addr - mac_address
    @is_del - 0 => newly learned MAC, 1 => aged out MAC
*/
typeonly define mac_entry
{
  u32 sw_if_index;
  u8  mac_addr[6];
  u8  is_del;
  u8  spare;
};

/** \brief L2 MAC event for a list of learned or aged MACs
    @param client_index - opaque cookie to identify the sender
    @param pid - client pid registered to receive notification
    @param n_macs - number of learned/aged MAC enntries
    @param mac - array of learned/aged MAC entries
*/
define l2_macs_event
{
  u32 client_index;
  u32 pid;
  u32 n_macs;
  vl_api_mac_entry_t mac[n_macs];
};

/** \brief Set interface L2 flags (such as L2_LEARN, L2_FWD, 
    L2_FLOOD, L2_UU_FLOOD, or L2_ARP_TERM bits). This can be used
    to disable one or more of the features represented by the
    flag bits on an interface to override what is set as default
    for all interfaces in the bridge domain
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param sw_if_index - interface 
    @param is_set - if non-zero, set the bits, else clear them
    @param feature_bitmap - non-zero bits (as above) to set or clear
*/
define l2_flags
{
  u32 client_index;
  u32 context;
  u32 sw_if_index;
  u8 is_set;
  u32 feature_bitmap;
};

/** \brief Set interface L2 flags response
    @param context - sender context, to match reply w/ request
    @param retval - return code for the set l2 bits request
    @param resulting_feature_bitmap - the internal l2 feature bitmap after the request is implemented
*/
define l2_flags_reply
{
  u32 context;
  i32 retval;
  u32 resulting_feature_bitmap;
};

/** \brief L2 bridge domain set mac age
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the bridge domain to create
    @param mac_age - mac aging time in min, 0 for disabled
*/
autoreply define bridge_domain_set_mac_age
{
  u32 client_index;
  u32 context;
  u32 bd_id;
  u8 mac_age;
};

/** \brief L2 bridge domain add or delete request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the bridge domain to create
    @param flood - enable/disable bcast/mcast flooding in the bd
    @param uu_flood - enable/disable uknown unicast flood in the bd
    @param forward - enable/disable forwarding on all interfaces in the bd
    @param learn - enable/disable learning on all interfaces in the bd
    @param arp_term - enable/disable arp termination in the bd
    @param mac_age - mac aging time in min, 0 for disabled
    @param is_add - add or delete flag
*/
autoreply define bridge_domain_add_del
{
  u32 client_index;
  u32 context;
  u32 bd_id;
  u8 flood;
  u8 uu_flood;
  u8 forward;
  u8 learn;
  u8 arp_term;
  u8 mac_age;
  u8 bd_tag[64];
  u8 is_add;
};

/** \brief L2 bridge domain request operational state details
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the bridge domain id desired or ~0 to request all bds
*/
define bridge_domain_dump
{
  u32 client_index;
  u32 context;
  u32 bd_id;
};

/** \brief L2 bridge domain sw interface operational state response
    @param bd_id - the bridge domain id
    @param sw_if_index - sw_if_index in the domain
    @param shg - split horizon group for the interface
*/
typeonly manual_print manual_endian define bridge_domain_sw_if
{
  u32 context;
  u32 sw_if_index;
  u8 shg;
};

/** \brief L2 bridge domain operational state response
    @param bd_id - the bridge domain id
    @param flood - bcast/mcast flooding state on all interfaces in the bd
    @param uu_flood - uknown unicast flooding state on all interfaces in the bd
    @param forward - forwarding state on all interfaces in the bd
    @param learn - learning state on all interfaces in the bd
    @param arp_term - arp termination state on all interfaces in the bd
    @param mac_age - mac aging time in min, 0 for disabled
    @param n_sw_ifs - number of sw_if_index's in the domain
*/
manual_print manual_endian define bridge_domain_details
{
  u32 context;
  u32 bd_id;
  u8 flood;
  u8 uu_flood;
  u8 forward;
  u8 learn;
  u8 arp_term;
  u8 mac_age;
  u8 bd_tag[64];
  u32 bvi_sw_if_index;
  u32 n_sw_ifs;
  vl_api_bridge_domain_sw_if_t sw_if_details[n_sw_ifs];
};

/** \brief Set bridge flags (such as L2_LEARN, L2_FWD, L2_FLOOD, 
    L2_UU_FLOOD, or L2_ARP_TERM bits) request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param bd_id - the bridge domain to set the flags for
    @param is_set - if non-zero, set the flags, else clear them
    @param feature_bitmap - bits (as above) that are non-zero to set or clear
*/
define bridge_flags
{
  u32 client_index;
  u32 context;
  u32 bd_id;
  u8 is_set;
  u32 feature_bitmap;
};

/** \brief Set bridge flags response
    @param context - sender context, to match reply w/ request
    @param retval - return code for the set bridge flags request
    @param resulting_feature_bitmap - the internal L2 feature bitmap after the request is implemented
*/
define bridge_flags_reply
{
  u32 context;
  i32 retval;
  u32 resulting_feature_bitmap;
};

/** \brief L2 interface vlan tag rewrite configure request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param sw_if_index - interface the operation is applied to
    @param vtr_op - Choose from l2_vtr_op_t enum values
    @param push_dot1q - first pushed flag dot1q id set, else dot1ad
    @param tag1 - Needed for any push or translate vtr op
    @param tag2 - Needed for any push 2 or translate x-2 vtr ops
*/
autoreply define l2_interface_vlan_tag_rewrite
{
  u32 client_index;
  u32 context;
  u32 sw_if_index;
  u32 vtr_op;
  u32 push_dot1q;		// ethertype of first pushed tag is dot1q/dot1ad
  u32 tag1;			// first pushed tag
  u32 tag2;			// second pushed tag
};

/** \brief L2 interface pbb tag rewrite configure request
    @param client_index - opaque cookie to identify the sender
    @param context - sender context, to match reply w/ request
    @param sw_if_index - interface the operation is applied to
    @param vtr_op - Choose from l2_vtr_op_t enum values
    @param inner_tag - needed for translate_qinq vtr op only
    @param outer_tag - needed for translate_qinq vtr op only 
    @param b_dmac - B-tag remote mac address, needed for any push or translate_qinq vtr op
    @param b_smac - B-tag local mac address, needed for any push or translate qinq vtr op
    @param b_vlanid - B-tag vlanid, needed for any push or translate qinq vtr op
    @param i_sid - I-tag service id, needed for any push or translate qinq vtr op
*/
autoreply define l2_interface_pbb_tag_rewrite
{
  u32 client_index;
  u32 context;
  u32 sw_if_index;
  u32 vtr_op;
  u16 outer_tag;
  u8  b_dmac[6];
  u8  b_smac[6];
  u16 b_vlanid;
  u32 i_sid;
};

/*
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
f packet_infos(self): """List of packet infos""" return self._packet_infos @packet_infos.setter def packet_infos(self, value): self._packet_infos = value @classmethod def instance(cls): """Return the instance of this testcase""" return cls.test_instance @classmethod def set_debug_flags(cls, d): cls.debug_core = False cls.debug_gdb = False cls.debug_gdbserver = False if d is None: return dl = d.lower() if dl == "core": if resource.getrlimit(resource.RLIMIT_CORE)[0] <= 0: # give a heads up if this is actually useless cls.logger.critical("WARNING: core size limit is set 0, core " "files will NOT be created") cls.debug_core = True elif dl == "gdb": cls.debug_gdb = True elif dl == "gdbserver": cls.debug_gdbserver = True else: raise Exception("Unrecognized DEBUG option: '%s'" % d) @classmethod def setUpConstants(cls): """ Set-up the test case class based on environment variables """ try: s = os.getenv("STEP") cls.step = True if s.lower() in ("y", "yes", "1") else False except: cls.step = False try: d = os.getenv("DEBUG") except: d = None cls.set_debug_flags(d) cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp") cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH') debug_cli = "" if cls.step or cls.debug_gdb or cls.debug_gdbserver: debug_cli = "cli-listen localhost:5002" cls.vpp_cmdline = [cls.vpp_bin, "unix", "{", "nodaemon", debug_cli, "}", "api-segment", "{", "prefix", cls.shm_prefix, "}"] if cls.plugin_path is not None: cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path]) cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline) @classmethod def wait_for_enter(cls): if cls.debug_gdbserver: print(double_line_delim) print("Spawned GDB server with PID: %d" % cls.vpp.pid) elif cls.debug_gdb: print(double_line_delim) print("Spawned VPP with PID: %d" % cls.vpp.pid) else: cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid) return print(single_line_delim) print("You can debug the VPP using e.g.:") if cls.debug_gdbserver: print("gdb " + cls.vpp_bin + " -ex 'target remote localhost:7777'") print("Now is the time to attach a gdb by running the above " "command, set up breakpoints etc. and then resume VPP from " "within gdb by issuing the 'continue' command") elif cls.debug_gdb: print("gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid) print("Now is the time to attach a gdb by running the above " "command and set up breakpoints etc.") print(single_line_delim) raw_input("Press ENTER to continue running the testcase...") @classmethod def run_vpp(cls): cmdline = cls.vpp_cmdline if cls.debug_gdbserver: gdbserver = '/usr/bin/gdbserver' if not os.path.isfile(gdbserver) or \ not os.access(gdbserver, os.X_OK): raise Exception("gdbserver binary '%s' does not exist or is " "not executable" % gdbserver) cmdline = [gdbserver, 'localhost:7777'] + cls.vpp_cmdline cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline)) try: cls.vpp = subprocess.Popen(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1) except Exception as e: cls.logger.critical("Couldn't start vpp: %s" % e) raise cls.wait_for_enter() @classmethod def setUpClass(cls): """ Perform class setup before running the testcase Remove shared memory files, start vpp and connect the vpp-api """ cls.logger = getLogger(cls.__name__) cls.tempdir = tempfile.mkdtemp( prefix='vpp-unittest-' + cls.__name__ + '-') cls.shm_prefix = cls.tempdir.split("/")[-1] os.chdir(cls.tempdir) cls.logger.info("Temporary dir is %s, shm prefix is %s", cls.tempdir, cls.shm_prefix) cls.setUpConstants() cls.pg_streams = [] cls.packet_infos = {} cls.verbose = 0 cls.vpp_dead = False print(double_line_delim) print(colorize(getdoc(cls).splitlines()[0], YELLOW)) print(double_line_delim) # need to catch exceptions here because if we raise, then the cleanup # doesn't get called and we might end with a zombie vpp try: cls.run_vpp() cls.vpp_stdout_queue = Queue() cls.vpp_stdout_reader_thread = Thread(target=pump_output, args=( cls.vpp.stdout, cls.vpp_stdout_queue)) cls.vpp_stdout_reader_thread.start() cls.vpp_stderr_queue = Queue() cls.vpp_stderr_reader_thread = Thread(target=pump_output, args=( cls.vpp.stderr, cls.vpp_stderr_queue)) cls.vpp_stderr_reader_thread.start() cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls) if cls.step: hook = StepHook(cls) else: hook = PollHook(cls) cls.vapi.register_hook(hook) time.sleep(0.1) hook.poll_vpp() try: cls.vapi.connect() except: if cls.debug_gdbserver: print(colorize("You're running VPP inside gdbserver but " "VPP-API connection failed, did you forget " "to 'continue' VPP from within gdb?", RED)) raise except: t, v, tb = sys.exc_info() try: cls.quit() except: pass raise t, v, tb @classmethod def quit(cls): """ Disconnect vpp-api, kill vpp and cleanup shared memory files """ if (cls.debug_gdbserver or cls.debug_gdb) and hasattr(cls, 'vpp'): cls.vpp.poll() if cls.vpp.returncode is None: print(double_line_delim) print("VPP or GDB server is still running") print(single_line_delim) raw_input("When done debugging, press ENTER to kill the process" " and finish running the testcase...") if hasattr(cls, 'vpp'): if hasattr(cls, 'vapi'): cls.vapi.disconnect() cls.vpp.poll() if cls.vpp.returncode is None: cls.vpp.terminate() del cls.vpp if hasattr(cls, 'vpp_stdout_queue'): cls.logger.info(single_line_delim) cls.logger.info('VPP output to stdout while running %s:', cls.__name__) cls.logger.info(single_line_delim) f = open(cls.tempdir + '/vpp_stdout.txt', 'w') while not cls.vpp_stdout_queue.empty(): line = cls.vpp_stdout_queue.get_nowait() f.write(line) cls.logger.info('VPP stdout: %s' % line.rstrip('\n')) if hasattr(cls, 'vpp_stderr_queue'): cls.logger.info(single_line_delim) cls.logger.info('VPP output to stderr while running %s:', cls.__name__) cls.logger.info(single_line_delim) f = open(cls.tempdir + '/vpp_stderr.txt', 'w') while not cls.vpp_stderr_queue.empty(): line = cls.vpp_stderr_queue.get_nowait() f.write(line) cls.logger.info('VPP stderr: %s' % line.rstrip('\n')) cls.logger.info(single_line_delim) @classmethod def tearDownClass(cls): """ Perform final cleanup after running all tests in this test-case """ cls.quit() def tearDown(self): """ Show various debug prints after each test """ if not self.vpp_dead: self.logger.debug(self.vapi.cli("show trace")) self.logger.info(self.vapi.ppcli("show int")) self.logger.info(self.vapi.ppcli("show hardware")) self.logger.info(self.vapi.ppcli("show error")) self.logger.info(self.vapi.ppcli("show run")) def setUp(self): """ Clear trace before running each test""" self.vapi.cli("clear trace") # store the test instance inside the test class - so that objects # holding the class can access instance methods (like assertEqual) type(self).test_instance = self @classmethod def pg_enable_capture(cls, interfaces): """ Enable capture on packet-generator interfaces :param interfaces: iterable interface indexes """ for i in interfaces: i.enable_capture() @classmethod def pg_start(cls): """ Enable the packet-generator and send all prepared packet streams Remove the packet streams afterwards """ cls.vapi.cli("trace add pg-input 50") # 50 is maximum cls.vapi.cli('packet-generator enable') sleep(1) # give VPP some time to process the packets for stream in cls.pg_streams: cls.vapi.cli('packet-generator delete %s' % stream) cls.pg_streams = [] @classmethod def create_pg_interfaces(cls, interfaces): """ Create packet-generator interfaces :param interfaces: iterable indexes of the interfaces """ result = [] for i in interfaces: intf = VppPGInterface(cls, i) setattr(cls, intf.name, intf) result.append(intf) cls.pg_interfaces = result return result @classmethod def create_loopback_interfaces(cls, interfaces): """ Create loopback interfaces :param interfaces: iterable indexes of the interfaces """ result = [] for i in interfaces: intf = VppLoInterface(cls, i) setattr(cls, intf.name, intf) result.append(intf) cls.lo_interfaces = result return result @staticmethod def extend_packet(packet, size): """ Extend packet to given size by padding with spaces NOTE: Currently works only when Raw layer is present. :param packet: packet :param size: target size """ packet_len = len(packet) + 4 extend = size - packet_len if extend > 0: packet[Raw].load += ' ' * extend def add_packet_info_to_list(self, info): """ Add packet info to the testcase's packet info list :param info: packet info """ info.index = len(self.packet_infos) self.packet_infos[info.index] = info def create_packet_info(self, src_pg_index, dst_pg_index): """ Create packet info object containing the source and destination indexes and add it to the testcase's packet info list :param src_pg_index: source packet-generator index :param dst_pg_index: destination packet-generator index :returns: _PacketInfo object """ info = _PacketInfo() self.add_packet_info_to_list(info) info.src = src_pg_index info.dst = dst_pg_index return info @staticmethod def info_to_payload(info): """ Convert _PacketInfo object to packet payload :param info: _PacketInfo object :returns: string containing serialized data from packet info """ return "%d %d %d" % (info.index, info.src, info.dst) @staticmethod def payload_to_info(payload): """ Convert packet payload to _PacketInfo object :param payload: packet payload :returns: _PacketInfo object containing de-serialized data from payload """ numbers = payload.split() info = _PacketInfo() info.index = int(numbers[0]) info.src = int(numbers[1]) info.dst = int(numbers[2]) return info def get_next_packet_info(self, info): """ Iterate over the packet info list stored in the testcase Start iteration with first element if info is None Continue based on index in info if info is specified :param info: info or None :returns: next info in list or None if no more infos """ if info is None: next_index = 0 else: next_index = info.index + 1 if next_index == len(self.packet_infos): return None else: return self.packet_infos[next_index] def get_next_packet_info_for_interface(self, src_index, info): """ Search the packet info list for the next packet info with same source interface index :param src_index: source interface index to search for :param info: packet info - where to start the search :returns: packet info or None """ while True: info = self.get_next_packet_info(info) if info is None: return None if info.src == src_index: return info def get_next_packet_info_for_interface2(self, src_index, dst_index, info): """ Search the packet info list for the next packet info with same source and destination interface indexes :param src_index: source interface index to search for :param dst_index: destination interface index to search for :param info: packet info - where to start the search :returns: packet info or None """ while True: info = self.get_next_packet_info_for_interface(src_index, info) if info is None: return None if info.dst == dst_index: return info class VppTestResult(unittest.TestResult): """ @property result_string String variable to store the test case result string. @property errors List variable containing 2-tuples of TestCase instances and strings holding formatted tracebacks. Each tuple represents a test which raised an unexpected exception. @property failures List variable containing 2-tuples of TestCase instances and strings holding formatted tracebacks. Each tuple represents a test where a failure was explicitly signalled using the TestCase.assert*() methods. """ def __init__(self, stream, descriptions, verbosity): """ :param stream File descriptor to store where to report test results. Set to the standard error stream by default. :param descriptions Boolean variable to store information if to use test case descriptions. :param verbosity Integer variable to store required verbosity level. """ unittest.TestResult.__init__(self, stream, descriptions, verbosity) self.stream = stream self.descriptions = descriptions self.verbosity = verbosity self.result_string = None def addSuccess(self, test): """ Record a test succeeded result :param test: """ unittest.TestResult.addSuccess(self, test) self.result_string = colorize("OK", GREEN) def addSkip(self, test, reason): """ Record a test skipped. :param test: :param reason: """ unittest.TestResult.addSkip(self, test, reason) self.result_string = colorize("SKIP", YELLOW) def addFailure(self, test, err): """ Record a test failed result :param test: :param err: error message """ unittest.TestResult.addFailure(self, test, err) if hasattr(test, 'tempdir'): self.result_string = colorize("FAIL", RED) + \ ' [ temp dir used by test case: ' + test.tempdir + ' ]' else: self.result_string = colorize("FAIL", RED) + ' [no temp dir]' def addError(self, test, err): """ Record a test error result :param test: :param err: error message """ unittest.TestResult.addError(self, test, err) if hasattr(test, 'tempdir'): self.result_string = colorize("ERROR", RED) + \ ' [ temp dir used by test case: ' + test.tempdir + ' ]' else: self.result_string = colorize("ERROR", RED) + ' [no temp dir]' def getDescription(self, test): """ Get test description :param test: :returns: test description """ # TODO: if none print warning not raise exception short_description = test.shortDescription() if self.descriptions and short_description: return short_description else: return str(test) def startTest(self, test): """ Start a test :param test: """ unittest.TestResult.startTest(self, test) if self.verbosity > 0: self.stream.writeln( "Starting " + self.getDescription(test) + " ...") self.stream.writeln(single_line_delim) def stopTest(self, test): """ Stop a test :param test: """ unittest.TestResult.stopTest(self, test) if self.verbosity > 0: self.stream.writeln(single_line_delim) self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string)) self.stream.writeln(single_line_delim) else: self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string)) def printErrors(self): """ Print errors from running the test case """ self.stream.writeln() self.printErrorList('ERROR', self.errors) self.printErrorList('FAIL', self.failures) def printErrorList(self, flavour, errors): """ Print error list to the output stream together with error type and test case description. :param flavour: error type :param errors: iterable errors """ for test, err in errors: self.stream.writeln(double_line_delim) self.stream.writeln("%s: %s" % (flavour, self.getDescription(test))) self.stream.writeln(single_line_delim) self.stream.writeln("%s" % err) class VppTestRunner(unittest.TextTestRunner): """ A basic test runner implementation which prints results on standard error. """ @property def resultclass(self): """Class maintaining the results of the tests""" return VppTestResult def run(self, test): """ Run the tests :param test: """ print("Running tests using custom test runner") # debug message return super(VppTestRunner, self).run(test)