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)
15 from inspect import getdoc
17 from scapy.utils import wrpcap, rdpcap
18 from scapy.packet import Raw
20 ## Static variables to store color formatting strings.
22 # These variables (RED, GREEN, YELLOW and LPURPLE) are used to configure
23 # the color of the text to be printed in the terminal. Variable END is used
24 # to revert the text color to the default one.
25 if hasattr(sys.stdout, 'isatty') and sys.stdout.isatty():
39 ## Private class to create packet info object.
41 # Help process information about the next packet.
42 # Set variables to default values.
43 class _PacketInfo(object):
49 # Integer variable to store the index of the packet.
51 # Integer variable to store the index of the source packet generator
52 # interface of the packet.
54 # Integer variable to store the index of the destination packet generator
55 # interface of the packet.
57 # Object variable to store the copy of the former packet.
59 ## Subclass of the python unittest.TestCase class.
61 # This subclass is a base class for test cases that are implemented as classes.
62 # It provides methods to create and run test case.
63 class VppTestCase(unittest.TestCase):
65 ## Class method to set class constants necessary to run test case.
66 # @param cls The class pointer.
68 def setUpConstants(cls):
69 cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
70 cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
71 cls.vpp_api_test_bin = os.getenv("VPP_TEST_API_TEST_BIN",
73 cls.vpp_cmdline = [cls.vpp_bin, "unix", "nodaemon", "api-segment", "{",
74 "prefix", "unittest", "}"]
75 if cls.plugin_path is not None:
76 cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
78 cls.vpp_api_test_cmdline = [cls.vpp_api_test_bin, "chroot", "prefix",
81 cls.verbose = int(os.getenv("V", 0))
86 # String variable to store the path to vpp (vector packet processor).
87 ## @var vpp_api_test_bin
88 # String variable to store the path to vpp_api_test (vpp API test tool).
90 # List of command line attributes for vpp.
91 ## @var vpp_api_test_cmdline
92 # List of command line attributes for vpp_api_test.
94 # Integer variable to store required verbosity level.
96 ## Class method to start the test case.
97 # 1. Initiate test case constants and set test case variables to default
99 # 2. Remove files from the shared memory.
100 # 3. Start vpp as a subprocess.
101 # @param cls The class pointer.
112 cls.packet_infos = {}
113 print "=================================================================="
114 print YELLOW + getdoc(cls) + END
115 print "=================================================================="
116 os.system("rm -f /dev/shm/unittest-global_vm")
117 os.system("rm -f /dev/shm/unittest-vpe-api")
118 os.system("rm -f /dev/shm/unittest-db")
119 cls.vpp = subprocess.Popen(cls.vpp_cmdline, stderr=subprocess.PIPE)
121 # List variable to store packet-generator streams for interfaces.
123 # Dictionary variable to store host MAC addresses connected to packet
124 # generator interfaces.
126 # Dictionary variable to store host IPv4 addresses connected to packet
127 # generator interfaces.
129 # Dictionary variable to store host IPv6 addresses connected to packet
130 # generator interfaces.
132 # Dictionary variable to store VPP MAC addresses of the packet
133 # generator interfaces.
135 # Dictionary variable to store VPP IPv4 addresses of the packet
136 # generator interfaces.
138 # Dictionary variable to store VPP IPv6 addresses of the packet
139 # generator interfaces.
141 # Test case object variable to store file descriptor of running vpp
142 # subprocess with open pipe to the standard error stream per
143 # VppTestCase object.
145 ## Class method to do cleaning when all tests (test_) defined for
146 # VppTestCase class are finished.
147 # 1. Terminate vpp and kill all vpp instances.
148 # 2. Remove files from the shared memory.
149 # @param cls The class pointer.
154 os.system("rm -f /dev/shm/unittest-global_vm")
155 os.system("rm -f /dev/shm/unittest-vpe-api")
156 os.system("rm -f /dev/shm/unittest-db")
158 ## Class method to define tear down action of the VppTestCase class.
159 # @param cls The class pointer.
161 def tearDownClass(cls):
164 ## Method to define tear down VPP actions of the test case.
165 # @param self The object pointer.
167 self.cli(2, "show int")
168 self.cli(2, "show trace")
169 self.cli(2, "show hardware")
170 self.cli(2, "show ip arp")
171 self.cli(2, "show ip fib")
172 self.cli(2, "show error")
173 self.cli(2, "show run")
175 ## Method to define setup action of the test case.
176 # @param self The object pointer.
178 self.cli(2, "clear trace")
180 ## Class method to print logs.
181 # Based on set level of verbosity print text in the terminal.
182 # @param cls The class pointer.
183 # @param s String variable to store text to be printed.
184 # @param v Integer variable to store required level of verbosity.
186 def log(cls, s, v=1):
188 print "LOG: " + LPURPLE + s + END
190 ## Class method to execute api commands.
191 # Based on set level of verbosity print the output of the api command in
193 # @param cls The class pointer.
194 # @param s String variable to store api command string.
197 p = subprocess.Popen(cls.vpp_api_test_cmdline,
198 stdout=subprocess.PIPE,
199 stdin=subprocess.PIPE,
200 stderr=subprocess.PIPE)
202 print "API: " + RED + s + END
204 out = p.communicate()[0]
205 out = out.replace("vat# ", "", 2)
208 print YELLOW + out + END
210 # Object variable to store file descriptor of vpp_api_test subprocess
211 # with open pipes to the standard output, inputs and error streams.
213 # Tuple variable to store standard output of vpp_api_test subprocess
214 # where the string "vat# " is replaced by empty string later.
216 ## Class method to execute cli commands.
217 # Based on set level of verbosity of the log and verbosity defined by
218 # environmental variable execute the cli command and print the output in
220 # CLI command is executed via vpp API test tool (exec + cli_command)
221 # @param cls The class pointer.
222 # @param v Integer variable to store required level of verbosity.
223 # @param s String variable to store cli command string.
228 p = subprocess.Popen(cls.vpp_api_test_cmdline,
229 stdout=subprocess.PIPE,
230 stdin=subprocess.PIPE,
231 stderr=subprocess.PIPE)
233 print "CLI: " + RED + s + END
234 p.stdin.write('exec ' + s)
235 out = p.communicate()[0]
236 out = out.replace("vat# ", "", 2)
239 print YELLOW + out + END
241 # Object variable to store file descriptor of vpp_api_test subprocess
242 # with open pipes to the standard output, inputs and error streams.
244 # Tuple variable to store standard output of vpp_api_test subprocess
245 # where the string "vat# " is replaced by empty string later.
247 ## Class method to create incoming packet stream for the packet-generator
249 # Delete old /tmp/pgX_in.pcap file if exists and create the empty one and
250 # fill it with provided packets and add it to pg_streams list.
251 # @param cls The class pointer.
252 # @param i Integer variable to store the index of the packet-generator
253 # interface to create packet stream for.
254 # @param pkts List variable to store packets to be added to the stream.
256 def pg_add_stream(cls, i, pkts):
257 os.system("rm -f /tmp/pg%u_in.pcap" % i)
258 wrpcap("/tmp/pg%u_in.pcap" % i, pkts)
259 # no equivalent API command
260 cls.cli(0, "packet-generator new pcap /tmp/pg%u_in.pcap source pg%u"
261 " name pcap%u" % (i, i, i))
262 cls.pg_streams.append('pcap%u' % i)
264 ## Class method to enable packet capturing for the packet-generator
266 # Delete old /tmp/pgX_out.pcap file if exists and set the packet-generator
267 # to capture outgoing packets to /tmp/pgX_out.pcap file.
268 # @param cls The class pointer.
269 # @param args List variable to store the indexes of the packet-generator
270 # interfaces to start packet capturing for.
272 def pg_enable_capture(cls, args):
274 os.system("rm -f /tmp/pg%u_out.pcap" % i)
275 cls.cli(0, "packet-generator capture pg%u pcap /tmp/pg%u_out.pcap"
278 ## Class method to start packet sending.
279 # Start to send packets for all defined pg streams. Delete every stream
280 # from the stream list when sent and clear the pg_streams list.
281 # @param cls The class pointer.
284 cls.cli(2, "trace add pg-input 50") # 50 is maximum
285 cls.cli(0, 'packet-generator enable')
286 for stream in cls.pg_streams:
287 cls.cli(0, 'packet-generator delete %s' % stream)
290 ## Class method to return captured packets.
291 # Return packet captured for the defined packet-generator interface. Open
292 # the corresponding pcap file (/tmp/pgX_out.pcap), read the content and
293 # store captured packets to output variable.
294 # @param cls The class pointer.
295 # @param o Integer variable to store the index of the packet-generator
297 # @return output List of packets captured on the defined packet-generator
298 # interface. If the corresponding pcap file (/tmp/pgX_out.pcap) does not
299 # exist return empty list.
301 def pg_get_capture(cls, o):
302 pcap_filename = "/tmp/pg%u_out.pcap" % o
304 output = rdpcap(pcap_filename)
305 except IOError: # TODO
306 cls.log("WARNING: File %s does not exist, probably because no"
307 " packets arrived" % pcap_filename)
310 ## @var pcap_filename
311 # File descriptor to the corresponding pcap file.
313 ## Class method to create packet-generator interfaces.
314 # Create packet-generator interfaces and add host MAC addresses connected
315 # to these packet-generator interfaces to the MY_MACS dictionary.
316 # @param cls The class pointer.
317 # @param args List variable to store the indexes of the packet-generator
318 # interfaces to be created.
320 def create_interfaces(cls, args):
322 cls.MY_MACS[i] = "02:00:00:00:ff:%02x" % i
323 cls.log("My MAC address is %s" % (cls.MY_MACS[i]))
324 cls.api("pg_create_interface if_id %u" % i)
325 cls.api("sw_interface_set_flags pg%u admin-up" % i)
327 ## Static method to extend packet to specified size
328 # Extend provided packet to the specified size (including Ethernet FCS).
329 # The packet is extended by adding corresponding number of spaces to the
331 # NOTE: Currently works only when Raw layer is present.
332 # @param packet Variable to store packet object.
333 # @param size Integer variable to store the required size of the packet.
335 def extend_packet(packet, size):
336 packet_len = len(packet) + 4
337 extend = size - packet_len
339 packet[Raw].load += ' ' * extend
341 # Integer variable to store the current packet length including
344 # Integer variable to store the size of the packet extension.
346 ## Method to add packet info object to the packet_infos list.
347 # Extend the existing packet_infos list with the given information from
349 # @param self The object pointer.
350 # @param info Object to store required information from the packet.
351 def add_packet_info_to_list(self, info):
352 info.index = len(self.packet_infos)
353 self.packet_infos[info.index] = info
355 # Info object attribute to store the packet order in the stream.
357 # List variable to store required information from packets.
359 ## Method to create packet info object.
360 # Create the existing packet_infos list with the given information from
362 # @param self The object pointer.
363 # @param pg_id Integer variable to store the index of the packet-generator
365 def create_packet_info(self, pg_id, target_id):
367 self.add_packet_info_to_list(info)
372 # Object to store required information from packet.
374 # Info object attribute to store the index of the source packet
375 # generator interface of the packet.
377 # Info object attribute to store the index of the destination packet
378 # generator interface of the packet.
380 ## Static method to return packet info string.
381 # Create packet info string from the provided info object that will be put
382 # to the packet payload.
383 # @param info Object to store required information from the packet.
384 # @return String of information about packet's order in the stream, source
385 # and destination packet generator interface.
387 def info_to_payload(info):
388 return "%d %d %d" % (info.index, info.src, info.dst)
390 ## Static method to create packet info object from the packet payload.
391 # Create packet info object and set its attribute values based on data
392 # gained from the packet payload.
393 # @param payload String variable to store packet payload.
394 # @return info Object to store required information about the packet.
396 def payload_to_info(payload):
397 numbers = payload.split()
399 info.index = int(numbers[0])
400 info.src = int(numbers[1])
401 info.dst = int(numbers[2])
404 # Info object attribute to store the packet order in the stream.
406 # Info object attribute to store the index of the source packet
407 # generator interface of the packet.
409 # Info object attribute to store the index of the destination packet
410 # generator interface of the packet.
412 ## Method to return packet info object of the next packet in
413 # the packet_infos list.
414 # Get the next packet info object from the packet_infos list by increasing
415 # the packet_infos list index by one.
416 # @param self The object pointer.
417 # @param info Object to store required information about the packet.
418 # @return packet_infos[next_index] Next info object from the packet_infos
419 # list with stored information about packets. Return None if the end of
420 # the list is reached.
421 def get_next_packet_info(self, info):
425 next_index = info.index + 1
426 if next_index == len(self.packet_infos):
429 return self.packet_infos[next_index]
431 # Integer variable to store the index of the next info object.
433 ## Method to return packet info object of the next packet with the required
434 # source packet generator interface.
435 # Iterate over the packet_infos list and search for the next packet info
436 # object with the required source packet generator interface.
437 # @param self The object pointer.
438 # @param src_pg Integer variable to store index of requested source packet
439 # generator interface.
440 # @param info Object to store required information about the packet.
441 # @return packet_infos[next_index] Next info object from the packet_infos
442 # list with stored information about packets. Return None if the end of
443 # the list is reached.
444 def get_next_packet_info_for_interface(self, src_pg, info):
446 info = self.get_next_packet_info(info)
449 if info.src == src_pg:
452 # Info object attribute to store the index of the source packet
453 # generator interface of the packet.
455 ## Method to return packet info object of the next packet with required
456 # source and destination packet generator interfaces.
457 # Search for the next packet info object with the required source and
458 # destination packet generator interfaces.
459 # @param self The object pointer.
460 # @param src_pg Integer variable to store the index of the requested source
461 # packet generator interface.
462 # @param dst_pg Integer variable to store the index of the requested source
463 # packet generator interface.
464 # @param info Object to store required information about the packet.
465 # @return info Object with the info about the next packet with with
466 # required source and destination packet generator interfaces. Return None
467 # if there is no other packet with required data.
468 def get_next_packet_info_for_interface2(self, src_pg, dst_pg, info):
470 info = self.get_next_packet_info_for_interface(src_pg, info)
473 if info.dst == dst_pg:
476 # Info object attribute to store the index of the destination packet
477 # generator interface of the packet.
480 ## Subclass of the python unittest.TestResult class.
482 # This subclass provides methods to compile information about which tests have
483 # succeeded and which have failed.
484 class VppTestResult(unittest.TestResult):
486 # @param stream File descriptor to store where to report test results. Set
487 # to the standard error stream by default.
488 # @param descriptions Boolean variable to store information if to use test
490 # @param verbosity Integer variable to store required verbosity level.
491 def __init__(self, stream, descriptions, verbosity):
492 unittest.TestResult.__init__(self, stream, descriptions, verbosity)
494 self.descriptions = descriptions
495 self.verbosity = verbosity
496 self.result_string = None
497 ## @var result_string
498 # String variable to store the test case result string.
501 ## Method called when the test case succeeds.
502 # Run the default implementation (that does nothing) and set the result
503 # string in case of test case success.
504 # @param self The object pointer.
505 # @param test Object variable to store the test case instance.
506 def addSuccess(self, test):
507 unittest.TestResult.addSuccess(self, test)
508 self.result_string = GREEN + "OK" + END
509 ## @var result_string
510 # String variable to store the test case result string.
512 ## Method called when the test case signals a failure.
513 # Run the default implementation that appends a tuple (test, formatted_err)
514 # to the instance's failures attribute, where formatted_err is a formatted
515 # traceback derived from err and set the result string in case of test case
517 # @param self The object pointer.
518 # @param test Object variable to store the test case instance.
519 # @param err Tuple variable to store the error data:
520 # (type, value, traceback).
521 def addFailure(self, test, err):
522 unittest.TestResult.addFailure(self, test, err)
523 self.result_string = RED + "FAIL" + END
524 ## @var result_string
525 # String variable to store the test case result string.
527 ## Method called when the test case raises an unexpected exception.
528 # Run the default implementation that appends a tuple (test, formatted_err)
529 # to the instance's error attribute, where formatted_err is a formatted
530 # traceback derived from err and set the result string in case of test case
531 # unexpected failure.
532 # @param self The object pointer.
533 # @param test Object variable to store the test case instance.
534 # @param err Tuple variable to store the error data:
535 # (type, value, traceback).
536 def addError(self, test, err):
537 unittest.TestResult.addError(self, test, err)
538 self.result_string = RED + "ERROR" + END
539 ## @var result_string
540 # String variable to store the test case result string.
542 ## Method to get the description of the test case.
543 # Used to get the description string from the test case object.
544 # @param self The object pointer.
545 # @param test Object variable to store the test case instance.
546 # @return String of the short description if exist otherwise return test
548 def getDescription(self, test):
549 # TODO: if none print warning not raise exception
550 short_description = test.shortDescription()
551 if self.descriptions and short_description:
552 return short_description
555 ## @var short_description
556 # String variable to store the short description of the test case.
558 ## Method called when the test case is about to be run.
559 # Run the default implementation and based on the set verbosity level write
560 # the starting string to the output stream.
561 # @param self The object pointer.
562 # @param test Object variable to store the test case instance.
563 def startTest(self, test):
564 unittest.TestResult.startTest(self, test)
565 if self.verbosity > 0:
566 self.stream.writeln("Starting " + self.getDescription(test) + " ...")
567 self.stream.writeln("------------------------------------------------------------------")
569 ## Method called after the test case has been executed.
570 # Run the default implementation and based on the set verbosity level write
571 # the result string to the output stream.
572 # @param self The object pointer.
573 # @param test Object variable to store the test case instance.
574 def stopTest(self, test):
575 unittest.TestResult.stopTest(self, test)
576 if self.verbosity > 0:
577 self.stream.writeln("------------------------------------------------------------------")
578 self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string))
579 self.stream.writeln("------------------------------------------------------------------")
581 self.stream.writeln("%-60s%s" % (self.getDescription(test), self.result_string))
583 ## Method to write errors and failures information to the output stream.
584 # Write content of errors and failures lists to the output stream.
585 # @param self The object pointer.
586 def printErrors(self):
587 self.stream.writeln()
588 self.printErrorList('ERROR', self.errors)
589 self.printErrorList('FAIL', self.failures)
591 # List variable containing 2-tuples of TestCase instances and strings
592 # holding formatted tracebacks. Each tuple represents a test which
593 # raised an unexpected exception.
595 # List variable containing 2-tuples of TestCase instances and strings
596 # holding formatted tracebacks. Each tuple represents a test where
597 # a failure was explicitly signalled using the TestCase.assert*()
600 ## Method to write the error information to the output stream.
601 # Write content of error lists to the output stream together with error
602 # type and test case description.
603 # @param self The object pointer.
604 # @param flavour String variable to store error type.
605 # @param errors List variable to store 2-tuples of TestCase instances and
606 # strings holding formatted tracebacks.
607 def printErrorList(self, flavour, errors):
608 for test, err in errors:
609 self.stream.writeln('=' * 70)
610 self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
611 self.stream.writeln('-' * 70)
612 self.stream.writeln("%s" % err)
614 # Object variable to store the test case instance.
616 # String variable to store formatted tracebacks.
619 ## Subclass of the python unittest.TextTestRunner class.
621 # A basic test runner implementation which prints results on standard error.
622 class VppTestRunner(unittest.TextTestRunner):
623 ## Class object variable to store the results of a set of tests.
624 resultclass = VppTestResult
626 ## Method to run the test.
627 # Print debug message in the terminal and run the standard run() method
628 # of the test runner collecting the result into the test result object.
629 # @param self The object pointer.
630 # @param test Object variable to store the test case instance.
631 # @return Test result object of the VppTestRunner.
633 print "Running tests using custom test runner" # debug message
634 return super(VppTestRunner, self).run(test)