3 # Module to handle test case execution.
5 # The module provides a set of tools for constructing and running tests and
6 # representing the results.
9 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
14 from inspect import getdoc
16 from scapy.utils import wrpcap, rdpcap
17 from scapy.packet import Raw
19 ## Static variables to store color formatting strings.
21 # These variables (RED, GREEN, YELLOW and LPURPLE) are used to configure
22 # the color of the text to be printed in the terminal. Variable END is used
23 # to revert the text color to the default one.
30 ## Private class to create packet info object.
32 # Help process information about the next packet.
33 # Set variables to default values.
34 class _PacketInfo(object):
40 # Integer variable to store the index of the packet.
42 # Integer variable to store the index of the source packet generator
43 # interface of the packet.
45 # Integer variable to store the index of the destination packet generator
46 # interface of the packet.
48 # Object variable to store the copy of the former packet.
50 ## Subclass of the python unittest.TestCase class.
52 # This subclass is a base class for test cases that are implemented as classes.
53 # It provides methods to create and run test case.
54 class VppTestCase(unittest.TestCase):
56 ## Class method to set class constants necessary to run test case.
57 # @param cls The class pointer.
59 def setUpConstants(cls):
60 cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
61 cls.vpp_api_test_bin = os.getenv("VPP_TEST_API_TEST_BIN",
63 cls.vpp_cmdline = [cls.vpp_bin, "unix", "nodaemon", "api-segment", "{",
64 "prefix", "unittest", "}"]
65 cls.vpp_api_test_cmdline = [cls.vpp_api_test_bin, "chroot", "prefix",
68 cls.verbose = int(os.getenv("V", 0))
73 # String variable to store the path to vpp (vector packet processor).
74 ## @var vpp_api_test_bin
75 # String variable to store the path to vpp_api_test (vpp API test tool).
77 # List of command line attributes for vpp.
78 ## @var vpp_api_test_cmdline
79 # List of command line attributes for vpp_api_test.
81 # Integer variable to store required verbosity level.
83 ## Class method to start the test case.
84 # 1. Initiate test case constants and set test case variables to default
86 # 2. Remove files from the shared memory.
87 # 3. Start vpp as a subprocess.
88 # @param cls The class pointer.
100 print "=================================================================="
101 print YELLOW + getdoc(cls) + END
102 print "=================================================================="
103 os.system("rm -f /dev/shm/unittest-global_vm")
104 os.system("rm -f /dev/shm/unittest-vpe-api")
105 os.system("rm -f /dev/shm/unittest-db")
106 cls.vpp = subprocess.Popen(cls.vpp_cmdline, stderr=subprocess.PIPE)
108 # List variable to store packet-generator streams for interfaces.
110 # Dictionary variable to store host MAC addresses connected to packet
111 # generator interfaces.
113 # Dictionary variable to store host IPv4 addresses connected to packet
114 # generator interfaces.
116 # Dictionary variable to store host IPv6 addresses connected to packet
117 # generator interfaces.
119 # Dictionary variable to store VPP MAC addresses of the packet
120 # generator interfaces.
122 # Dictionary variable to store VPP IPv4 addresses of the packet
123 # generator interfaces.
125 # Dictionary variable to store VPP IPv6 addresses of the packet
126 # generator interfaces.
128 # Test case object variable to store file descriptor of running vpp
129 # subprocess with open pipe to the standard error stream per
130 # VppTestCase object.
132 ## Class method to do cleaning when all tests (test_) defined for
133 # VppTestCase class are finished.
134 # 1. Terminate vpp and kill all vpp instances.
135 # 2. Remove files from the shared memory.
136 # @param cls The class pointer.
141 os.system("rm -f /dev/shm/unittest-global_vm")
142 os.system("rm -f /dev/shm/unittest-vpe-api")
143 os.system("rm -f /dev/shm/unittest-db")
145 ## Class method to define tear down action of the VppTestCase class.
146 # @param cls The class pointer.
148 def tearDownClass(cls):
151 ## Method to define tear down VPP actions of the test case.
152 # @param self The object pointer.
154 self.cli(2, "show int")
155 self.cli(2, "show trace")
156 self.cli(2, "show hardware")
157 self.cli(2, "show ip arp")
158 self.cli(2, "show ip fib")
159 self.cli(2, "show error")
160 self.cli(2, "show run")
162 ## Method to define setup action of the test case.
163 # @param self The object pointer.
165 self.cli(2, "clear trace")
167 ## Class method to print logs.
168 # Based on set level of verbosity print text in the terminal.
169 # @param cls The class pointer.
170 # @param s String variable to store text to be printed.
171 # @param v Integer variable to store required level of verbosity.
173 def log(cls, s, v=1):
175 print "LOG: " + LPURPLE + s + END
177 ## Class method to execute api commands.
178 # Based on set level of verbosity print the output of the api command in
180 # @param cls The class pointer.
181 # @param s String variable to store api command string.
184 p = subprocess.Popen(cls.vpp_api_test_cmdline,
185 stdout=subprocess.PIPE,
186 stdin=subprocess.PIPE,
187 stderr=subprocess.PIPE)
189 print "API: " + RED + s + END
191 out = p.communicate()[0]
192 out = out.replace("vat# ", "", 2)
195 print YELLOW + out + END
197 # Object variable to store file descriptor of vpp_api_test subprocess
198 # with open pipes to the standard output, inputs and error streams.
200 # Tuple variable to store standard output of vpp_api_test subprocess
201 # where the string "vat# " is replaced by empty string later.
203 ## Class method to execute cli commands.
204 # Based on set level of verbosity of the log and verbosity defined by
205 # environmental variable execute the cli command and print the output in
207 # CLI command is executed via vpp API test tool (exec + cli_command)
208 # @param cls The class pointer.
209 # @param v Integer variable to store required level of verbosity.
210 # @param s String variable to store cli command string.
215 p = subprocess.Popen(cls.vpp_api_test_cmdline,
216 stdout=subprocess.PIPE,
217 stdin=subprocess.PIPE,
218 stderr=subprocess.PIPE)
220 print "CLI: " + RED + s + END
221 p.stdin.write('exec ' + s)
222 out = p.communicate()[0]
223 out = out.replace("vat# ", "", 2)
226 print YELLOW + out + END
228 # Object variable to store file descriptor of vpp_api_test subprocess
229 # with open pipes to the standard output, inputs and error streams.
231 # Tuple variable to store standard output of vpp_api_test subprocess
232 # where the string "vat# " is replaced by empty string later.
234 ## Class method to create incoming packet stream for the packet-generator
236 # Delete old /tmp/pgX_in.pcap file if exists and create the empty one and
237 # fill it with provided packets and add it to pg_streams list.
238 # @param cls The class pointer.
239 # @param i Integer variable to store the index of the packet-generator
240 # interface to create packet stream for.
241 # @param pkts List variable to store packets to be added to the stream.
243 def pg_add_stream(cls, i, pkts):
244 os.system("sudo rm -f /tmp/pg%u_in.pcap" % i)
245 wrpcap("/tmp/pg%u_in.pcap" % i, pkts)
246 # no equivalent API command
247 cls.cli(0, "packet-generator new pcap /tmp/pg%u_in.pcap source pg%u"
248 " name pcap%u" % (i, i, i))
249 cls.pg_streams.append('pcap%u' % i)
251 ## Class method to enable packet capturing for the packet-generator
253 # Delete old /tmp/pgX_out.pcap file if exists and set the packet-generator
254 # to capture outgoing packets to /tmp/pgX_out.pcap file.
255 # @param cls The class pointer.
256 # @param args List variable to store the indexes of the packet-generator
257 # interfaces to start packet capturing for.
259 def pg_enable_capture(cls, args):
261 os.system("sudo rm -f /tmp/pg%u_out.pcap" % i)
262 cls.cli(0, "packet-generator capture pg%u pcap /tmp/pg%u_out.pcap"
265 ## Class method to start packet sending.
266 # Start to send packets for all defined pg streams. Delete every stream
267 # from the stream list when sent and clear the pg_streams list.
268 # @param cls The class pointer.
271 cls.cli(2, "trace add pg-input 50") # 50 is maximum
272 cls.cli(0, 'packet-generator enable')
273 for stream in cls.pg_streams:
274 cls.cli(0, 'packet-generator delete %s' % stream)
277 ## Class method to return captured packets.
278 # Return packet captured for the defined packet-generator interface. Open
279 # the corresponding pcap file (/tmp/pgX_out.pcap), read the content and
280 # store captured packets to output variable.
281 # @param cls The class pointer.
282 # @param o Integer variable to store the index of the packet-generator
284 # @return output List of packets captured on the defined packet-generator
285 # interface. If the corresponding pcap file (/tmp/pgX_out.pcap) does not
286 # exist return empty list.
288 def pg_get_capture(cls, o):
289 pcap_filename = "/tmp/pg%u_out.pcap" % o
291 output = rdpcap(pcap_filename)
292 except IOError: # TODO
293 cls.log("WARNING: File %s does not exist, probably because no"
294 " packets arrived" % pcap_filename)
297 ## @var pcap_filename
298 # File descriptor to the corresponding pcap file.
300 ## Class method to create packet-generator interfaces.
301 # Create packet-generator interfaces and add host MAC addresses connected
302 # to these packet-generator interfaces to the MY_MACS dictionary.
303 # @param cls The class pointer.
304 # @param args List variable to store the indexes of the packet-generator
305 # interfaces to be created.
307 def create_interfaces(cls, args):
309 cls.MY_MACS[i] = "02:00:00:00:ff:%02x" % i
310 cls.log("My MAC address is %s" % (cls.MY_MACS[i]))
311 cls.api("pg_create_interface if_id %u" % i)
312 cls.api("sw_interface_set_flags pg%u admin-up" % i)
314 ## Static method to extend packet to specified size
315 # Extend provided packet to the specified size (including Ethernet FCS).
316 # The packet is extended by adding corresponding number of spaces to the
318 # NOTE: Currently works only when Raw layer is present.
319 # @param packet Variable to store packet object.
320 # @param size Integer variable to store the required size of the packet.
322 def extend_packet(packet, size):
323 packet_len = len(packet) + 4
324 extend = size - packet_len
326 packet[Raw].load += ' ' * extend
328 # Integer variable to store the current packet length including
331 # Integer variable to store the size of the packet extension.
333 ## Method to add packet info object to the packet_infos list.
334 # Extend the existing packet_infos list with the given information from
336 # @param self The object pointer.
337 # @param info Object to store required information from the packet.
338 def add_packet_info_to_list(self, info):
339 info.index = len(self.packet_infos)
340 self.packet_infos[info.index] = info
342 # Info object attribute to store the packet order in the stream.
344 # List variable to store required information from packets.
346 ## Method to create packet info object.
347 # Create the existing packet_infos list with the given information from
349 # @param self The object pointer.
350 # @param pg_id Integer variable to store the index of the packet-generator
352 def create_packet_info(self, pg_id, target_id):
354 self.add_packet_info_to_list(info)
359 # Object to store required information from packet.
361 # Info object attribute to store the index of the source packet
362 # generator interface of the packet.
364 # Info object attribute to store the index of the destination packet
365 # generator interface of the packet.
367 ## Static method to return packet info string.
368 # Create packet info string from the provided info object that will be put
369 # to the packet payload.
370 # @param info Object to store required information from the packet.
371 # @return String of information about packet's order in the stream, source
372 # and destination packet generator interface.
374 def info_to_payload(info):
375 return "%d %d %d" % (info.index, info.src, info.dst)
377 ## Static method to create packet info object from the packet payload.
378 # Create packet info object and set its attribute values based on data
379 # gained from the packet payload.
380 # @param payload String variable to store packet payload.
381 # @return info Object to store required information about the packet.
383 def payload_to_info(payload):
384 numbers = payload.split()
386 info.index = int(numbers[0])
387 info.src = int(numbers[1])
388 info.dst = int(numbers[2])
391 # Info object attribute to store the packet order in the stream.
393 # Info object attribute to store the index of the source packet
394 # generator interface of the packet.
396 # Info object attribute to store the index of the destination packet
397 # generator interface of the packet.
399 ## Method to return packet info object of the next packet in
400 # the packet_infos list.
401 # Get the next packet info object from the packet_infos list by increasing
402 # the packet_infos list index by one.
403 # @param self The object pointer.
404 # @param info Object to store required information about the packet.
405 # @return packet_infos[next_index] Next info object from the packet_infos
406 # list with stored information about packets. Return None if the end of
407 # the list is reached.
408 def get_next_packet_info(self, info):
412 next_index = info.index + 1
413 if next_index == len(self.packet_infos):
416 return self.packet_infos[next_index]
418 # Integer variable to store the index of the next info object.
420 ## Method to return packet info object of the next packet with the required
421 # source packet generator interface.
422 # Iterate over the packet_infos list and search for the next packet info
423 # object with the required source packet generator interface.
424 # @param self The object pointer.
425 # @param src_pg Integer variable to store index of requested source packet
426 # generator interface.
427 # @param info Object to store required information about the packet.
428 # @return packet_infos[next_index] Next info object from the packet_infos
429 # list with stored information about packets. Return None if the end of
430 # the list is reached.
431 def get_next_packet_info_for_interface(self, src_pg, info):
433 info = self.get_next_packet_info(info)
436 if info.src == src_pg:
439 # Info object attribute to store the index of the source packet
440 # generator interface of the packet.
442 ## Method to return packet info object of the next packet with required
443 # source and destination packet generator interfaces.
444 # Search for the next packet info object with the required source and
445 # destination packet generator interfaces.
446 # @param self The object pointer.
447 # @param src_pg Integer variable to store the index of the requested source
448 # packet generator interface.
449 # @param dst_pg Integer variable to store the index of the requested source
450 # packet generator interface.
451 # @param info Object to store required information about the packet.
452 # @return info Object with the info about the next packet with with
453 # required source and destination packet generator interfaces. Return None
454 # if there is no other packet with required data.
455 def get_next_packet_info_for_interface2(self, src_pg, dst_pg, info):
457 info = self.get_next_packet_info_for_interface(src_pg, info)
460 if info.dst == dst_pg:
463 # Info object attribute to store the index of the destination packet
464 # generator interface of the packet.
467 ## Subclass of the python unittest.TestResult class.
469 # This subclass provides methods to compile information about which tests have
470 # succeeded and which have failed.
471 class VppTestResult(unittest.TestResult):
473 # @param stream File descriptor to store where to report test results. Set
474 # to the standard error stream by default.
475 # @param descriptions Boolean variable to store information if to use test
477 # @param verbosity Integer variable to store required verbosity level.
478 def __init__(self, stream, descriptions, verbosity):
479 unittest.TestResult.__init__(self, stream, descriptions, verbosity)
481 self.descriptions = descriptions
482 self.verbosity = verbosity
483 self.result_string = None
484 ## @var result_string
485 # String variable to store the test case result string.
488 ## Method called when the test case succeeds.
489 # Run the default implementation (that does nothing) and set the result
490 # string in case of test case success.
491 # @param self The object pointer.
492 # @param test Object variable to store the test case instance.
493 def addSuccess(self, test):
494 unittest.TestResult.addSuccess(self, test)
495 self.result_string = GREEN + "OK" + END
496 ## @var result_string
497 # String variable to store the test case result string.
499 ## Method called when the test case signals a failure.
500 # Run the default implementation that appends a tuple (test, formatted_err)
501 # to the instance's failures attribute, where formatted_err is a formatted
502 # traceback derived from err and set the result string in case of test case
504 # @param self The object pointer.
505 # @param test Object variable to store the test case instance.
506 # @param err Tuple variable to store the error data:
507 # (type, value, traceback).
508 def addFailure(self, test, err):
509 unittest.TestResult.addFailure(self, test, err)
510 self.result_string = RED + "FAIL" + END
511 ## @var result_string
512 # String variable to store the test case result string.
514 ## Method called when the test case raises an unexpected exception.
515 # Run the default implementation that appends a tuple (test, formatted_err)
516 # to the instance's error attribute, where formatted_err is a formatted
517 # traceback derived from err and set the result string in case of test case
518 # unexpected failure.
519 # @param self The object pointer.
520 # @param test Object variable to store the test case instance.
521 # @param err Tuple variable to store the error data:
522 # (type, value, traceback).
523 def addError(self, test, err):
524 unittest.TestResult.addError(self, test, err)
525 self.result_string = RED + "ERROR" + END
526 ## @var result_string
527 # String variable to store the test case result string.
529 ## Method to get the description of the test case.
530 # Used to get the description string from the test case object.
531 # @param self The object pointer.
532 # @param test Object variable to store the test case instance.
533 # @return String of the short description if exist otherwise return test
535 def getDescription(self, test):
536 # TODO: if none print warning not raise exception
537 short_description = test.shortDescription()
538 if self.descriptions and short_description:
539 return short_description
542 ## @var short_description
543 # String variable to store the short description of the test case.
545 ## Method called when the test case is about to be run.
546 # Run the default implementation and based on the set verbosity level write
547 # the starting string to the output stream.
548 # @param self The object pointer.
549 # @param test Object variable to store the test case instance.
550 def startTest(self, test):
551 unittest.TestResult.startTest(self, test)
552 if self.verbosity > 0:
553 self.stream.writeln("Starting " + self.getDescription(test) + " ...")
554 self.stream.writeln("------------------------------------------------------------------")
556 ## Method called after the test case has been executed.
557 # Run the default implementation and based on the set verbosity level write
558 # the result string to the output stream.
559 # @param self The object pointer.
560 # @param test Object variable to store the test case instance.
561 def stopTest(self, test):
562 unittest.TestResult.stopTest(self, test)
563 if self.verbosity > 0:
564 self.stream.writeln("------------------------------------------------------------------")
565 self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string))
566 self.stream.writeln("------------------------------------------------------------------")
568 self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string))
570 ## Method to write errors and failures information to the output stream.
571 # Write content of errors and failures lists to the output stream.
572 # @param self The object pointer.
573 def printErrors(self):
574 self.stream.writeln()
575 self.printErrorList('ERROR', self.errors)
576 self.printErrorList('FAIL', self.failures)
578 # List variable containing 2-tuples of TestCase instances and strings
579 # holding formatted tracebacks. Each tuple represents a test which
580 # raised an unexpected exception.
582 # List variable containing 2-tuples of TestCase instances and strings
583 # holding formatted tracebacks. Each tuple represents a test where
584 # a failure was explicitly signalled using the TestCase.assert*()
587 ## Method to write the error information to the output stream.
588 # Write content of error lists to the output stream together with error
589 # type and test case description.
590 # @param self The object pointer.
591 # @param flavour String variable to store error type.
592 # @param errors List variable to store 2-tuples of TestCase instances and
593 # strings holding formatted tracebacks.
594 def printErrorList(self, flavour, errors):
595 for test, err in errors:
596 self.stream.writeln('=' * 70)
597 self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
598 self.stream.writeln('-' * 70)
599 self.stream.writeln("%s" % err)
601 # Object variable to store the test case instance.
603 # String variable to store formatted tracebacks.
606 ## Subclass of the python unittest.TextTestRunner class.
608 # A basic test runner implementation which prints results on standard error.
609 class VppTestRunner(unittest.TextTestRunner):
610 ## Class object variable to store the results of a set of tests.
611 resultclass = VppTestResult
613 ## Method to run the test.
614 # Print debug message in the terminal and run the standard run() method
615 # of the test runner collecting the result into the test result object.
616 # @param self The object pointer.
617 # @param test Object variable to store the test case instance.
618 # @return Test result object of the VppTestRunner.
620 print "Running tests using custom test runner" # debug message
621 return super(VppTestRunner, self).run(test)