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 = "defaults { %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. ---")
819 """Show various debug prints after each test"""
821 f"--- START tearDown() {self.__class__.__name__}.{self._testMethodName}({self._testMethodDoc}) ---"
825 if not self.vpp_dead:
826 self.logger.debug(self.vapi.cli("show trace max 1000"))
827 self.logger.info(self.vapi.ppcli("show interface"))
828 self.logger.info(self.vapi.ppcli("show hardware"))
829 self.logger.info(self.statistics.set_errors_str())
830 self.logger.info(self.vapi.ppcli("show run"))
831 self.logger.info(self.vapi.ppcli("show log"))
832 self.logger.info(self.vapi.ppcli("show bihash"))
833 self.logger.info("Logging testcase specific show commands.")
834 self.show_commands_at_teardown()
835 if self.remove_configured_vpp_objects_on_tear_down:
836 self.registry.remove_vpp_config(self.logger)
837 # Save/Dump VPP api trace log
838 m = self._testMethodName
839 api_trace = "vpp_api_trace.%s.%d.log" % (m, self.vpp.pid)
840 tmp_api_trace = "/tmp/%s" % api_trace
841 vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
842 self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
843 self.logger.info("Moving %s to %s\n" % (tmp_api_trace, vpp_api_trace_log))
844 shutil.move(tmp_api_trace, vpp_api_trace_log)
845 except VppTransportSocketIOError:
847 "VppTransportSocketIOError: Vpp dead. Cannot log show commands."
851 self.registry.unregister_all(self.logger)
852 # Remove any leftover pcap files
853 if hasattr(self, "pg_interfaces") and len(self.pg_interfaces) > 0:
854 testcase_dir = os.path.dirname(self.pg_interfaces[0].out_path)
855 for p in Path(testcase_dir).glob("pg*.pcap"):
858 retries = retries - 1
859 self.logger.debug(f"Unlinking {p}")
864 self.logger.debug(f"OSError: unlinking {p}")
865 self.sleep(0.25, f"{retries} retries left")
868 f"--- END tearDown() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
872 """Clear trace before running each test"""
873 super(VppAsfTestCase, self).setUp()
875 f"--- START setUp() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
877 # Save testname include in pcap history filenames
878 if hasattr(self, "pg_interfaces"):
879 for i in self.pg_interfaces:
880 i.test_name = self._testMethodName
881 self.reporter.send_keep_alive(self)
883 self.wait_for_coredump()
886 testcase=self.__class__.__name__,
887 method_name=self._testMethodName,
889 self.sleep(0.1, "during setUp")
890 self.vpp_stdout_deque.append(
891 "--- test setUp() for %s.%s(%s) starts here ---\n"
892 % (self.__class__.__name__, self._testMethodName, self._testMethodDoc)
894 self.vpp_stderr_deque.append(
895 "--- test setUp() for %s.%s(%s) starts here ---\n"
896 % (self.__class__.__name__, self._testMethodName, self._testMethodDoc)
898 self.vapi.cli("clear trace")
899 # store the test instance inside the test class - so that objects
900 # holding the class can access instance methods (like assertEqual)
901 type(self).test_instance = self
903 f"--- END setUp() {self.__class__.__name__}.{self._testMethodName}('{self._testMethodDoc}') ---"
907 def get_vpp_time(cls):
908 # processes e.g. "Time now 2.190522, Wed, 11 Mar 2020 17:29:54 GMT"
909 # returns float("2.190522")
910 timestr = cls.vapi.cli("show clock")
911 head, sep, tail = timestr.partition(",")
912 head, sep, tail = head.partition("Time now")
916 def sleep_on_vpp_time(cls, sec):
917 """Sleep according to time in VPP world"""
918 # On a busy system with many processes
919 # we might end up with VPP time being slower than real world
920 # So take that into account when waiting for VPP to do something
921 start_time = cls.get_vpp_time()
922 while cls.get_vpp_time() - start_time < sec:
926 def create_loopback_interfaces(cls, count):
928 Create loopback interfaces.
930 :param count: number of interfaces created.
931 :returns: List of created interfaces.
933 result = [VppLoInterface(cls) for i in range(count)]
935 setattr(cls, intf.name, intf)
936 cls.lo_interfaces = result
939 def assert_equal(self, real_value, expected_value, name_or_class=None):
940 if name_or_class is None:
941 self.assertEqual(real_value, expected_value)
944 msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
946 getdoc(name_or_class).strip(),
948 str(name_or_class(real_value)),
950 str(name_or_class(expected_value)),
953 msg = "Invalid %s: %s does not match expected value %s" % (
959 self.assertEqual(real_value, expected_value, msg)
961 def assert_in_range(self, real_value, expected_min, expected_max, name=None):
965 msg = "Invalid %s: %s out of range <%s,%s>" % (
971 self.assertTrue(expected_min <= real_value <= expected_max, msg)
973 def get_counter(self, counter):
974 if counter.startswith("/"):
975 counter_value = self.statistics.get_counter(counter)
977 counters = self.vapi.cli("sh errors").split("\n")
979 for i in range(1, len(counters) - 1):
980 results = counters[i].split()
981 if results[1] == counter:
982 counter_value = int(results[0])
986 def assert_counter_equal(self, counter, expected_value, thread=None, index=0):
987 c = self.get_counter(counter)
988 if thread is not None:
991 c = sum(x[index] for x in c)
993 "validate counter `%s[%s]', expected: %s, real value: %s"
994 % (counter, index, expected_value, c)
996 self.assert_equal(c, expected_value, "counter `%s[%s]'" % (counter, index))
998 def assert_error_counter_equal(self, counter, expected_value):
999 counter_value = self.statistics[counter].sum()
1000 self.assert_equal(counter_value, expected_value, "error counter `%s'" % counter)
1003 def sleep(cls, timeout, remark=None):
1004 # /* Allow sleep(0) to maintain win32 semantics, and as decreed
1005 # * by Guido, only the main thread can be interrupted.
1007 # https://github.com/python/cpython/blob/6673decfa0fb078f60587f5cb5e98460eea137c2/Modules/timemodule.c#L1892 # noqa
1010 if hasattr(os, "sched_yield"):
1016 cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
1017 before = time.time()
1020 if after - before > 2 * timeout:
1022 "unexpected self.sleep() result - slept for %es instead of ~%es!",
1028 "Finished sleep (%s) - slept %es (wanted %es)",
1034 def virtual_sleep(self, timeout, remark=None):
1035 self.logger.debug("Moving VPP time by %s (%s)", timeout, remark)
1036 self.vapi.cli("set clock adjust %s" % timeout)
1038 def snapshot_stats(self, stats_diff):
1039 """Return snapshot of interesting stats based on diff dictionary."""
1041 for sw_if_index in stats_diff:
1042 for counter in stats_diff[sw_if_index]:
1043 stats_snapshot[counter] = self.statistics[counter]
1044 self.logger.debug(f"Took statistics stats_snapshot: {stats_snapshot}")
1045 return stats_snapshot
1047 def compare_stats_with_snapshot(self, stats_diff, stats_snapshot):
1048 """Assert appropriate difference between current stats and snapshot."""
1049 for sw_if_index in stats_diff:
1050 for cntr, diff in stats_diff[sw_if_index].items():
1051 if sw_if_index == "err":
1053 self.statistics[cntr].sum(),
1054 stats_snapshot[cntr].sum() + diff,
1055 f"'{cntr}' counter value (previous value: "
1056 f"{stats_snapshot[cntr].sum()}, "
1057 f"expected diff: {diff})",
1062 self.statistics[cntr][:, sw_if_index].sum(),
1063 stats_snapshot[cntr][:, sw_if_index].sum() + diff,
1064 f"'{cntr}' counter value (previous value: "
1065 f"{stats_snapshot[cntr][:, sw_if_index].sum()}, "
1066 f"expected diff: {diff})",
1068 except IndexError as e:
1069 # if diff is 0, then this most probably a case where
1070 # test declares multiple interfaces but traffic hasn't
1071 # passed through this one yet - which means the counter
1072 # value is 0 and can be ignored
1075 f"Couldn't sum counter: {cntr} on sw_if_index: {sw_if_index}"
1079 def get_testcase_doc_name(test):
1080 return getdoc(test.__class__).splitlines()[0]
1083 def get_test_description(descriptions, test):
1084 short_description = test.shortDescription()
1085 if descriptions and short_description:
1086 return short_description
1091 def get_failed_testcase_linkname(failed_dir, testcase_dirname):
1092 return os.path.join(failed_dir, f"{testcase_dirname}-FAILED")
1095 def get_testcase_dirname(testcase_class_name):
1096 return f"vpp-unittest-{testcase_class_name}"
1099 class TestCaseInfo(object):
1100 def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
1101 self.logger = logger
1102 self.tempdir = tempdir
1103 self.vpp_pid = vpp_pid
1104 self.vpp_bin_path = vpp_bin_path
1105 self.core_crash_test = None
1108 class VppTestResult(unittest.TestResult):
1110 @property result_string
1111 String variable to store the test case result string.
1113 List variable containing 2-tuples of TestCase instances and strings
1114 holding formatted tracebacks. Each tuple represents a test which
1115 raised an unexpected exception.
1117 List variable containing 2-tuples of TestCase instances and strings
1118 holding formatted tracebacks. Each tuple represents a test where
1119 a failure was explicitly signalled using the TestCase.assert*()
1123 failed_test_cases_info = set()
1124 core_crash_test_cases_info = set()
1125 current_test_case_info = None
1127 def __init__(self, stream=None, descriptions=None, verbosity=None, runner=None):
1129 :param stream File descriptor to store where to report test results.
1130 Set to the standard error stream by default.
1131 :param descriptions Boolean variable to store information if to use
1132 test case descriptions.
1133 :param verbosity Integer variable to store required verbosity level.
1135 super(VppTestResult, self).__init__(stream, descriptions, verbosity)
1136 self.stream = stream
1137 self.descriptions = descriptions
1138 self.verbosity = verbosity
1139 self.result_code = TestResultCode.TEST_RUN
1140 self.result_string = None
1141 self.runner = runner
1144 def decodePcapFiles(self, test, when_configured=False):
1145 if when_configured == False or config.decode_pcaps == True:
1146 if hasattr(test, "pg_interfaces") and len(test.pg_interfaces) > 0:
1147 testcase_dir = os.path.dirname(test.pg_interfaces[0].out_path)
1148 test.pg_interfaces[0].decode_pcap_files(
1149 testcase_dir, f"suite{test.__class__.__name__}"
1151 test.pg_interfaces[0].decode_pcap_files(
1152 testcase_dir, test._testMethodName
1155 def addSuccess(self, test):
1157 Record a test succeeded result
1162 self.log_result("addSuccess", test)
1163 self.decodePcapFiles(test, when_configured=True)
1164 unittest.TestResult.addSuccess(self, test)
1165 self.result_string = colorize("OK", GREEN)
1166 self.result_code = TestResultCode.PASS
1167 self.send_result_through_pipe(test, self.result_code)
1169 def addExpectedFailure(self, test, err):
1170 self.log_result("addExpectedFailure", test, err)
1171 self.decodePcapFiles(test)
1172 super().addExpectedFailure(test, err)
1173 self.result_string = colorize("FAIL", GREEN)
1174 self.result_code = TestResultCode.EXPECTED_FAIL
1175 self.send_result_through_pipe(test, self.result_code)
1177 def addUnexpectedSuccess(self, test):
1178 self.log_result("addUnexpectedSuccess", test)
1179 self.decodePcapFiles(test, when_configured=True)
1180 super().addUnexpectedSuccess(test)
1181 self.result_string = colorize("OK", RED)
1182 self.result_code = TestResultCode.UNEXPECTED_PASS
1183 self.send_result_through_pipe(test, self.result_code)
1185 def addSkip(self, test, reason):
1187 Record a test skipped.
1193 self.log_result("addSkip", test, reason=reason)
1194 unittest.TestResult.addSkip(self, test, reason)
1195 self.result_string = colorize("SKIP", YELLOW)
1197 if reason == "not enough cpus":
1198 self.result_code = TestResultCode.SKIP_CPU_SHORTAGE
1200 self.result_code = TestResultCode.SKIP
1201 self.send_result_through_pipe(test, self.result_code)
1203 def symlink_failed(self):
1204 if self.current_test_case_info:
1206 failed_dir = config.failed_dir
1207 link_path = get_failed_testcase_linkname(
1208 failed_dir, os.path.basename(self.current_test_case_info.tempdir)
1211 self.current_test_case_info.logger.debug(
1212 "creating a link to the failed test"
1214 self.current_test_case_info.logger.debug(
1215 "os.symlink(%s, %s)"
1216 % (self.current_test_case_info.tempdir, link_path)
1218 if os.path.exists(link_path):
1219 self.current_test_case_info.logger.debug("symlink already exists")
1221 os.symlink(self.current_test_case_info.tempdir, link_path)
1223 except Exception as e:
1224 self.current_test_case_info.logger.error(e)
1226 def send_result_through_pipe(self, test, result):
1227 if hasattr(self, "test_framework_result_pipe"):
1228 pipe = self.test_framework_result_pipe
1230 pipe.send((test.id(), result))
1232 def log_result(self, fn, test, err=None, reason=None):
1233 if self.current_test_case_info:
1234 if isinstance(test, unittest.suite._ErrorHolder):
1235 test_name = test.description
1237 test_name = "%s.%s(%s)" % (
1238 test.__class__.__name__,
1239 test._testMethodName,
1240 test._testMethodDoc,
1244 extra_msg += f", error is {err}"
1246 extra_msg += f", reason is {reason}"
1247 self.current_test_case_info.logger.debug(
1248 f"--- {fn}() {test_name} called{extra_msg}"
1251 self.current_test_case_info.logger.debug(
1252 "formatted exception is:\n%s" % "".join(format_exception(*err))
1255 def add_error(self, test, err, unittest_fn, result_code):
1256 self.result_code = result_code
1257 if result_code == TestResultCode.FAIL:
1258 self.log_result("addFailure", test, err=err)
1259 error_type_str = colorize("FAIL", RED)
1260 elif result_code == TestResultCode.ERROR:
1261 self.log_result("addError", test, err=err)
1262 error_type_str = colorize("ERROR", RED)
1264 raise Exception(f"Unexpected result code {result_code}")
1265 self.decodePcapFiles(test)
1267 unittest_fn(self, test, err)
1268 if self.current_test_case_info:
1269 self.result_string = "%s [ temp dir used by test case: %s ]" % (
1271 self.current_test_case_info.tempdir,
1273 self.symlink_failed()
1274 self.failed_test_cases_info.add(self.current_test_case_info)
1275 if is_core_present(self.current_test_case_info.tempdir):
1276 if not self.current_test_case_info.core_crash_test:
1277 if isinstance(test, unittest.suite._ErrorHolder):
1278 test_name = str(test)
1280 test_name = "'{!s}' ({!s})".format(
1281 get_testcase_doc_name(test), test.id()
1283 self.current_test_case_info.core_crash_test = test_name
1284 self.core_crash_test_cases_info.add(self.current_test_case_info)
1286 self.result_string = "%s [no temp dir]" % error_type_str
1288 self.send_result_through_pipe(test, result_code)
1290 def addFailure(self, test, err):
1292 Record a test failed result
1295 :param err: error message
1298 self.add_error(test, err, unittest.TestResult.addFailure, TestResultCode.FAIL)
1300 def addError(self, test, err):
1302 Record a test error result
1305 :param err: error message
1308 self.add_error(test, err, unittest.TestResult.addError, TestResultCode.ERROR)
1310 def getDescription(self, test):
1312 Get test description
1315 :returns: test description
1318 return get_test_description(self.descriptions, test)
1320 def startTest(self, test):
1328 def print_header(test):
1329 if test.__class__ in self.printed:
1332 test_doc = getdoc(test)
1334 raise Exception("No doc string for test '%s'" % test.id())
1336 test_title = test_doc.splitlines()[0].rstrip()
1337 test_title = colorize(test_title, GREEN)
1338 if test.is_tagged_run_solo():
1339 test_title = colorize(f"SOLO RUN: {test_title}", YELLOW)
1341 # This block may overwrite the colorized title above,
1342 # but we want this to stand out and be fixed
1343 if test.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
1344 test_title = colorize(f"FIXME with VPP workers: {test_title}", RED)
1346 if test.has_tag(TestCaseTag.FIXME_ASAN):
1347 test_title = colorize(f"FIXME with ASAN: {test_title}", RED)
1348 test.skip_fixme_asan()
1350 if hasattr(test, "vpp_worker_count"):
1351 if test.vpp_worker_count == 0:
1352 test_title += " [main thread only]"
1353 elif test.vpp_worker_count == 1:
1354 test_title += " [1 worker thread]"
1356 test_title += f" [{test.vpp_worker_count} worker threads]"
1358 if test.__class__.skipped_due_to_cpu_lack:
1359 test_title = colorize(
1360 f"{test_title} [skipped - not enough cpus, "
1361 f"required={test.__class__.get_cpus_required()}, "
1362 f"available={max_vpp_cpus}]",
1366 print(double_line_delim)
1368 print(double_line_delim)
1369 self.printed.append(test.__class__)
1372 self.start_test = time.time()
1373 unittest.TestResult.startTest(self, test)
1374 if self.verbosity > 0:
1375 self.stream.writeln("Starting " + self.getDescription(test) + " ...")
1376 self.stream.writeln(single_line_delim)
1378 def stopTest(self, test):
1380 Called when the given test has been run
1385 unittest.TestResult.stopTest(self, test)
1387 result_code_to_suffix = {
1388 TestResultCode.PASS: "",
1389 TestResultCode.FAIL: "",
1390 TestResultCode.ERROR: "",
1391 TestResultCode.SKIP: "",
1392 TestResultCode.TEST_RUN: "",
1393 TestResultCode.SKIP_CPU_SHORTAGE: "",
1394 TestResultCode.EXPECTED_FAIL: " [EXPECTED FAIL]",
1395 TestResultCode.UNEXPECTED_PASS: " [UNEXPECTED PASS]",
1398 if self.verbosity > 0:
1399 self.stream.writeln(single_line_delim)
1400 self.stream.writeln(
1403 self.getDescription(test),
1405 result_code_to_suffix[self.result_code],
1408 self.stream.writeln(single_line_delim)
1410 self.stream.writeln(
1413 self.getDescription(test),
1414 time.time() - self.start_test,
1416 result_code_to_suffix[self.result_code],
1420 self.send_result_through_pipe(test, TestResultCode.TEST_RUN)
1422 def printErrors(self):
1424 Print errors from running the test case
1426 if len(self.errors) > 0 or len(self.failures) > 0:
1427 self.stream.writeln()
1428 self.printErrorList("ERROR", self.errors)
1429 self.printErrorList("FAIL", self.failures)
1431 # ^^ that is the last output from unittest before summary
1432 if not self.runner.print_summary:
1433 devnull = unittest.runner._WritelnDecorator(open(os.devnull, "w"))
1434 self.stream = devnull
1435 self.runner.stream = devnull
1437 def printErrorList(self, flavour, errors):
1439 Print error list to the output stream together with error type
1440 and test case description.
1442 :param flavour: error type
1443 :param errors: iterable errors
1446 for test, err in errors:
1447 self.stream.writeln(double_line_delim)
1448 self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
1449 self.stream.writeln(single_line_delim)
1450 self.stream.writeln("%s" % err)
1453 class VppTestRunner(unittest.TextTestRunner):
1455 A basic test runner implementation which prints results to standard error.
1459 def resultclass(self):
1460 """Class maintaining the results of the tests"""
1461 return VppTestResult
1465 keep_alive_pipe=None,
1475 # ignore stream setting here, use hard-coded stdout to be in sync
1476 # with prints from VppAsfTestCase methods ...
1477 super(VppTestRunner, self).__init__(
1478 sys.stdout, descriptions, verbosity, failfast, buffer, resultclass, **kwargs
1480 KeepAliveReporter.pipe = keep_alive_pipe
1482 self.orig_stream = self.stream
1483 self.resultclass.test_framework_result_pipe = result_pipe
1485 self.print_summary = print_summary
1487 def _makeResult(self):
1488 return self.resultclass(self.stream, self.descriptions, self.verbosity, self)
1490 def run(self, test):
1497 faulthandler.enable() # emit stack trace to stderr if killed by signal
1499 result = super(VppTestRunner, self).run(test)
1500 if not self.print_summary:
1501 self.stream = self.orig_stream
1502 result.stream = self.orig_stream
1506 class Worker(Thread):
1507 def __init__(self, executable_args, logger, env=None, *args, **kwargs):
1508 super(Worker, self).__init__(*args, **kwargs)
1509 self.logger = logger
1510 self.args = executable_args
1511 if hasattr(self, "testcase") and self.testcase.debug_all:
1512 if self.testcase.debug_gdbserver:
1514 "/usr/bin/gdbserver",
1515 "localhost:{port}".format(port=self.testcase.gdbserver_port),
1517 elif self.testcase.debug_gdb and hasattr(self, "wait_for_gdb"):
1518 self.args.append(self.wait_for_gdb)
1519 self.app_bin = executable_args[0]
1520 self.app_name = os.path.basename(self.app_bin)
1521 if hasattr(self, "role"):
1522 self.app_name += " {role}".format(role=self.role)
1525 env = {} if env is None else env
1526 self.env = copy.deepcopy(env)
1528 def wait_for_enter(self):
1529 if not hasattr(self, "testcase"):
1531 if self.testcase.debug_all and self.testcase.debug_gdbserver:
1533 print(double_line_delim)
1535 "Spawned GDB Server for '{app}' with PID: {pid}".format(
1536 app=self.app_name, pid=self.process.pid
1539 elif self.testcase.debug_all and self.testcase.debug_gdb:
1541 print(double_line_delim)
1543 "Spawned '{app}' with PID: {pid}".format(
1544 app=self.app_name, pid=self.process.pid
1549 print(single_line_delim)
1550 print("You can debug '{app}' using:".format(app=self.app_name))
1551 if self.testcase.debug_gdbserver:
1555 + " -ex 'target remote localhost:{port}'".format(
1556 port=self.testcase.gdbserver_port
1560 "Now is the time to attach gdb by running the above "
1561 "command, set up breakpoints etc., then resume from "
1562 "within gdb by issuing the 'continue' command"
1564 self.testcase.gdbserver_port += 1
1565 elif self.testcase.debug_gdb:
1569 + " -ex 'attach {pid}'".format(pid=self.process.pid)
1572 "Now is the time to attach gdb by running the above "
1573 "command and set up breakpoints etc., then resume from"
1574 " within gdb by issuing the 'continue' command"
1576 print(single_line_delim)
1577 input("Press ENTER to continue running the testcase...")
1580 executable = self.args[0]
1581 if not os.path.exists(executable) or not os.access(
1582 executable, os.F_OK | os.X_OK
1584 # Exit code that means some system file did not exist,
1585 # could not be opened, or had some other kind of error.
1586 self.result = os.EX_OSFILE
1587 raise EnvironmentError(
1588 "executable '%s' is not found or executable." % executable
1591 "Running executable '{app}': '{cmd}'".format(
1592 app=self.app_name, cmd=" ".join(self.args)
1595 env = os.environ.copy()
1596 env.update(self.env)
1597 env["CK_LOG_FILE_NAME"] = "-"
1598 self.process = subprocess.Popen(
1599 ["stdbuf", "-o0", "-e0"] + self.args,
1602 preexec_fn=os.setpgrp,
1603 stdout=subprocess.PIPE,
1604 stderr=subprocess.PIPE,
1606 self.wait_for_enter()
1607 out, err = self.process.communicate()
1608 self.logger.debug("Finished running `{app}'".format(app=self.app_name))
1609 self.logger.info("Return code is `%s'" % self.process.returncode)
1610 self.logger.info(single_line_delim)
1612 "Executable `{app}' wrote to stdout:".format(app=self.app_name)
1614 self.logger.info(single_line_delim)
1615 self.logger.info(out.decode("utf-8"))
1616 self.logger.info(single_line_delim)
1618 "Executable `{app}' wrote to stderr:".format(app=self.app_name)
1620 self.logger.info(single_line_delim)
1621 self.logger.info(err.decode("utf-8"))
1622 self.logger.info(single_line_delim)
1623 self.result = self.process.returncode
1626 if __name__ == "__main__":