3 from __future__ import print_function
15 from collections import deque
16 from threading import Thread, Event
17 from inspect import getdoc, isclass
18 from traceback import format_exception
19 from logging import FileHandler, DEBUG, Formatter
20 from scapy.packet import Raw
21 from hook import StepHook, PollHook, VppDiedError
22 from vpp_pg_interface import VppPGInterface
23 from vpp_sub_interface import VppSubInterface
24 from vpp_lo_interface import VppLoInterface
25 from vpp_papi_provider import VppPapiProvider
26 from vpp_papi.vpp_stats import VPPStats
27 from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
29 from vpp_object import VppObjectRegistry
30 from util import ppp, is_core_present
31 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
32 from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
33 from scapy.layers.inet6 import ICMPv6EchoReply
34 if os.name == 'posix' and sys.version_info[0] < 3:
35 # using subprocess32 is recommended by python official documentation
36 # @ https://docs.python.org/2/library/subprocess.html
37 import subprocess32 as subprocess
49 debug_framework = False
50 if os.getenv('TEST_DEBUG', "0") == "1":
51 debug_framework = True
56 Test framework module.
58 The module provides a set of tools for constructing and running tests and
59 representing the results.
63 class _PacketInfo(object):
64 """Private class to create packet info object.
66 Help process information about the next packet.
67 Set variables to default values.
69 #: Store the index of the packet.
71 #: Store the index of the source packet generator interface of the packet.
73 #: Store the index of the destination packet generator interface
76 #: Store expected ip version
78 #: Store expected upper protocol
80 #: Store the copy of the former packet.
83 def __eq__(self, other):
84 index = self.index == other.index
85 src = self.src == other.src
86 dst = self.dst == other.dst
87 data = self.data == other.data
88 return index and src and dst and data
91 def pump_output(testclass):
92 """ pump output from vpp stdout/stderr to proper queues """
95 while not testclass.pump_thread_stop_flag.is_set():
96 readable = select.select([testclass.vpp.stdout.fileno(),
97 testclass.vpp.stderr.fileno(),
98 testclass.pump_thread_wakeup_pipe[0]],
100 if testclass.vpp.stdout.fileno() in readable:
101 read = os.read(testclass.vpp.stdout.fileno(), 102400)
103 split = read.splitlines(True)
104 if len(stdout_fragment) > 0:
105 split[0] = "%s%s" % (stdout_fragment, split[0])
106 if len(split) > 0 and split[-1].endswith("\n"):
110 stdout_fragment = split[-1]
111 testclass.vpp_stdout_deque.extend(split[:limit])
112 if not testclass.cache_vpp_output:
113 for line in split[:limit]:
114 testclass.logger.debug(
115 "VPP STDOUT: %s" % line.rstrip("\n"))
116 if testclass.vpp.stderr.fileno() in readable:
117 read = os.read(testclass.vpp.stderr.fileno(), 102400)
119 split = read.splitlines(True)
120 if len(stderr_fragment) > 0:
121 split[0] = "%s%s" % (stderr_fragment, split[0])
122 if len(split) > 0 and split[-1].endswith("\n"):
126 stderr_fragment = split[-1]
127 testclass.vpp_stderr_deque.extend(split[:limit])
128 if not testclass.cache_vpp_output:
129 for line in split[:limit]:
130 testclass.logger.debug(
131 "VPP STDERR: %s" % line.rstrip("\n"))
132 # ignoring the dummy pipe here intentionally - the flag will take care
133 # of properly terminating the loop
136 def running_extended_tests():
137 s = os.getenv("EXTENDED_TESTS", "n")
138 return True if s.lower() in ("y", "yes", "1") else False
141 def running_on_centos():
142 os_id = os.getenv("OS_ID", "")
143 return True if "centos" in os_id.lower() else False
146 class KeepAliveReporter(object):
148 Singleton object which reports test start to parent process
153 self.__dict__ = self._shared_state
161 def pipe(self, pipe):
162 if self._pipe is not None:
163 raise Exception("Internal error - pipe should only be set once.")
166 def send_keep_alive(self, test, desc=None):
168 Write current test tmpdir & desc to keep-alive pipe to signal liveness
170 if self.pipe is None:
171 # if not running forked..
175 desc = '%s (%s)' % (desc, unittest.util.strclass(test))
179 self.pipe.send((desc, test.vpp_bin, test.tempdir, test.vpp.pid))
182 class VppTestCase(unittest.TestCase):
183 """This subclass is a base class for VPP test cases that are implemented as
184 classes. It provides methods to create and run test case.
188 def packet_infos(self):
189 """List of packet infos"""
190 return self._packet_infos
193 def get_packet_count_for_if_idx(cls, dst_if_index):
194 """Get the number of packet info for specified destination if index"""
195 if dst_if_index in cls._packet_count_for_dst_if_idx:
196 return cls._packet_count_for_dst_if_idx[dst_if_index]
202 """Return the instance of this testcase"""
203 return cls.test_instance
206 def set_debug_flags(cls, d):
207 cls.debug_core = False
208 cls.debug_gdb = False
209 cls.debug_gdbserver = False
214 cls.debug_core = True
217 elif dl == "gdbserver":
218 cls.debug_gdbserver = True
220 raise Exception("Unrecognized DEBUG option: '%s'" % d)
223 def get_least_used_cpu(self):
224 cpu_usage_list = [set(range(psutil.cpu_count()))]
225 vpp_processes = [p for p in psutil.process_iter(attrs=['pid', 'name'])
226 if 'vpp_main' == p.info['name']]
227 for vpp_process in vpp_processes:
228 for cpu_usage_set in cpu_usage_list:
230 cpu_num = vpp_process.cpu_num()
231 if cpu_num in cpu_usage_set:
232 cpu_usage_set_index = cpu_usage_list.index(
234 if cpu_usage_set_index == len(cpu_usage_list) - 1:
235 cpu_usage_list.append({cpu_num})
237 cpu_usage_list[cpu_usage_set_index + 1].add(
239 cpu_usage_set.remove(cpu_num)
241 except psutil.NoSuchProcess:
244 for cpu_usage_set in cpu_usage_list:
245 if len(cpu_usage_set) > 0:
246 min_usage_set = cpu_usage_set
249 return random.choice(tuple(min_usage_set))
252 def print_header(cls):
253 if not hasattr(cls, '_header_printed'):
254 print(double_line_delim)
255 print(colorize(getdoc(cls).splitlines()[0], GREEN))
256 print(double_line_delim)
257 cls._header_printed = True
260 def setUpConstants(cls):
261 """ Set-up the test case class based on environment variables """
262 s = os.getenv("STEP", "n")
263 cls.step = True if s.lower() in ("y", "yes", "1") else False
264 d = os.getenv("DEBUG", None)
265 c = os.getenv("CACHE_OUTPUT", "1")
266 cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True
267 cls.set_debug_flags(d)
268 cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
269 cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
270 cls.extern_plugin_path = os.getenv('EXTERN_PLUGINS')
272 if cls.plugin_path is not None:
273 if cls.extern_plugin_path is not None:
274 plugin_path = "%s:%s" % (
275 cls.plugin_path, cls.extern_plugin_path)
277 plugin_path = cls.plugin_path
278 elif cls.extern_plugin_path is not None:
279 plugin_path = cls.extern_plugin_path
281 if cls.step or cls.debug_gdb or cls.debug_gdbserver:
282 debug_cli = "cli-listen localhost:5002"
284 size = os.getenv("COREDUMP_SIZE")
286 coredump_size = "coredump-size %s" % size
287 if coredump_size is None:
288 coredump_size = "coredump-size unlimited"
290 cpu_core_number = cls.get_least_used_cpu()
292 cls.vpp_cmdline = [cls.vpp_bin, "unix",
293 "{", "nodaemon", debug_cli, "full-coredump",
294 coredump_size, "runtime-dir", cls.tempdir, "}",
295 "api-trace", "{", "on", "}", "api-segment", "{",
296 "prefix", cls.shm_prefix, "}", "cpu", "{",
297 "main-core", str(cpu_core_number), "}", "statseg",
298 "{", "socket-name", cls.stats_sock, "}", "plugins",
299 "{", "plugin", "dpdk_plugin.so", "{", "disable",
300 "}", "plugin", "unittest_plugin.so", "{", "enable",
302 if plugin_path is not None:
303 cls.vpp_cmdline.extend(["plugin_path", plugin_path])
304 cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
305 cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
308 def wait_for_enter(cls):
309 if cls.debug_gdbserver:
310 print(double_line_delim)
311 print("Spawned GDB server with PID: %d" % cls.vpp.pid)
313 print(double_line_delim)
314 print("Spawned VPP with PID: %d" % cls.vpp.pid)
316 cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid)
318 print(single_line_delim)
319 print("You can debug the VPP using e.g.:")
320 if cls.debug_gdbserver:
321 print("gdb " + cls.vpp_bin + " -ex 'target remote localhost:7777'")
322 print("Now is the time to attach a gdb by running the above "
323 "command, set up breakpoints etc. and then resume VPP from "
324 "within gdb by issuing the 'continue' command")
326 print("gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid)
327 print("Now is the time to attach a gdb by running the above "
328 "command and set up breakpoints etc.")
329 print(single_line_delim)
330 raw_input("Press ENTER to continue running the testcase...")
334 cmdline = cls.vpp_cmdline
336 if cls.debug_gdbserver:
337 gdbserver = '/usr/bin/gdbserver'
338 if not os.path.isfile(gdbserver) or \
339 not os.access(gdbserver, os.X_OK):
340 raise Exception("gdbserver binary '%s' does not exist or is "
341 "not executable" % gdbserver)
343 cmdline = [gdbserver, 'localhost:7777'] + cls.vpp_cmdline
344 cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
347 cls.vpp = subprocess.Popen(cmdline,
348 stdout=subprocess.PIPE,
349 stderr=subprocess.PIPE,
351 except Exception as e:
352 cls.logger.critical("Couldn't start vpp: %s" % e)
358 def wait_for_stats_socket(cls):
359 deadline = time.time() + 3
361 while time.time() < deadline or \
362 cls.debug_gdb or cls.debug_gdbserver:
363 if os.path.exists(cls.stats_sock):
368 cls.logger.critical("Couldn't stat : {}".format(cls.stats_sock))
373 Perform class setup before running the testcase
374 Remove shared memory files, start vpp and connect the vpp-api
376 gc.collect() # run garbage collection first
378 cls.print_header(cls)
379 cls.logger = get_logger(cls.__name__)
380 if hasattr(cls, 'parallel_handler'):
381 cls.logger.addHandler(cls.parallel_handler)
382 cls.tempdir = tempfile.mkdtemp(
383 prefix='vpp-unittest-%s-' % cls.__name__)
384 cls.stats_sock = "%s/stats.sock" % cls.tempdir
385 cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
386 cls.file_handler.setFormatter(
387 Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
389 cls.file_handler.setLevel(DEBUG)
390 cls.logger.addHandler(cls.file_handler)
391 cls.shm_prefix = os.path.basename(cls.tempdir)
392 os.chdir(cls.tempdir)
393 cls.logger.info("Temporary dir is %s, shm prefix is %s",
394 cls.tempdir, cls.shm_prefix)
396 cls.reset_packet_infos()
398 cls._zombie_captures = []
401 cls.registry = VppObjectRegistry()
402 cls.vpp_startup_failed = False
403 cls.reporter = KeepAliveReporter()
404 # need to catch exceptions here because if we raise, then the cleanup
405 # doesn't get called and we might end with a zombie vpp
408 cls.reporter.send_keep_alive(cls, 'setUpClass')
409 VppTestResult.current_test_case_info = TestCaseInfo(
410 cls.logger, cls.tempdir, cls.vpp.pid, cls.vpp_bin)
411 cls.vpp_stdout_deque = deque()
412 cls.vpp_stderr_deque = deque()
413 cls.pump_thread_stop_flag = Event()
414 cls.pump_thread_wakeup_pipe = os.pipe()
415 cls.pump_thread = Thread(target=pump_output, args=(cls,))
416 cls.pump_thread.daemon = True
417 cls.pump_thread.start()
418 if cls.debug_gdb or cls.debug_gdbserver:
422 cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls,
428 cls.vapi.register_hook(hook)
429 cls.wait_for_stats_socket()
430 cls.statistics = VPPStats(socketname=cls.stats_sock)
434 cls.vpp_startup_failed = True
436 "VPP died shortly after startup, check the"
437 " output to standard error for possible cause")
443 cls.vapi.disconnect()
446 if cls.debug_gdbserver:
447 print(colorize("You're running VPP inside gdbserver but "
448 "VPP-API connection failed, did you forget "
449 "to 'continue' VPP from within gdb?", RED))
461 Disconnect vpp-api, kill vpp and cleanup shared memory files
463 if (cls.debug_gdbserver or cls.debug_gdb) and hasattr(cls, 'vpp'):
465 if cls.vpp.returncode is None:
466 print(double_line_delim)
467 print("VPP or GDB server is still running")
468 print(single_line_delim)
469 raw_input("When done debugging, press ENTER to kill the "
470 "process and finish running the testcase...")
472 # first signal that we want to stop the pump thread, then wake it up
473 if hasattr(cls, 'pump_thread_stop_flag'):
474 cls.pump_thread_stop_flag.set()
475 if hasattr(cls, 'pump_thread_wakeup_pipe'):
476 os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
477 if hasattr(cls, 'pump_thread'):
478 cls.logger.debug("Waiting for pump thread to stop")
479 cls.pump_thread.join()
480 if hasattr(cls, 'vpp_stderr_reader_thread'):
481 cls.logger.debug("Waiting for stdderr pump to stop")
482 cls.vpp_stderr_reader_thread.join()
484 if hasattr(cls, 'vpp'):
485 if hasattr(cls, 'vapi'):
486 cls.vapi.disconnect()
489 if cls.vpp.returncode is None:
490 cls.logger.debug("Sending TERM to vpp")
492 cls.logger.debug("Waiting for vpp to die")
493 cls.vpp.communicate()
496 if cls.vpp_startup_failed:
497 stdout_log = cls.logger.info
498 stderr_log = cls.logger.critical
500 stdout_log = cls.logger.info
501 stderr_log = cls.logger.info
503 if hasattr(cls, 'vpp_stdout_deque'):
504 stdout_log(single_line_delim)
505 stdout_log('VPP output to stdout while running %s:', cls.__name__)
506 stdout_log(single_line_delim)
507 vpp_output = "".join(cls.vpp_stdout_deque)
508 with open(cls.tempdir + '/vpp_stdout.txt', 'w') as f:
510 stdout_log('\n%s', vpp_output)
511 stdout_log(single_line_delim)
513 if hasattr(cls, 'vpp_stderr_deque'):
514 stderr_log(single_line_delim)
515 stderr_log('VPP output to stderr while running %s:', cls.__name__)
516 stderr_log(single_line_delim)
517 vpp_output = "".join(cls.vpp_stderr_deque)
518 with open(cls.tempdir + '/vpp_stderr.txt', 'w') as f:
520 stderr_log('\n%s', vpp_output)
521 stderr_log(single_line_delim)
524 def tearDownClass(cls):
525 """ Perform final cleanup after running all tests in this test-case """
526 cls.reporter.send_keep_alive(cls, 'tearDownClass')
528 cls.file_handler.close()
529 cls.reset_packet_infos()
531 debug_internal.on_tear_down_class(cls)
534 """ Show various debug prints after each test """
535 self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
536 (self.__class__.__name__, self._testMethodName,
537 self._testMethodDoc))
538 if not self.vpp_dead:
539 self.logger.debug(self.vapi.cli("show trace"))
540 self.logger.info(self.vapi.ppcli("show interface"))
541 self.logger.info(self.vapi.ppcli("show hardware"))
542 self.logger.info(self.statistics.set_errors_str())
543 self.logger.info(self.vapi.ppcli("show run"))
544 self.logger.info(self.vapi.ppcli("show log"))
545 self.registry.remove_vpp_config(self.logger)
546 # Save/Dump VPP api trace log
547 api_trace = "vpp_api_trace.%s.log" % self._testMethodName
548 tmp_api_trace = "/tmp/%s" % api_trace
549 vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
550 self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
551 self.logger.info("Moving %s to %s\n" % (tmp_api_trace,
553 os.rename(tmp_api_trace, vpp_api_trace_log)
554 self.logger.info(self.vapi.ppcli("api trace custom-dump %s" %
557 self.registry.unregister_all(self.logger)
560 """ Clear trace before running each test"""
561 self.reporter.send_keep_alive(self)
562 self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
563 (self.__class__.__name__, self._testMethodName,
564 self._testMethodDoc))
566 raise Exception("VPP is dead when setting up the test")
567 self.sleep(.1, "during setUp")
568 self.vpp_stdout_deque.append(
569 "--- test setUp() for %s.%s(%s) starts here ---\n" %
570 (self.__class__.__name__, self._testMethodName,
571 self._testMethodDoc))
572 self.vpp_stderr_deque.append(
573 "--- test setUp() for %s.%s(%s) starts here ---\n" %
574 (self.__class__.__name__, self._testMethodName,
575 self._testMethodDoc))
576 self.vapi.cli("clear trace")
577 # store the test instance inside the test class - so that objects
578 # holding the class can access instance methods (like assertEqual)
579 type(self).test_instance = self
582 def pg_enable_capture(cls, interfaces=None):
584 Enable capture on packet-generator interfaces
586 :param interfaces: iterable interface indexes (if None,
587 use self.pg_interfaces)
590 if interfaces is None:
591 interfaces = cls.pg_interfaces
596 def register_capture(cls, cap_name):
597 """ Register a capture in the testclass """
598 # add to the list of captures with current timestamp
599 cls._captures.append((time.time(), cap_name))
600 # filter out from zombies
601 cls._zombie_captures = [(stamp, name)
602 for (stamp, name) in cls._zombie_captures
607 """ Remove any zombie captures and enable the packet generator """
608 # how long before capture is allowed to be deleted - otherwise vpp
609 # crashes - 100ms seems enough (this shouldn't be needed at all)
612 for stamp, cap_name in cls._zombie_captures:
613 wait = stamp + capture_ttl - now
615 cls.sleep(wait, "before deleting capture %s" % cap_name)
617 cls.logger.debug("Removing zombie capture %s" % cap_name)
618 cls.vapi.cli('packet-generator delete %s' % cap_name)
620 cls.vapi.cli("trace add pg-input 50") # 50 is maximum
621 cls.vapi.cli('packet-generator enable')
622 cls._zombie_captures = cls._captures
626 def create_pg_interfaces(cls, interfaces):
628 Create packet-generator interfaces.
630 :param interfaces: iterable indexes of the interfaces.
631 :returns: List of created interfaces.
636 intf = VppPGInterface(cls, i)
637 setattr(cls, intf.name, intf)
639 cls.pg_interfaces = result
643 def create_loopback_interfaces(cls, count):
645 Create loopback interfaces.
647 :param count: number of interfaces created.
648 :returns: List of created interfaces.
650 result = [VppLoInterface(cls) for i in range(count)]
652 setattr(cls, intf.name, intf)
653 cls.lo_interfaces = result
657 def extend_packet(packet, size, padding=' '):
659 Extend packet to given size by padding with spaces or custom padding
660 NOTE: Currently works only when Raw layer is present.
662 :param packet: packet
663 :param size: target size
664 :param padding: padding used to extend the payload
667 packet_len = len(packet) + 4
668 extend = size - packet_len
670 num = (extend / len(padding)) + 1
671 packet[Raw].load += (padding * num)[:extend]
674 def reset_packet_infos(cls):
675 """ Reset the list of packet info objects and packet counts to zero """
676 cls._packet_infos = {}
677 cls._packet_count_for_dst_if_idx = {}
680 def create_packet_info(cls, src_if, dst_if):
682 Create packet info object containing the source and destination indexes
683 and add it to the testcase's packet info list
685 :param VppInterface src_if: source interface
686 :param VppInterface dst_if: destination interface
688 :returns: _PacketInfo object
692 info.index = len(cls._packet_infos)
693 info.src = src_if.sw_if_index
694 info.dst = dst_if.sw_if_index
695 if isinstance(dst_if, VppSubInterface):
696 dst_idx = dst_if.parent.sw_if_index
698 dst_idx = dst_if.sw_if_index
699 if dst_idx in cls._packet_count_for_dst_if_idx:
700 cls._packet_count_for_dst_if_idx[dst_idx] += 1
702 cls._packet_count_for_dst_if_idx[dst_idx] = 1
703 cls._packet_infos[info.index] = info
707 def info_to_payload(info):
709 Convert _PacketInfo object to packet payload
711 :param info: _PacketInfo object
713 :returns: string containing serialized data from packet info
715 return "%d %d %d %d %d" % (info.index, info.src, info.dst,
719 def payload_to_info(payload):
721 Convert packet payload to _PacketInfo object
723 :param payload: packet payload
725 :returns: _PacketInfo object containing de-serialized data from payload
728 numbers = payload.split()
730 info.index = int(numbers[0])
731 info.src = int(numbers[1])
732 info.dst = int(numbers[2])
733 info.ip = int(numbers[3])
734 info.proto = int(numbers[4])
737 def get_next_packet_info(self, info):
739 Iterate over the packet info list stored in the testcase
740 Start iteration with first element if info is None
741 Continue based on index in info if info is specified
743 :param info: info or None
744 :returns: next info in list or None if no more infos
749 next_index = info.index + 1
750 if next_index == len(self._packet_infos):
753 return self._packet_infos[next_index]
755 def get_next_packet_info_for_interface(self, src_index, info):
757 Search the packet info list for the next packet info with same source
760 :param src_index: source interface index to search for
761 :param info: packet info - where to start the search
762 :returns: packet info or None
766 info = self.get_next_packet_info(info)
769 if info.src == src_index:
772 def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
774 Search the packet info list for the next packet info with same source
775 and destination interface indexes
777 :param src_index: source interface index to search for
778 :param dst_index: destination interface index to search for
779 :param info: packet info - where to start the search
780 :returns: packet info or None
784 info = self.get_next_packet_info_for_interface(src_index, info)
787 if info.dst == dst_index:
790 def assert_equal(self, real_value, expected_value, name_or_class=None):
791 if name_or_class is None:
792 self.assertEqual(real_value, expected_value)
795 msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
796 msg = msg % (getdoc(name_or_class).strip(),
797 real_value, str(name_or_class(real_value)),
798 expected_value, str(name_or_class(expected_value)))
800 msg = "Invalid %s: %s does not match expected value %s" % (
801 name_or_class, real_value, expected_value)
803 self.assertEqual(real_value, expected_value, msg)
805 def assert_in_range(self,
813 msg = "Invalid %s: %s out of range <%s,%s>" % (
814 name, real_value, expected_min, expected_max)
815 self.assertTrue(expected_min <= real_value <= expected_max, msg)
817 def assert_packet_checksums_valid(self, packet,
818 ignore_zero_udp_checksums=True):
819 received = packet.__class__(str(packet))
821 ppp("Verifying packet checksums for packet:", received))
822 udp_layers = ['UDP', 'UDPerror']
823 checksum_fields = ['cksum', 'chksum']
826 temp = received.__class__(str(received))
828 layer = temp.getlayer(counter)
830 for cf in checksum_fields:
831 if hasattr(layer, cf):
832 if ignore_zero_udp_checksums and \
833 0 == getattr(layer, cf) and \
834 layer.name in udp_layers:
837 checksums.append((counter, cf))
840 counter = counter + 1
841 if 0 == len(checksums):
843 temp = temp.__class__(str(temp))
844 for layer, cf in checksums:
845 calc_sum = getattr(temp[layer], cf)
847 getattr(received[layer], cf), calc_sum,
848 "packet checksum on layer #%d: %s" % (layer, temp[layer].name))
850 "Checksum field `%s` on `%s` layer has correct value `%s`" %
851 (cf, temp[layer].name, calc_sum))
853 def assert_checksum_valid(self, received_packet, layer,
855 ignore_zero_checksum=False):
856 """ Check checksum of received packet on given layer """
857 received_packet_checksum = getattr(received_packet[layer], field_name)
858 if ignore_zero_checksum and 0 == received_packet_checksum:
860 recalculated = received_packet.__class__(str(received_packet))
861 delattr(recalculated[layer], field_name)
862 recalculated = recalculated.__class__(str(recalculated))
863 self.assert_equal(received_packet_checksum,
864 getattr(recalculated[layer], field_name),
865 "packet checksum on layer: %s" % layer)
867 def assert_ip_checksum_valid(self, received_packet,
868 ignore_zero_checksum=False):
869 self.assert_checksum_valid(received_packet, 'IP',
870 ignore_zero_checksum=ignore_zero_checksum)
872 def assert_tcp_checksum_valid(self, received_packet,
873 ignore_zero_checksum=False):
874 self.assert_checksum_valid(received_packet, 'TCP',
875 ignore_zero_checksum=ignore_zero_checksum)
877 def assert_udp_checksum_valid(self, received_packet,
878 ignore_zero_checksum=True):
879 self.assert_checksum_valid(received_packet, 'UDP',
880 ignore_zero_checksum=ignore_zero_checksum)
882 def assert_embedded_icmp_checksum_valid(self, received_packet):
883 if received_packet.haslayer(IPerror):
884 self.assert_checksum_valid(received_packet, 'IPerror')
885 if received_packet.haslayer(TCPerror):
886 self.assert_checksum_valid(received_packet, 'TCPerror')
887 if received_packet.haslayer(UDPerror):
888 self.assert_checksum_valid(received_packet, 'UDPerror',
889 ignore_zero_checksum=True)
890 if received_packet.haslayer(ICMPerror):
891 self.assert_checksum_valid(received_packet, 'ICMPerror')
893 def assert_icmp_checksum_valid(self, received_packet):
894 self.assert_checksum_valid(received_packet, 'ICMP')
895 self.assert_embedded_icmp_checksum_valid(received_packet)
897 def assert_icmpv6_checksum_valid(self, pkt):
898 if pkt.haslayer(ICMPv6DestUnreach):
899 self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
900 self.assert_embedded_icmp_checksum_valid(pkt)
901 if pkt.haslayer(ICMPv6EchoRequest):
902 self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
903 if pkt.haslayer(ICMPv6EchoReply):
904 self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
906 def assert_packet_counter_equal(self, counter, expected_value):
907 counters = self.vapi.cli("sh errors").split('\n')
909 for i in range(1, len(counters)-1):
910 results = counters[i].split()
911 if results[1] == counter:
912 counter_value = int(results[0])
914 self.assert_equal(counter_value, expected_value,
915 "packet counter `%s'" % counter)
918 def sleep(cls, timeout, remark=None):
919 if hasattr(cls, 'logger'):
920 cls.logger.debug("Starting sleep for %ss (%s)" % (timeout, remark))
924 if after - before > 2 * timeout:
925 cls.logger.error("unexpected time.sleep() result - "
926 "slept for %ss instead of ~%ss!" % (
927 after - before, timeout))
928 if hasattr(cls, 'logger'):
930 "Finished sleep (%s) - slept %ss (wanted %ss)" % (
931 remark, after - before, timeout))
933 def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
934 self.vapi.cli("clear trace")
935 intf.add_stream(pkts)
936 self.pg_enable_capture(self.pg_interfaces)
940 for i in self.pg_interfaces:
941 i.get_capture(0, timeout=timeout)
942 i.assert_nothing_captured(remark=remark)
945 def send_and_expect(self, input, pkts, output):
946 self.vapi.cli("clear trace")
947 input.add_stream(pkts)
948 self.pg_enable_capture(self.pg_interfaces)
950 if isinstance(object, (list,)):
953 rx.append(output.get_capture(len(pkts)))
955 rx = output.get_capture(len(pkts))
958 def send_and_expect_only(self, input, pkts, output, timeout=None):
959 self.vapi.cli("clear trace")
960 input.add_stream(pkts)
961 self.pg_enable_capture(self.pg_interfaces)
963 if isinstance(object, (list,)):
967 rx.append(output.get_capture(len(pkts)))
969 rx = output.get_capture(len(pkts))
973 for i in self.pg_interfaces:
975 i.get_capture(0, timeout=timeout)
976 i.assert_nothing_captured()
982 def get_testcase_doc_name(test):
983 return getdoc(test.__class__).splitlines()[0]
986 def get_test_description(descriptions, test):
987 short_description = test.shortDescription()
988 if descriptions and short_description:
989 return short_description
994 class TestCaseInfo(object):
995 def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
997 self.tempdir = tempdir
998 self.vpp_pid = vpp_pid
999 self.vpp_bin_path = vpp_bin_path
1000 self.core_crash_test = None
1003 class VppTestResult(unittest.TestResult):
1005 @property result_string
1006 String variable to store the test case result string.
1008 List variable containing 2-tuples of TestCase instances and strings
1009 holding formatted tracebacks. Each tuple represents a test which
1010 raised an unexpected exception.
1012 List variable containing 2-tuples of TestCase instances and strings
1013 holding formatted tracebacks. Each tuple represents a test where
1014 a failure was explicitly signalled using the TestCase.assert*()
1018 failed_test_cases_info = set()
1019 core_crash_test_cases_info = set()
1020 current_test_case_info = None
1022 def __init__(self, stream, descriptions, verbosity):
1024 :param stream File descriptor to store where to report test results.
1025 Set to the standard error stream by default.
1026 :param descriptions Boolean variable to store information if to use
1027 test case descriptions.
1028 :param verbosity Integer variable to store required verbosity level.
1030 unittest.TestResult.__init__(self, stream, descriptions, verbosity)
1031 self.stream = stream
1032 self.descriptions = descriptions
1033 self.verbosity = verbosity
1034 self.result_string = None
1036 def addSuccess(self, test):
1038 Record a test succeeded result
1043 if self.current_test_case_info:
1044 self.current_test_case_info.logger.debug(
1045 "--- addSuccess() %s.%s(%s) called" % (test.__class__.__name__,
1046 test._testMethodName,
1047 test._testMethodDoc))
1048 unittest.TestResult.addSuccess(self, test)
1049 self.result_string = colorize("OK", GREEN)
1051 self.send_result_through_pipe(test, PASS)
1053 def addSkip(self, test, reason):
1055 Record a test skipped.
1061 if self.current_test_case_info:
1062 self.current_test_case_info.logger.debug(
1063 "--- addSkip() %s.%s(%s) called, reason is %s" %
1064 (test.__class__.__name__, test._testMethodName,
1065 test._testMethodDoc, reason))
1066 unittest.TestResult.addSkip(self, test, reason)
1067 self.result_string = colorize("SKIP", YELLOW)
1069 self.send_result_through_pipe(test, SKIP)
1071 def symlink_failed(self):
1072 if self.current_test_case_info:
1074 failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
1075 link_path = os.path.join(
1078 os.path.basename(self.current_test_case_info.tempdir))
1079 if self.current_test_case_info.logger:
1080 self.current_test_case_info.logger.debug(
1081 "creating a link to the failed test")
1082 self.current_test_case_info.logger.debug(
1083 "os.symlink(%s, %s)" %
1084 (self.current_test_case_info.tempdir, link_path))
1085 if os.path.exists(link_path):
1086 if self.current_test_case_info.logger:
1087 self.current_test_case_info.logger.debug(
1088 'symlink already exists')
1090 os.symlink(self.current_test_case_info.tempdir, link_path)
1092 except Exception as e:
1093 if self.current_test_case_info.logger:
1094 self.current_test_case_info.logger.error(e)
1096 def send_result_through_pipe(self, test, result):
1097 if hasattr(self, 'test_framework_result_pipe'):
1098 pipe = self.test_framework_result_pipe
1100 pipe.send((test.id(), result))
1102 def log_error(self, test, err, fn_name):
1103 if self.current_test_case_info:
1104 if isinstance(test, unittest.suite._ErrorHolder):
1105 test_name = test.description
1107 test_name = '%s.%s(%s)' % (test.__class__.__name__,
1108 test._testMethodName,
1109 test._testMethodDoc)
1110 self.current_test_case_info.logger.debug(
1111 "--- %s() %s called, err is %s" %
1112 (fn_name, test_name, err))
1113 self.current_test_case_info.logger.debug(
1114 "formatted exception is:\n%s" %
1115 "".join(format_exception(*err)))
1117 def add_error(self, test, err, unittest_fn, error_type):
1118 if error_type == FAIL:
1119 self.log_error(test, err, 'addFailure')
1120 error_type_str = colorize("FAIL", RED)
1121 elif error_type == ERROR:
1122 self.log_error(test, err, 'addError')
1123 error_type_str = colorize("ERROR", RED)
1125 raise Exception('Error type %s cannot be used to record an '
1126 'error or a failure' % error_type)
1128 unittest_fn(self, test, err)
1129 if self.current_test_case_info:
1130 self.result_string = "%s [ temp dir used by test case: %s ]" % \
1132 self.current_test_case_info.tempdir)
1133 self.symlink_failed()
1134 self.failed_test_cases_info.add(self.current_test_case_info)
1135 if is_core_present(self.current_test_case_info.tempdir):
1136 if not self.current_test_case_info.core_crash_test:
1137 if isinstance(test, unittest.suite._ErrorHolder):
1138 test_name = str(test)
1140 test_name = "'{}' ({})".format(
1141 get_testcase_doc_name(test), test.id())
1142 self.current_test_case_info.core_crash_test = test_name
1143 self.core_crash_test_cases_info.add(
1144 self.current_test_case_info)
1146 self.result_string = '%s [no temp dir]' % error_type_str
1148 self.send_result_through_pipe(test, error_type)
1150 def addFailure(self, test, err):
1152 Record a test failed result
1155 :param err: error message
1158 self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
1160 def addError(self, test, err):
1162 Record a test error result
1165 :param err: error message
1168 self.add_error(test, err, unittest.TestResult.addError, ERROR)
1170 def getDescription(self, test):
1172 Get test description
1175 :returns: test description
1178 return get_test_description(self.descriptions, test)
1180 def startTest(self, test):
1187 test.print_header(test.__class__)
1189 unittest.TestResult.startTest(self, test)
1190 if self.verbosity > 0:
1191 self.stream.writeln(
1192 "Starting " + self.getDescription(test) + " ...")
1193 self.stream.writeln(single_line_delim)
1195 def stopTest(self, test):
1197 Called when the given test has been run
1202 unittest.TestResult.stopTest(self, test)
1203 if self.verbosity > 0:
1204 self.stream.writeln(single_line_delim)
1205 self.stream.writeln("%-73s%s" % (self.getDescription(test),
1206 self.result_string))
1207 self.stream.writeln(single_line_delim)
1209 self.stream.writeln("%-73s%s" % (self.getDescription(test),
1210 self.result_string))
1212 self.send_result_through_pipe(test, TEST_RUN)
1214 def printErrors(self):
1216 Print errors from running the test case
1218 self.stream.writeln()
1219 self.printErrorList('ERROR', self.errors)
1220 self.printErrorList('FAIL', self.failures)
1222 def printErrorList(self, flavour, errors):
1224 Print error list to the output stream together with error type
1225 and test case description.
1227 :param flavour: error type
1228 :param errors: iterable errors
1231 for test, err in errors:
1232 self.stream.writeln(double_line_delim)
1233 self.stream.writeln("%s: %s" %
1234 (flavour, self.getDescription(test)))
1235 self.stream.writeln(single_line_delim)
1236 self.stream.writeln("%s" % err)
1239 class VppTestRunner(unittest.TextTestRunner):
1241 A basic test runner implementation which prints results to standard error.
1244 def resultclass(self):
1245 """Class maintaining the results of the tests"""
1246 return VppTestResult
1248 def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
1249 result_pipe=None, failfast=False, buffer=False,
1252 # ignore stream setting here, use hard-coded stdout to be in sync
1253 # with prints from VppTestCase methods ...
1254 super(VppTestRunner, self).__init__(sys.stdout, descriptions,
1255 verbosity, failfast, buffer,
1257 reporter = KeepAliveReporter()
1258 reporter.pipe = keep_alive_pipe
1260 VppTestResult.test_framework_result_pipe = result_pipe
1262 def run(self, test):
1269 faulthandler.enable() # emit stack trace to stderr if killed by signal
1271 result = super(VppTestRunner, self).run(test)
1275 class Worker(Thread):
1276 def __init__(self, args, logger, env={}):
1277 self.logger = logger
1280 self.env = copy.deepcopy(env)
1281 super(Worker, self).__init__()
1284 executable = self.args[0]
1285 self.logger.debug("Running executable w/args `%s'" % self.args)
1286 env = os.environ.copy()
1287 env.update(self.env)
1288 env["CK_LOG_FILE_NAME"] = "-"
1289 self.process = subprocess.Popen(
1290 self.args, shell=False, env=env, preexec_fn=os.setpgrp,
1291 stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1292 out, err = self.process.communicate()
1293 self.logger.debug("Finished running `%s'" % executable)
1294 self.logger.info("Return code is `%s'" % self.process.returncode)
1295 self.logger.info(single_line_delim)
1296 self.logger.info("Executable `%s' wrote to stdout:" % executable)
1297 self.logger.info(single_line_delim)
1298 self.logger.info(out)
1299 self.logger.info(single_line_delim)
1300 self.logger.info("Executable `%s' wrote to stderr:" % executable)
1301 self.logger.info(single_line_delim)
1302 self.logger.info(err)
1303 self.logger.info(single_line_delim)
1304 self.result = self.process.returncode