3 from __future__ import print_function
18 from pathlib import Path
19 from collections import deque
20 from threading import Thread, Event
21 from inspect import getdoc, isclass
22 from traceback import format_exception
23 from logging import FileHandler, DEBUG, Formatter
25 from abc import ABC, abstractmethod
27 from config import config, max_vpp_cpus
28 import hook as hookmodule
29 from vpp_lo_interface import VppLoInterface
30 from vpp_papi_provider import VppPapiProvider
32 from vpp_papi.vpp_stats import VPPStats
33 from vpp_papi.vpp_transport_socket import VppTransportSocketIOError
43 from vpp_object import VppObjectRegistry
44 from util import is_core_present
45 from test_result_code import TestResultCode
47 logger = logging.getLogger(__name__)
49 # Set up an empty logger for the testcase that can be overridden as necessary
50 null_logger = logging.getLogger("VppAsfTestCase")
51 null_logger.addHandler(logging.NullHandler())
54 if config.debug_framework:
58 Test framework module.
60 The module provides a set of tools for constructing and running tests and
61 representing the results.
65 class VppDiedError(Exception):
66 """exception for reporting that the subprocess has died."""
70 for k, v in signal.__dict__.items()
71 if k.startswith("SIG") and not k.startswith("SIG_")
74 def __init__(self, rv=None, testcase=None, method_name=None):
76 self.signal_name = None
77 self.testcase = testcase
78 self.method_name = method_name
81 self.signal_name = VppDiedError.signals_by_value[-rv]
82 except (KeyError, TypeError):
85 if testcase is None and method_name is None:
88 in_msg = " while running %s.%s" % (testcase, method_name)
91 msg = "VPP subprocess died unexpectedly%s with return code: %d%s." % (
94 " [%s]" % (self.signal_name if self.signal_name is not None else ""),
97 msg = "VPP subprocess died unexpectedly%s." % in_msg
99 super(VppDiedError, self).__init__(msg)
102 def pump_output(testclass):
103 """pump output from vpp stdout/stderr to proper queues"""
106 while not testclass.pump_thread_stop_flag.is_set():
107 readable = select.select(
109 testclass.vpp.stdout.fileno(),
110 testclass.vpp.stderr.fileno(),
111 testclass.pump_thread_wakeup_pipe[0],
116 if testclass.vpp.stdout.fileno() in readable:
117 read = os.read(testclass.vpp.stdout.fileno(), 102400)
119 split = read.decode("ascii", errors="backslashreplace").splitlines(True)
120 if len(stdout_fragment) > 0:
121 split[0] = "%s%s" % (stdout_fragment, split[0])
122 if len(split) > 0 and split[-1].endswith("\n"):
126 stdout_fragment = split[-1]
127 testclass.vpp_stdout_deque.extend(split[:limit])
128 if not config.cache_vpp_output:
129 for line in split[:limit]:
130 testclass.logger.info("VPP STDOUT: %s" % line.rstrip("\n"))
131 if testclass.vpp.stderr.fileno() in readable:
132 read = os.read(testclass.vpp.stderr.fileno(), 102400)
134 split = read.decode("ascii", errors="backslashreplace").splitlines(True)
135 if len(stderr_fragment) > 0:
136 split[0] = "%s%s" % (stderr_fragment, split[0])
137 if len(split) > 0 and split[-1].endswith("\n"):
141 stderr_fragment = split[-1]
143 testclass.vpp_stderr_deque.extend(split[:limit])
144 if not config.cache_vpp_output:
145 for line in split[:limit]:
146 testclass.logger.error("VPP STDERR: %s" % line.rstrip("\n"))
147 # ignoring the dummy pipe here intentionally - the
148 # flag will take care of properly terminating the loop
151 def _is_platform_aarch64():
152 return platform.machine() == "aarch64"
155 is_platform_aarch64 = _is_platform_aarch64()
158 def _is_distro_ubuntu2204():
159 with open("/etc/os-release") as f:
160 for line in f.readlines():
166 is_distro_ubuntu2204 = _is_distro_ubuntu2204()
169 def _is_distro_debian11():
170 with open("/etc/os-release") as f:
171 for line in f.readlines():
172 if "bullseye" in line:
177 is_distro_debian11 = _is_distro_debian11()
180 def _is_distro_ubuntu2204():
181 with open("/etc/os-release") as f:
182 for line in f.readlines():
188 class KeepAliveReporter(object):
190 Singleton object which reports test start to parent process
196 self.__dict__ = self._shared_state
204 def pipe(self, pipe):
205 if self._pipe is not None:
206 raise Exception("Internal error - pipe should only be set once.")
209 def send_keep_alive(self, test, desc=None):
211 Write current test tmpdir & desc to keep-alive pipe to signal liveness
213 if self.pipe is None:
214 # if not running forked..
218 desc = "%s (%s)" % (desc, unittest.util.strclass(test))
222 self.pipe.send((desc, config.vpp, test.tempdir, test.vpp.pid))
225 class TestCaseTag(Enum):
226 # marks the suites that must run at the end
227 # using only a single test runner
229 # marks the suites broken on VPP multi-worker
230 FIXME_VPP_WORKERS = 2
231 # marks the suites broken when ASan is enabled
233 # marks suites broken on Ubuntu-22.04
235 # marks suites broken on Debian-11
237 # marks suites broken on debug vpp image
241 def create_tag_decorator(e):
244 cls.test_tags.append(e)
245 except AttributeError:
252 tag_run_solo = create_tag_decorator(TestCaseTag.RUN_SOLO)
253 tag_fixme_vpp_workers = create_tag_decorator(TestCaseTag.FIXME_VPP_WORKERS)
254 tag_fixme_asan = create_tag_decorator(TestCaseTag.FIXME_ASAN)
255 tag_fixme_ubuntu2204 = create_tag_decorator(TestCaseTag.FIXME_UBUNTU2204)
256 tag_fixme_debian11 = create_tag_decorator(TestCaseTag.FIXME_DEBIAN11)
257 tag_fixme_vpp_debug = create_tag_decorator(TestCaseTag.FIXME_VPP_DEBUG)
271 class CPUInterface(ABC):
273 skipped_due_to_cpu_lack = False
277 def get_cpus_required(cls):
281 def assign_cpus(cls, cpus):
285 class VppAsfTestCase(CPUInterface, unittest.TestCase):
286 """This subclass is a base class for VPP test cases that are implemented as
287 classes. It provides methods to create and run test case.
290 extra_vpp_statseg_config = ""
291 extra_vpp_config = []
292 extra_vpp_plugin_config = []
294 vapi_response_timeout = 5
295 remove_configured_vpp_objects_on_tear_down = True
298 def has_tag(cls, tag):
299 """if the test case has a given tag - return true"""
301 return tag in cls.test_tags
302 except AttributeError:
307 def is_tagged_run_solo(cls):
308 """if the test case class is timing-sensitive - return true"""
309 return cls.has_tag(TestCaseTag.RUN_SOLO)
312 def skip_fixme_asan(cls):
313 """if @tag_fixme_asan & ASan is enabled - mark for skip"""
314 if cls.has_tag(TestCaseTag.FIXME_ASAN):
315 vpp_extra_cmake_args = os.environ.get("VPP_EXTRA_CMAKE_ARGS", "")
316 if "DVPP_ENABLE_SANITIZE_ADDR=ON" in vpp_extra_cmake_args:
317 cls = unittest.skip("Skipping @tag_fixme_asan tests")(cls)
321 """Return the instance of this testcase"""
322 return cls.test_instance
325 def set_debug_flags(cls, d):
326 cls.gdbserver_port = 7777
327 cls.debug_core = False
328 cls.debug_gdb = False
329 cls.debug_gdbserver = False
330 cls.debug_all = False
331 cls.debug_attach = False
336 cls.debug_core = True
337 elif dl == "gdb" or dl == "gdb-all":
339 elif dl == "gdbserver" or dl == "gdbserver-all":
340 cls.debug_gdbserver = True
342 cls.debug_attach = True
344 raise Exception("Unrecognized DEBUG option: '%s'" % d)
345 if dl == "gdb-all" or dl == "gdbserver-all":
349 def get_vpp_worker_count(cls):
350 if not hasattr(cls, "vpp_worker_count"):
351 if cls.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
352 cls.vpp_worker_count = 0
354 cls.vpp_worker_count = config.vpp_worker_count
355 return cls.vpp_worker_count
358 def get_cpus_required(cls):
359 return 1 + cls.get_vpp_worker_count()
362 def setUpConstants(cls):
363 """Set-up the test case class based on environment variables"""
364 cls.step = config.step
365 cls.plugin_path = ":".join(config.vpp_plugin_dir)
366 cls.test_plugin_path = ":".join(config.vpp_test_plugin_dir)
367 cls.extern_plugin_path = ":".join(config.extern_plugin_dir)
369 if cls.step or cls.debug_gdb or cls.debug_gdbserver:
370 debug_cli = "cli-listen localhost:5002"
371 size = re.search(r"\d+[gG]", config.coredump_size)
373 coredump_size = f"coredump-size {config.coredump_size}".lower()
375 coredump_size = "coredump-size unlimited"
376 default_variant = config.variant
377 if default_variant is not None:
378 default_variant = "default { variant %s 100 }" % default_variant
382 api_fuzzing = config.api_fuzz
383 if api_fuzzing is None:
404 cls.get_api_segment_prefix(),
411 if cls.extern_plugin_path not in (None, ""):
412 cls.extra_vpp_plugin_config.append("add-path %s" % cls.extern_plugin_path)
413 if cls.get_vpp_worker_count():
414 cls.vpp_cmdline.extend(
415 ["corelist-workers", ",".join([str(x) for x in cls.cpus[1:]])]
417 cls.vpp_cmdline.extend(
428 cls.get_stats_sock_path(),
429 cls.extra_vpp_statseg_config,
434 cls.get_api_sock_path(),
455 "lisp_unittest_plugin.so",
460 "unittest_plugin.so",
465 + cls.extra_vpp_plugin_config
471 if cls.extra_vpp_config is not None:
472 cls.vpp_cmdline.extend(cls.extra_vpp_config)
474 if not cls.debug_attach:
475 cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
476 cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
479 def wait_for_enter(cls):
480 if cls.debug_gdbserver:
481 print(double_line_delim)
482 print("Spawned GDB server with PID: %d" % cls.vpp.pid)
484 print(double_line_delim)
485 print("Spawned VPP with PID: %d" % cls.vpp.pid)
487 cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid)
489 print(single_line_delim)
490 print("You can debug VPP using:")
491 if cls.debug_gdbserver:
493 f"sudo gdb {config.vpp} "
494 f"-ex 'target remote localhost:{cls.gdbserver_port}'"
497 "Now is the time to attach gdb by running the above "
498 "command, set up breakpoints etc., then resume VPP from "
499 "within gdb by issuing the 'continue' command"
501 cls.gdbserver_port += 1
503 print(f"sudo gdb {config.vpp} -ex 'attach {cls.vpp.pid}'")
505 "Now is the time to attach gdb by running the above "
506 "command and set up breakpoints etc., then resume VPP from"
507 " within gdb by issuing the 'continue' command"
509 print(single_line_delim)
510 input("Press ENTER to continue running the testcase...")
518 cls.logger.debug(f"Assigned cpus: {cls.cpus}")
519 cmdline = cls.vpp_cmdline
521 if cls.debug_gdbserver:
522 gdbserver = "/usr/bin/gdbserver"
523 if not os.path.isfile(gdbserver) or not os.access(gdbserver, os.X_OK):
525 "gdbserver binary '%s' does not exist or is "
526 "not executable" % gdbserver
531 "localhost:{port}".format(port=cls.gdbserver_port),
533 cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
536 cls.vpp = subprocess.Popen(
537 cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE
539 except subprocess.CalledProcessError as e:
541 "Subprocess returned with non-0 return code: (%s)", e.returncode
546 "Subprocess returned with OS error: (%s) %s", e.errno, e.strerror
549 except Exception as e:
550 cls.logger.exception("Subprocess returned unexpected from %s:", cmdline)
556 def wait_for_coredump(cls):
557 corefile = cls.tempdir + "/core"
558 if os.path.isfile(corefile):
559 cls.logger.error("Waiting for coredump to complete: %s", corefile)
560 curr_size = os.path.getsize(corefile)
561 deadline = time.time() + 60
563 while time.time() < deadline:
566 curr_size = os.path.getsize(corefile)
567 if size == curr_size:
572 "Timed out waiting for coredump to complete: %s", corefile
575 cls.logger.error("Coredump complete: %s, size %d", corefile, curr_size)
578 def get_stats_sock_path(cls):
579 return "%s/stats.sock" % cls.tempdir
582 def get_api_sock_path(cls):
583 return "%s/api.sock" % cls.tempdir
586 def get_api_segment_prefix(cls):
587 return os.path.basename(cls.tempdir) # Only used for VAPI
590 def get_tempdir(cls):
592 tmpdir = f"{config.tmp_dir}/unittest-attach-gdb"
594 tmpdir = f"{config.tmp_dir}/{get_testcase_dirname(cls.__name__)}"
595 if config.wipe_tmp_dir:
596 shutil.rmtree(tmpdir, ignore_errors=True)
601 def create_file_handler(cls):
602 if config.log_dir is None:
603 cls.file_handler = FileHandler(f"{cls.tempdir}/log.txt")
606 logdir = f"{config.log_dir}/{get_testcase_dirname(cls.__name__)}"
607 if config.wipe_tmp_dir:
608 shutil.rmtree(logdir, ignore_errors=True)
610 cls.file_handler = FileHandler(f"{logdir}/log.txt")
615 Perform class setup before running the testcase
616 Remove shared memory files, start vpp and connect the vpp-api
618 super(VppAsfTestCase, cls).setUpClass()
619 cls.logger = get_logger(cls.__name__)
620 cls.logger.debug(f"--- START setUpClass() {cls.__name__} ---")
621 random.seed(config.rnd_seed)
622 if hasattr(cls, "parallel_handler"):
623 cls.logger.addHandler(cls.parallel_handler)
624 cls.logger.propagate = False
625 cls.set_debug_flags(config.debug)
626 cls.tempdir = cls.get_tempdir()
627 cls.create_file_handler()
628 cls.file_handler.setFormatter(
629 Formatter(fmt="%(asctime)s,%(msecs)03d %(message)s", datefmt="%H:%M:%S")
631 cls.file_handler.setLevel(DEBUG)
632 cls.logger.addHandler(cls.file_handler)
633 cls.logger.debug("--- setUpClass() for %s called ---" % cls.__name__)
634 os.chdir(cls.tempdir)
636 "Temporary dir is %s, api socket is %s",
638 cls.get_api_sock_path(),
640 cls.logger.debug("Random seed is %s", config.rnd_seed)
644 cls.registry = VppObjectRegistry()
645 cls.vpp_startup_failed = False
646 cls.reporter = KeepAliveReporter()
647 # need to catch exceptions here because if we raise, then the cleanup
648 # doesn't get called and we might end with a zombie vpp
654 cls.reporter.send_keep_alive(cls, "setUpClass")
655 VppTestResult.current_test_case_info = TestCaseInfo(
656 cls.logger, cls.tempdir, cls.vpp.pid, config.vpp
658 cls.vpp_stdout_deque = deque()
659 cls.vpp_stderr_deque = deque()
660 if not cls.debug_attach:
661 cls.pump_thread_stop_flag = Event()
662 cls.pump_thread_wakeup_pipe = os.pipe()
663 cls.pump_thread = Thread(target=pump_output, args=(cls,))
664 cls.pump_thread.daemon = True
665 cls.pump_thread.start()
666 if cls.debug_gdb or cls.debug_gdbserver or cls.debug_attach:
667 cls.vapi_response_timeout = 0
668 cls.vapi = VppPapiProvider(cls.__name__, cls, cls.vapi_response_timeout)
670 hook = hookmodule.StepHook(cls)
672 hook = hookmodule.PollHook(cls)
673 cls.vapi.register_hook(hook)
674 cls.statistics = VPPStats(socketname=cls.get_stats_sock_path())
678 cls.wait_for_coredump()
679 cls.vpp_startup_failed = True
681 "VPP died shortly after startup, check the"
682 " output to standard error for possible cause"
687 except (vpp_papi.VPPIOError, Exception) as e:
688 cls.logger.debug("Exception connecting to vapi: %s" % e)
689 cls.vapi.disconnect()
691 if cls.debug_gdbserver:
694 "You're running VPP inside gdbserver but "
695 "VPP-API connection failed, did you forget "
696 "to 'continue' VPP from within gdb?",
702 last_line = cls.vapi.cli("show thread").split("\n")[-2]
703 cls.vpp_worker_count = int(last_line.split(" ")[0])
704 print("Detected VPP with %s workers." % cls.vpp_worker_count)
705 except vpp_papi.VPPRuntimeError as e:
706 cls.logger.debug("%s" % e)
709 except Exception as e:
710 cls.logger.debug("Exception connecting to VPP: %s" % e)
713 cls.logger.debug(f"--- END setUpClass() {cls.__name__} ---")
716 def _debug_quit(cls):
717 if cls.debug_gdbserver or cls.debug_gdb:
721 if cls.vpp.returncode is None:
723 print(double_line_delim)
724 print("VPP or GDB server is still running")
725 print(single_line_delim)
727 "When done debugging, press ENTER to kill the "
728 "process and finish running the testcase..."
730 except AttributeError:
736 Disconnect vpp-api, kill vpp and cleanup shared memory files
740 # first signal that we want to stop the pump thread, then wake it up
741 if hasattr(cls, "pump_thread_stop_flag"):
742 cls.pump_thread_stop_flag.set()
743 if hasattr(cls, "pump_thread_wakeup_pipe"):
744 os.write(cls.pump_thread_wakeup_pipe[1], b"ding dong wake up")
745 if hasattr(cls, "pump_thread"):
746 cls.logger.debug("Waiting for pump thread to stop")
747 cls.pump_thread.join()
748 if hasattr(cls, "vpp_stderr_reader_thread"):
749 cls.logger.debug("Waiting for stderr pump to stop")
750 cls.vpp_stderr_reader_thread.join()
752 if hasattr(cls, "vpp"):
753 if hasattr(cls, "vapi"):
754 cls.logger.debug(cls.vapi.vpp.get_stats())
755 cls.logger.debug("Disconnecting class vapi client on %s", cls.__name__)
756 cls.vapi.disconnect()
757 cls.logger.debug("Deleting class vapi attribute on %s", cls.__name__)
760 if not cls.debug_attach and cls.vpp.returncode is None:
761 cls.wait_for_coredump()
762 cls.logger.debug("Sending TERM to vpp")
764 cls.logger.debug("Waiting for vpp to die")
766 outs, errs = cls.vpp.communicate(timeout=5)
767 except subprocess.TimeoutExpired:
769 outs, errs = cls.vpp.communicate()
770 cls.logger.debug("Deleting class vpp attribute on %s", cls.__name__)
771 if not cls.debug_attach:
772 cls.vpp.stdout.close()
773 cls.vpp.stderr.close()
776 if cls.vpp_startup_failed:
777 stdout_log = cls.logger.info
778 stderr_log = cls.logger.critical
780 stdout_log = cls.logger.info
781 stderr_log = cls.logger.info
783 if hasattr(cls, "vpp_stdout_deque"):
784 stdout_log(single_line_delim)
785 stdout_log("VPP output to stdout while running %s:", cls.__name__)
786 stdout_log(single_line_delim)
787 vpp_output = "".join(cls.vpp_stdout_deque)
788 with open(cls.tempdir + "/vpp_stdout.txt", "w") as f:
790 stdout_log("\n%s", vpp_output)
791 stdout_log(single_line_delim)
793 if hasattr(cls, "vpp_stderr_deque"):
794 stderr_log(single_line_delim)
795 stderr_log("VPP output to stderr while running %s:", cls.__name__)
796 stderr_log(single_line_delim)
797 vpp_output = "".join(cls.vpp_stderr_deque)
798 with open(cls.tempdir + "/vpp_stderr.txt", "w") as f:
800 stderr_log("\n%s", vpp_output)
801 stderr_log(single_line_delim)
804 def tearDownClass(cls):
805 """Perform final cleanup after running all tests in this test-case"""
806 cls.logger.debug(f"--- START tearDownClass() {cls.__name__} ---")
807 cls.reporter.send_keep_alive(cls, "tearDownClass")
809 cls.file_handler.close()
810 if config.debug_framework:
811 debug_internal.on_tear_down_class(cls)
812 cls.logger.debug(f"--- END tearDownClass() {cls.__name__} ---")
814 def show_commands_at_teardown(self):
815 """Allow subclass specific teardown logging additions."""
816 self.logger.info("--- No test specific show commands provided. ---")
818 def unlink_testcase_file(self, path):
820 retries = MAX_ATTEMPTS
822 retries = retries - 1
823 self.logger.debug(f"Unlinking {path}")
826 # Loop until unlink() fails with FileNotFoundError to ensure file is removed
827 except FileNotFoundError:
830 self.logger.debug(f"OSError: unlinking {path}")
831 self.sleep(0.25, f"{retries} retries left")
832 if retries == 0 and os.path.isfile(path):
834 f"Unable to delete testcase file in {MAX_ATTEMPTS} attempts: {path}"
839 """Show various debug prints after each test"""
841 f"--- START tearDown() {self.__class__.__name__}.{self._testMethodName}({self._testMethodDoc}) ---"
845 if not self.vpp_dead:
846 self.logger.debug(self.vapi.cli("show trace max 1000"))
847 self.logger.info(self.vapi.ppcli("show interface"))
848 self.logger.info(self.vapi.ppcli("show hardware"))
849 self.logger.info(self.statistics.set_errors_str())
850 self.logger.info(self.vapi.ppcli("show run"))
851 self.logger.info(self.vapi.ppcli("show log"))
852 self.logger.info(self.vapi.ppcli("show bihash"))
853 self.logger.info("Logging testcase specific show commands.")
854 self.show_commands_at_teardown()
855 if self.remove_configured_vpp_objects_on_tear_down:
856 self.registry.remove_vpp_config(self.logger)
857 # Save/Dump VPP api trace log
858 m = self._testMethodName
859 api_trace = "vpp_api_trace.%s.%d.log" % (m, self.vpp.pid)
860 tmp_api_trace = "/tmp/%s" % api_trace
861 vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
862 self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
863 self.logger.info("Moving %s to %s\n" % (tmp_api_trace, vpp_api_trace_log))
864 shutil.move(tmp_api_trace, vpp_api_trace_log)
865 except VppTransportSocketIOError:
867 "VppTransportSocketIOError: Vpp dead. Cannot log show commands."
871 self.registry.unregister_all(self.logger)
872 # Remove any leftover pcap files
873 if hasattr(self, "pg_interfaces") and len(self.pg_interfaces) > 0:
874 testcase_dir = os.path.dirname(self.pg_interfaces[0].out_path)
875 for p in Path(testcase_dir).glob("pg*.pcap"):
876 self.unlink_testcase_file(p)
878 f"--- END tearDown() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
882 """Clear trace before running each test"""
883 super(VppAsfTestCase, self).setUp()
885 f"--- START setUp() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
887 # Save testname include in pcap history filenames
888 if hasattr(self, "pg_interfaces"):
889 for i in self.pg_interfaces:
890 i.test_name = self._testMethodName
891 self.reporter.send_keep_alive(self)
893 self.wait_for_coredump()
896 testcase=self.__class__.__name__,
897 method_name=self._testMethodName,
899 self.sleep(0.1, "during setUp")
900 self.vpp_stdout_deque.append(
901 "--- test setUp() for %s.%s(%s) starts here ---\n"
902 % (self.__class__.__name__, self._testMethodName, self._testMethodDoc)
904 self.vpp_stderr_deque.append(
905 "--- test setUp() for %s.%s(%s) starts here ---\n"
906 % (self.__class__.__name__, self._testMethodName, self._testMethodDoc)
908 self.vapi.cli("clear trace")
909 # store the test instance inside the test class - so that objects
910 # holding the class can access instance methods (like assertEqual)
911 type(self).test_instance = self
913 f"--- END setUp() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
917 def get_vpp_time(cls):
918 # processes e.g. "Time now 2.190522, Wed, 11 Mar 2020 17:29:54 GMT"
919 # returns float("2.190522")
920 timestr = cls.vapi.cli("show clock")
921 head, sep, tail = timestr.partition(",")
922 head, sep, tail = head.partition("Time now")
926 def sleep_on_vpp_time(cls, sec):
927 """Sleep according to time in VPP world"""
928 # On a busy system with many processes
929 # we might end up with VPP time being slower than real world
930 # So take that into account when waiting for VPP to do something
931 start_time = cls.get_vpp_time()
932 while cls.get_vpp_time() - start_time < sec:
936 def create_loopback_interfaces(cls, count):
938 Create loopback interfaces.
940 :param count: number of interfaces created.
941 :returns: List of created interfaces.
943 result = [VppLoInterface(cls) for i in range(count)]
945 setattr(cls, intf.name, intf)
946 cls.lo_interfaces = result
949 def assert_equal(self, real_value, expected_value, name_or_class=None):
950 if name_or_class is None:
951 self.assertEqual(real_value, expected_value)
954 msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
956 getdoc(name_or_class).strip(),
958 str(name_or_class(real_value)),
960 str(name_or_class(expected_value)),
963 msg = "Invalid %s: %s does not match expected value %s" % (
969 self.assertEqual(real_value, expected_value, msg)
971 def assert_in_range(self, real_value, expected_min, expected_max, name=None):
975 msg = "Invalid %s: %s out of range <%s,%s>" % (
981 self.assertTrue(expected_min <= real_value <= expected_max, msg)
983 def get_counter(self, counter):
984 if counter.startswith("/"):
985 counter_value = self.statistics.get_counter(counter)
987 counters = self.vapi.cli("sh errors").split("\n")
989 for i in range(1, len(counters) - 1):
990 results = counters[i].split()
991 if results[1] == counter:
992 counter_value = int(results[0])
996 def assert_counter_equal(self, counter, expected_value, thread=None, index=0):
997 c = self.get_counter(counter)
998 if thread is not None:
1001 c = sum(x[index] for x in c)
1003 "validate counter `%s[%s]', expected: %s, real value: %s"
1004 % (counter, index, expected_value, c)
1006 self.assert_equal(c, expected_value, "counter `%s[%s]'" % (counter, index))
1008 def assert_error_counter_equal(self, counter, expected_value):
1009 counter_value = self.statistics[counter].sum()
1010 self.assert_equal(counter_value, expected_value, "error counter `%s'" % counter)
1013 def sleep(cls, timeout, remark=None):
1014 # /* Allow sleep(0) to maintain win32 semantics, and as decreed
1015 # * by Guido, only the main thread can be interrupted.
1017 # https://github.com/python/cpython/blob/6673decfa0fb078f60587f5cb5e98460eea137c2/Modules/timemodule.c#L1892 # noqa
1020 if hasattr(os, "sched_yield"):
1026 cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
1027 before = time.time()
1030 if after - before > 2 * timeout:
1032 "unexpected self.sleep() result - slept for %es instead of ~%es!",
1038 "Finished sleep (%s) - slept %es (wanted %es)",
1044 def virtual_sleep(self, timeout, remark=None):
1045 self.logger.debug("Moving VPP time by %s (%s)", timeout, remark)
1046 self.vapi.cli("set clock adjust %s" % timeout)
1048 def snapshot_stats(self, stats_diff):
1049 """Return snapshot of interesting stats based on diff dictionary."""
1051 for sw_if_index in stats_diff:
1052 for counter in stats_diff[sw_if_index]:
1053 stats_snapshot[counter] = self.statistics[counter]
1054 self.logger.debug(f"Took statistics stats_snapshot: {stats_snapshot}")
1055 return stats_snapshot
1057 def compare_stats_with_snapshot(self, stats_diff, stats_snapshot):
1058 """Assert appropriate difference between current stats and snapshot."""
1059 for sw_if_index in stats_diff:
1060 for cntr, diff in stats_diff[sw_if_index].items():
1061 if sw_if_index == "err":
1063 self.statistics[cntr].sum(),
1064 stats_snapshot[cntr].sum() + diff,
1065 f"'{cntr}' counter value (previous value: "
1066 f"{stats_snapshot[cntr].sum()}, "
1067 f"expected diff: {diff})",
1072 self.statistics[cntr][:, sw_if_index].sum(),
1073 stats_snapshot[cntr][:, sw_if_index].sum() + diff,
1074 f"'{cntr}' counter value (previous value: "
1075 f"{stats_snapshot[cntr][:, sw_if_index].sum()}, "
1076 f"expected diff: {diff})",
1078 except IndexError as e:
1079 # if diff is 0, then this most probably a case where
1080 # test declares multiple interfaces but traffic hasn't
1081 # passed through this one yet - which means the counter
1082 # value is 0 and can be ignored
1085 f"Couldn't sum counter: {cntr} on sw_if_index: {sw_if_index}"
1089 def get_testcase_doc_name(test):
1090 return getdoc(test.__class__).splitlines()[0]
1093 def get_test_description(descriptions, test):
1094 short_description = test.shortDescription()
1095 if descriptions and short_description:
1096 return short_description
1101 def get_failed_testcase_linkname(failed_dir, testcase_dirname):
1102 return os.path.join(failed_dir, f"{testcase_dirname}-FAILED")
1105 def get_testcase_dirname(testcase_class_name):
1106 return f"vpp-unittest-{testcase_class_name}"
1109 class TestCaseInfo(object):
1110 def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
1111 self.logger = logger
1112 self.tempdir = tempdir
1113 self.vpp_pid = vpp_pid
1114 self.vpp_bin_path = vpp_bin_path
1115 self.core_crash_test = None
1118 class VppTestResult(unittest.TestResult):
1120 @property result_string
1121 String variable to store the test case result string.
1123 List variable containing 2-tuples of TestCase instances and strings
1124 holding formatted tracebacks. Each tuple represents a test which
1125 raised an unexpected exception.
1127 List variable containing 2-tuples of TestCase instances and strings
1128 holding formatted tracebacks. Each tuple represents a test where
1129 a failure was explicitly signalled using the TestCase.assert*()
1133 failed_test_cases_info = set()
1134 core_crash_test_cases_info = set()
1135 current_test_case_info = None
1137 def __init__(self, stream=None, descriptions=None, verbosity=None, runner=None):
1139 :param stream File descriptor to store where to report test results.
1140 Set to the standard error stream by default.
1141 :param descriptions Boolean variable to store information if to use
1142 test case descriptions.
1143 :param verbosity Integer variable to store required verbosity level.
1145 super(VppTestResult, self).__init__(stream, descriptions, verbosity)
1146 self.stream = stream
1147 self.descriptions = descriptions
1148 self.verbosity = verbosity
1149 self.result_code = TestResultCode.TEST_RUN
1150 self.result_string = None
1151 self.runner = runner
1154 def decodePcapFiles(self, test, when_configured=False):
1155 if when_configured == False or config.decode_pcaps == True:
1156 if hasattr(test, "pg_interfaces") and len(test.pg_interfaces) > 0:
1157 testcase_dir = os.path.dirname(test.pg_interfaces[0].out_path)
1158 test.pg_interfaces[0].decode_pcap_files(
1159 testcase_dir, f"suite{test.__class__.__name__}"
1161 test.pg_interfaces[0].decode_pcap_files(
1162 testcase_dir, test._testMethodName
1165 def addSuccess(self, test):
1167 Record a test succeeded result
1172 self.log_result("addSuccess", test)
1173 self.decodePcapFiles(test, when_configured=True)
1174 unittest.TestResult.addSuccess(self, test)
1175 self.result_string = colorize("OK", GREEN)
1176 self.result_code = TestResultCode.PASS
1177 self.send_result_through_pipe(test, self.result_code)
1179 def addExpectedFailure(self, test, err):
1180 self.log_result("addExpectedFailure", test, err)
1181 self.decodePcapFiles(test)
1182 super().addExpectedFailure(test, err)
1183 self.result_string = colorize("FAIL", GREEN)
1184 self.result_code = TestResultCode.EXPECTED_FAIL
1185 self.send_result_through_pipe(test, self.result_code)
1187 def addUnexpectedSuccess(self, test):
1188 self.log_result("addUnexpectedSuccess", test)
1189 self.decodePcapFiles(test, when_configured=True)
1190 super().addUnexpectedSuccess(test)
1191 self.result_string = colorize("OK", RED)
1192 self.result_code = TestResultCode.UNEXPECTED_PASS
1193 self.send_result_through_pipe(test, self.result_code)
1195 def addSkip(self, test, reason):
1197 Record a test skipped.
1203 self.log_result("addSkip", test, reason=reason)
1204 unittest.TestResult.addSkip(self, test, reason)
1205 self.result_string = colorize("SKIP", YELLOW)
1207 if reason == "not enough cpus":
1208 self.result_code = TestResultCode.SKIP_CPU_SHORTAGE
1210 self.result_code = TestResultCode.SKIP
1211 self.send_result_through_pipe(test, self.result_code)
1213 def symlink_failed(self):
1214 if self.current_test_case_info:
1216 failed_dir = config.failed_dir
1217 link_path = get_failed_testcase_linkname(
1218 failed_dir, os.path.basename(self.current_test_case_info.tempdir)
1221 self.current_test_case_info.logger.debug(
1222 "creating a link to the failed test"
1224 self.current_test_case_info.logger.debug(
1225 "os.symlink(%s, %s)"
1226 % (self.current_test_case_info.tempdir, link_path)
1228 if os.path.exists(link_path):
1229 self.current_test_case_info.logger.debug("symlink already exists")
1231 os.symlink(self.current_test_case_info.tempdir, link_path)
1233 except Exception as e:
1234 self.current_test_case_info.logger.error(e)
1236 def send_result_through_pipe(self, test, result):
1237 if hasattr(self, "test_framework_result_pipe"):
1238 pipe = self.test_framework_result_pipe
1240 pipe.send((test.id(), result))
1242 def log_result(self, fn, test, err=None, reason=None):
1243 if self.current_test_case_info:
1244 if isinstance(test, unittest.suite._ErrorHolder):
1245 test_name = test.description
1247 test_name = "%s.%s(%s)" % (
1248 test.__class__.__name__,
1249 test._testMethodName,
1250 test._testMethodDoc,
1254 extra_msg += f", error is {err}"
1256 extra_msg += f", reason is {reason}"
1257 self.current_test_case_info.logger.debug(
1258 f"--- {fn}() {test_name} called{extra_msg}"
1261 self.current_test_case_info.logger.debug(
1262 "formatted exception is:\n%s" % "".join(format_exception(*err))
1265 def add_error(self, test, err, unittest_fn, result_code):
1266 self.result_code = result_code
1267 if result_code == TestResultCode.FAIL:
1268 self.log_result("addFailure", test, err=err)
1269 error_type_str = colorize("FAIL", RED)
1270 elif result_code == TestResultCode.ERROR:
1271 self.log_result("addError", test, err=err)
1272 error_type_str = colorize("ERROR", RED)
1274 raise Exception(f"Unexpected result code {result_code}")
1275 self.decodePcapFiles(test)
1277 unittest_fn(self, test, err)
1278 if self.current_test_case_info:
1279 self.result_string = "%s [ temp dir used by test case: %s ]" % (
1281 self.current_test_case_info.tempdir,
1283 self.symlink_failed()
1284 self.failed_test_cases_info.add(self.current_test_case_info)
1285 if is_core_present(self.current_test_case_info.tempdir):
1286 if not self.current_test_case_info.core_crash_test:
1287 if isinstance(test, unittest.suite._ErrorHolder):
1288 test_name = str(test)
1290 test_name = "'{!s}' ({!s})".format(
1291 get_testcase_doc_name(test), test.id()
1293 self.current_test_case_info.core_crash_test = test_name
1294 self.core_crash_test_cases_info.add(self.current_test_case_info)
1296 self.result_string = "%s [no temp dir]" % error_type_str
1298 self.send_result_through_pipe(test, result_code)
1300 def addFailure(self, test, err):
1302 Record a test failed result
1305 :param err: error message
1308 self.add_error(test, err, unittest.TestResult.addFailure, TestResultCode.FAIL)
1310 def addError(self, test, err):
1312 Record a test error result
1315 :param err: error message
1318 self.add_error(test, err, unittest.TestResult.addError, TestResultCode.ERROR)
1320 def getDescription(self, test):
1322 Get test description
1325 :returns: test description
1328 return get_test_description(self.descriptions, test)
1330 def startTest(self, test):
1338 def print_header(test):
1339 if test.__class__ in self.printed:
1342 test_doc = getdoc(test)
1344 raise Exception("No doc string for test '%s'" % test.id())
1346 test_title = test_doc.splitlines()[0].rstrip()
1347 test_title = colorize(test_title, GREEN)
1348 if test.is_tagged_run_solo():
1349 test_title = colorize(f"SOLO RUN: {test_title}", YELLOW)
1351 # This block may overwrite the colorized title above,
1352 # but we want this to stand out and be fixed
1353 if test.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
1354 test_title = colorize(f"FIXME with VPP workers: {test_title}", RED)
1356 if test.has_tag(TestCaseTag.FIXME_ASAN):
1357 test_title = colorize(f"FIXME with ASAN: {test_title}", RED)
1358 test.skip_fixme_asan()
1360 if hasattr(test, "vpp_worker_count"):
1361 if test.vpp_worker_count == 0:
1362 test_title += " [main thread only]"
1363 elif test.vpp_worker_count == 1:
1364 test_title += " [1 worker thread]"
1366 test_title += f" [{test.vpp_worker_count} worker threads]"
1368 if test.__class__.skipped_due_to_cpu_lack:
1369 test_title = colorize(
1370 f"{test_title} [skipped - not enough cpus, "
1371 f"required={test.__class__.get_cpus_required()}, "
1372 f"available={max_vpp_cpus}]",
1376 print(double_line_delim)
1378 print(double_line_delim)
1379 self.printed.append(test.__class__)
1382 self.start_test = time.time()
1383 unittest.TestResult.startTest(self, test)
1384 if self.verbosity > 0:
1385 self.stream.writeln("Starting " + self.getDescription(test) + " ...")
1386 self.stream.writeln(single_line_delim)
1388 def stopTest(self, test):
1390 Called when the given test has been run
1395 unittest.TestResult.stopTest(self, test)
1397 result_code_to_suffix = {
1398 TestResultCode.PASS: "",
1399 TestResultCode.FAIL: "",
1400 TestResultCode.ERROR: "",
1401 TestResultCode.SKIP: "",
1402 TestResultCode.TEST_RUN: "",
1403 TestResultCode.SKIP_CPU_SHORTAGE: "",
1404 TestResultCode.EXPECTED_FAIL: " [EXPECTED FAIL]",
1405 TestResultCode.UNEXPECTED_PASS: " [UNEXPECTED PASS]",
1408 if self.verbosity > 0:
1409 self.stream.writeln(single_line_delim)
1410 self.stream.writeln(
1413 self.getDescription(test),
1415 result_code_to_suffix[self.result_code],
1418 self.stream.writeln(single_line_delim)
1420 self.stream.writeln(
1423 self.getDescription(test),
1424 time.time() - self.start_test,
1426 result_code_to_suffix[self.result_code],
1430 self.send_result_through_pipe(test, TestResultCode.TEST_RUN)
1432 def printErrors(self):
1434 Print errors from running the test case
1436 if len(self.errors) > 0 or len(self.failures) > 0:
1437 self.stream.writeln()
1438 self.printErrorList("ERROR", self.errors)
1439 self.printErrorList("FAIL", self.failures)
1441 # ^^ that is the last output from unittest before summary
1442 if not self.runner.print_summary:
1443 devnull = unittest.runner._WritelnDecorator(open(os.devnull, "w"))
1444 self.stream = devnull
1445 self.runner.stream = devnull
1447 def printErrorList(self, flavour, errors):
1449 Print error list to the output stream together with error type
1450 and test case description.
1452 :param flavour: error type
1453 :param errors: iterable errors
1456 for test, err in errors:
1457 self.stream.writeln(double_line_delim)
1458 self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
1459 self.stream.writeln(single_line_delim)
1460 self.stream.writeln("%s" % err)
1463 class VppTestRunner(unittest.TextTestRunner):
1465 A basic test runner implementation which prints results to standard error.
1469 def resultclass(self):
1470 """Class maintaining the results of the tests"""
1471 return VppTestResult
1475 keep_alive_pipe=None,
1485 # ignore stream setting here, use hard-coded stdout to be in sync
1486 # with prints from VppAsfTestCase methods ...
1487 super(VppTestRunner, self).__init__(
1488 sys.stdout, descriptions, verbosity, failfast, buffer, resultclass, **kwargs
1490 KeepAliveReporter.pipe = keep_alive_pipe
1492 self.orig_stream = self.stream
1493 self.resultclass.test_framework_result_pipe = result_pipe
1495 self.print_summary = print_summary
1497 def _makeResult(self):
1498 return self.resultclass(self.stream, self.descriptions, self.verbosity, self)
1500 def run(self, test):
1507 faulthandler.enable() # emit stack trace to stderr if killed by signal
1509 result = super(VppTestRunner, self).run(test)
1510 if not self.print_summary:
1511 self.stream = self.orig_stream
1512 result.stream = self.orig_stream
1516 class Worker(Thread):
1517 def __init__(self, executable_args, logger, env=None, *args, **kwargs):
1518 super(Worker, self).__init__(*args, **kwargs)
1519 self.logger = logger
1520 self.args = executable_args
1521 if hasattr(self, "testcase") and self.testcase.debug_all:
1522 if self.testcase.debug_gdbserver:
1524 "/usr/bin/gdbserver",
1525 "localhost:{port}".format(port=self.testcase.gdbserver_port),
1527 elif self.testcase.debug_gdb and hasattr(self, "wait_for_gdb"):
1528 self.args.append(self.wait_for_gdb)
1529 self.app_bin = executable_args[0]
1530 self.app_name = os.path.basename(self.app_bin)
1531 if hasattr(self, "role"):
1532 self.app_name += " {role}".format(role=self.role)
1535 env = {} if env is None else env
1536 self.env = copy.deepcopy(env)
1538 def wait_for_enter(self):
1539 if not hasattr(self, "testcase"):
1541 if self.testcase.debug_all and self.testcase.debug_gdbserver:
1543 print(double_line_delim)
1545 "Spawned GDB Server for '{app}' with PID: {pid}".format(
1546 app=self.app_name, pid=self.process.pid
1549 elif self.testcase.debug_all and self.testcase.debug_gdb:
1551 print(double_line_delim)
1553 "Spawned '{app}' with PID: {pid}".format(
1554 app=self.app_name, pid=self.process.pid
1559 print(single_line_delim)
1560 print("You can debug '{app}' using:".format(app=self.app_name))
1561 if self.testcase.debug_gdbserver:
1565 + " -ex 'target remote localhost:{port}'".format(
1566 port=self.testcase.gdbserver_port
1570 "Now is the time to attach gdb by running the above "
1571 "command, set up breakpoints etc., then resume from "
1572 "within gdb by issuing the 'continue' command"
1574 self.testcase.gdbserver_port += 1
1575 elif self.testcase.debug_gdb:
1579 + " -ex 'attach {pid}'".format(pid=self.process.pid)
1582 "Now is the time to attach gdb by running the above "
1583 "command and set up breakpoints etc., then resume from"
1584 " within gdb by issuing the 'continue' command"
1586 print(single_line_delim)
1587 input("Press ENTER to continue running the testcase...")
1590 executable = self.args[0]
1591 if not os.path.exists(executable) or not os.access(
1592 executable, os.F_OK | os.X_OK
1594 # Exit code that means some system file did not exist,
1595 # could not be opened, or had some other kind of error.
1596 self.result = os.EX_OSFILE
1597 raise EnvironmentError(
1598 "executable '%s' is not found or executable." % executable
1601 "Running executable '{app}': '{cmd}'".format(
1602 app=self.app_name, cmd=" ".join(self.args)
1605 env = os.environ.copy()
1606 env.update(self.env)
1607 env["CK_LOG_FILE_NAME"] = "-"
1608 self.process = subprocess.Popen(
1609 ["stdbuf", "-o0", "-e0"] + self.args,
1612 preexec_fn=os.setpgrp,
1613 stdout=subprocess.PIPE,
1614 stderr=subprocess.PIPE,
1616 self.wait_for_enter()
1617 out, err = self.process.communicate()
1618 self.logger.debug("Finished running `{app}'".format(app=self.app_name))
1619 self.logger.info("Return code is `%s'" % self.process.returncode)
1620 self.logger.info(single_line_delim)
1622 "Executable `{app}' wrote to stdout:".format(app=self.app_name)
1624 self.logger.info(single_line_delim)
1625 self.logger.info(out.decode("utf-8"))
1626 self.logger.info(single_line_delim)
1628 "Executable `{app}' wrote to stderr:".format(app=self.app_name)
1630 self.logger.info(single_line_delim)
1631 self.logger.info(err.decode("utf-8"))
1632 self.logger.info(single_line_delim)
1633 self.result = self.process.returncode
1636 if __name__ == "__main__":