#!/usr/bin/env python3 from __future__ import print_function import logging import sys import os import select import signal import subprocess import unittest import re import time import faulthandler import random import copy import platform import shutil from collections import deque from threading import Thread, Event from inspect import getdoc, isclass from traceback import format_exception from logging import FileHandler, DEBUG, Formatter from enum import Enum from abc import ABC, abstractmethod from struct import pack, unpack import scapy.compat from scapy.packet import Raw, Packet from config import config, available_cpus, num_cpus, max_vpp_cpus import hook as hookmodule from vpp_pg_interface import VppPGInterface from vpp_sub_interface import VppSubInterface from vpp_lo_interface import VppLoInterface from vpp_bvi_interface import VppBviInterface from vpp_papi_provider import VppPapiProvider from vpp_papi import VppEnum import vpp_papi from vpp_papi.vpp_stats import VPPStats from vpp_papi.vpp_transport_socket import VppTransportSocketIOError from log import ( RED, GREEN, YELLOW, double_line_delim, single_line_delim, get_logger, colorize, ) from vpp_object import VppObjectRegistry from util import ppp, is_core_present from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest from scapy.layers.inet6 import ICMPv6EchoReply from vpp_running import use_running logger = logging.getLogger(__name__) # Set up an empty logger for the testcase that can be overridden as necessary null_logger = logging.getLogger("VppTestCase") null_logger.addHandler(logging.NullHandler()) PASS = 0 FAIL = 1 ERROR = 2 SKIP = 3 TEST_RUN = 4 SKIP_CPU_SHORTAGE = 5 if config.debug_framework: import debug_internal """ Test framework module. The module provides a set of tools for constructing and running tests and representing the results. """ class VppDiedError(Exception): """exception for reporting that the subprocess has died.""" signals_by_value = { v: k for k, v in signal.__dict__.items() if k.startswith("SIG") and not k.startswith("SIG_") } def __init__(self, rv=None, testcase=None, method_name=None): self.rv = rv self.signal_name = None self.testcase = testcase self.method_name = method_name try: self.signal_name = VppDiedError.signals_by_value[-rv] except (KeyError, TypeError): pass if testcase is None and method_name is None: in_msg = "" else: in_msg = " while running %s.%s" % (testcase, method_name) if self.rv: msg = "VPP subprocess died unexpectedly%s with return code: %d%s." % ( in_msg, self.rv, " [%s]" % (self.signal_name if self.signal_name is not None else ""), ) else: msg = "VPP subprocess died unexpectedly%s." % in_msg super(VppDiedError, self).__init__(msg) class _PacketInfo(object): """Private class to create packet info object. Help process information about the next packet. Set variables to default values. """ #: Store the index of the packet. index = -1 #: Store the index of the source packet generator interface of the packet. src = -1 #: Store the index of the destination packet generator interface #: of the packet. dst = -1 #: Store expected ip version ip = -1 #: Store expected upper protocol proto = -1 #: Store the copy of the former packet. data = None def __eq__(self, other): index = self.index == other.index src = self.src == other.src dst = self.dst == other.dst data = self.data == other.data return index and src and dst and data def pump_output(testclass): """pump output from vpp stdout/stderr to proper queues""" if not hasattr(testclass, "vpp"): return stdout_fragment = "" stderr_fragment = "" while not testclass.pump_thread_stop_flag.is_set(): readable = select.select( [ testclass.vpp.stdout.fileno(), testclass.vpp.stderr.fileno(), testclass.pump_thread_wakeup_pipe[0], ], [], [], )[0] if testclass.vpp.stdout.fileno() in readable: read = os.read(testclass.vpp.stdout.fileno(), 102400) if len(read) > 0: split = read.decode("ascii", errors="backslashreplace").splitlines(True) if len(stdout_fragment) > 0: split[0] = "%s%s" % (stdout_fragment, split[0]) if len(split) > 0 and split[-1].endswith("\n"): limit = None else: limit = -1 stdout_fragment = split[-1] testclass.vpp_stdout_deque.extend(split[:limit]) if not config.cache_vpp_output: for line in split[:limit]: testclass.logger.info("VPP STDOUT: %s" % line.rstrip("\n")) if testclass.vpp.stderr.fileno() in readable: read = os.read(testclass.vpp.stderr.fileno(), 102400) if len(read) > 0: split = read.decode("ascii", errors="backslashreplace").splitlines(True) if len(stderr_fragment) > 0: split[0] = "%s%s" % (stderr_fragment, split[0]) if len(split) > 0 and split[-1].endswith("\n"): limit = None else: limit = -1 stderr_fragment = split[-1] testclass.vpp_stderr_deque.extend(split[:limit]) if not config.cache_vpp_output: for line in split[:limit]: testclass.logger.error("VPP STDERR: %s" % line.rstrip("\n")) # ignoring the dummy pipe here intentionally - the # flag will take care of properly terminating the loop def _is_platform_aarch64(): return platform.machine() == "aarch64" is_platform_aarch64 = _is_platform_aarch64() def _is_distro_ubuntu2204(): with open("/etc/os-release") as f: for line in f.readlines(): if "jammy" in line: return True return False is_distro_ubuntu2204 = _is_distro_ubuntu2204() def _is_distro_debian11(): with open("/etc/os-release") as f: for line in f.readlines(): if "bullseye" in line: return True return False is_distro_debian11 = _is_distro_debian11() class KeepAliveReporter(object): """ Singleton object which reports test start to parent process """ _shared_state = {} def __init__(self): self.__dict__ = self._shared_state self._pipe = None @property def pipe(self): return self._pipe @pipe.setter def pipe(self, pipe): if self._pipe is not None: raise Exception("Internal error - pipe should only be set once.") self._pipe = pipe def send_keep_alive(self, test, desc=None): """ Write current test tmpdir & desc to keep-alive pipe to signal liveness """ if not hasattr(test, "vpp") or self.pipe is None: # if not running forked.. return if isclass(test): desc = "%s (%s)" % (desc, unittest.util.strclass(test)) else: desc = test.id() self.pipe.send((desc, config.vpp, test.tempdir, test.vpp.pid)) class TestCaseTag(Enum): # marks the suites that must run at the end # using only a single test runner RUN_SOLO = 1 # marks the suites broken on VPP multi-worker FIXME_VPP_WORKERS = 2 # marks the suites broken when ASan is enabled FIXME_ASAN = 3 # marks suites broken on Ubuntu-22.04 FIXME_UBUNTU2204 = 4 # marks suites broken on Debian-11 FIXME_DEBIAN11 = 5 # marks suites broken on debug vpp image FIXME_VPP_DEBUG = 6 def create_tag_decorator(e): def decorator(cls): try: cls.test_tags.append(e) except AttributeError: cls.test_tags = [e] return cls return decorator tag_run_solo = create_tag_decorator(TestCaseTag.RUN_SOLO) tag_fixme_vpp_workers = create_tag_decorator(TestCaseTag.FIXME_VPP_WORKERS) tag_fixme_asan = create_tag_decorator(TestCaseTag.FIXME_ASAN) tag_fixme_ubuntu2204 = create_tag_decorator(TestCaseTag.FIXME_UBUNTU2204) tag_fixme_debian11 = create_tag_decorator(TestCaseTag.FIXME_DEBIAN11) tag_fixme_vpp_debug = create_tag_decorator(TestCaseTag.FIXME_VPP_DEBUG) class DummyVpp: returncode = None pid = 0xCAFEBAFE def poll(self): pass def terminate(self): pass class CPUInterface(ABC): cpus = [] skipped_due_to_cpu_lack = False @classmethod @abstractmethod def get_cpus_required(cls): pass @classmethod def assign_cpus(cls, cpus): cls.cpus = cpus @use_running class VppTestCase(CPUInterface, unittest.TestCase): """This subclass is a base class for VPP test cases that are implemented as classes. It provides methods to create and run test case. """ extra_vpp_statseg_config = "" extra_vpp_config = [] extra_vpp_plugin_config = [] logger = null_logger vapi_response_timeout = 5 remove_configured_vpp_objects_on_tear_down = True @property def packet_infos(self): """List of packet infos""" return self._packet_infos @classmethod def get_packet_count_for_if_idx(cls, dst_if_index): """Get the number of packet info for specified destination if index""" if dst_if_index in cls._packet_count_for_dst_if_idx: return cls._packet_count_for_dst_if_idx[dst_if_index] else: return 0 @classmethod def has_tag(cls, tag): """if the test case has a given tag - return true""" try: return tag in cls.test_tags except AttributeError: pass return False @classmethod def is_tagged_run_solo(cls): """if the test case class is timing-sensitive - return true""" return cls.has_tag(TestCaseTag.RUN_SOLO) @classmethod def skip_fixme_asan(cls): """if @tag_fixme_asan & ASan is enabled - mark for skip""" if cls.has_tag(TestCaseTag.FIXME_ASAN): vpp_extra_cmake_args = os.environ.get("VPP_EXTRA_CMAKE_ARGS", "") if "DVPP_ENABLE_SANITIZE_ADDR=ON" in vpp_extra_cmake_args: cls = unittest.skip("Skipping @tag_fixme_asan tests")(cls) @classmethod def skip_fixme_ubuntu2204(cls): """if distro is ubuntu 22.04 and @tag_fixme_ubuntu2204 mark for skip""" if cls.has_tag(TestCaseTag.FIXME_UBUNTU2204): cls = unittest.skip("Skipping @tag_fixme_ubuntu2204 tests")(cls) @classmethod def skip_fixme_debian11(cls): """if distro is Debian-11 and @tag_fixme_debian11 mark for skip""" if cls.has_tag(TestCaseTag.FIXME_DEBIAN11): cls = unittest.skip("Skipping @tag_fixme_debian11 tests")(cls) @classmethod def skip_fixme_vpp_debug(cls): cls = unittest.skip("Skipping @tag_fixme_vpp_debug tests")(cls) @classmethod def instance(cls): """Return the instance of this testcase""" return cls.test_instance @classmethod def set_debug_flags(cls, d): cls.gdbserver_port = 7777 cls.debug_core = False cls.debug_gdb = False cls.debug_gdbserver = False cls.debug_all = False cls.debug_attach = False if d is None: return dl = d.lower() if dl == "core": cls.debug_core = True elif dl == "gdb" or dl == "gdb-all": cls.debug_gdb = True elif dl == "gdbserver" or dl == "gdbserver-all": cls.debug_gdbserver = True elif dl == "attach": cls.debug_attach = True else: raise Exception("Unrecognized DEBUG option: '%s'" % d) if dl == "gdb-all" or dl == "gdbserver-all": cls.debug_all = True @classmethod def get_vpp_worker_count(cls): if not hasattr(cls, "vpp_worker_count"): if cls.has_tag(TestCaseTag.FIXME_VPP_WORKERS): cls.vpp_worker_count = 0 else: cls.vpp_worker_count = config.vpp_worker_count return cls.vpp_worker_count @classmethod def get_cpus_required(cls): return 1 + cls.get_vpp_worker_count() @classmethod def setUpConstants(cls): """Set-up the test case class based on environment variables""" cls.step = config.step cls.plugin_path = ":".join(config.vpp_plugin_dir) cls.test_plugin_path = ":".
/*
 * Copyright (c) 2016 Cisco and/or its affiliates.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <lb/lb.h>
#include <lb/util.h>

static clib_error_t *
lb_vip_command_fn (vlib_main_t * vm,
              unformat_input_t * input, vlib_cli_command_t * cmd)
{
  unformat_input_t _line_input, *line_input = &_line_input;
  ip46_address_t prefix;
  u8 plen;
  u32 new_len = 1024;
  u8 del = 0;
  int ret;
  u32 gre4 = 0;
  lb_vip_type_t type;

  if (!unformat_user (input, unformat_line_input, line_input))
    return 0;

  if (!unformat(line_input, "%U", unformat_ip46_prefix, &prefix, &plen, IP46_TYPE_ANY, &plen))
    return clib_error_return (0, "invalid vip prefix: '%U'",
                                  format_unformat_error, line_input);

  while (unformat_check_input (line_input) != UNFORMAT_END_OF_INPUT)
  {
    if (unformat(line_input, "new_len %d", &new_len))
      ;
    else if (unformat(line_input, "del"))
      del = 1;
    else if (unformat(line_input, "encap gre4"))
      gre4 = 1;
    else if (unformat(line_input, "encap gre6"))
      gre4 = 0;
    else
      return clib_error_return (0, "parse error: '%U'",
                              format_unformat_error, line_input);
  }

  unformat_free (line_input);


  if (ip46_prefix_is_ip4(&prefix, plen)) {
    type = (gre4)?LB_VIP_TYPE_IP4_GRE4:LB_VIP_TYPE_IP4_GRE6;
  } else {
    type = (gre4)?LB_VIP_TYPE_IP6_GRE4:LB_VIP_TYPE_IP6_GRE6;
  }

  lb_garbage_collection();

  u32 index;
  if (!del) {
    if ((ret = lb_vip_add(&prefix, plen, type, new_len, &index))) {
      return clib_error_return (0, "lb_vip_add error %d", ret);
    } else {
      vlib_cli_output(vm, "lb_vip_add ok %d", index);
    }
  } else {
    if ((ret = lb_vip_find_index(&prefix, plen, &index)))
      return clib_error_return (0, "lb_vip_find_index error %d", ret);
    else if ((ret = lb_vip_del(index)))
      return clib_error_return (0, "lb_vip_del error %d", ret);
  }
  return NULL;
}

VLIB_CLI_COMMAND (lb_vip_command, static) =
{
  .path = "lb vip",
  .short_help = "lb vip <prefix> [encap (gre6|gre4)] [new_len <n>] [del]",
  .function = lb_vip_command_fn,
};

static clib_error_t *
lb_as_command_fn (vlib_main_t * vm,
              unformat_input_t * input, vlib_cli_command_t * cmd)
{
  unformat_input_t _line_input, *line_input = &_line_input;
  ip46_address_t vip_prefix, as_addr;
  u8 vip_plen;
  ip46_address_t *as_array = 0;
  u32 vip_index;
  u8 del = 0;
  int ret;

  if (!unformat_user (input, unformat_line_input, line_input))
    return 0;

  if (!unformat(line_input, "%U", unformat_ip46_prefix, &vip_prefix, &vip_plen, IP46_TYPE_ANY))
    return clib_error_return (0, "invalid as address: '%U'",
                              format_unformat_error, line_input);

  if ((ret = lb_vip_find_index(&vip_prefix, vip_plen, &vip_index)))
    return clib_error_return (0, "lb_vip_find_index error %d", ret);

  while (unformat_check_input (line_input) != UNFORMAT_END_OF_INPUT)
  {
    if (unformat(line_input, "%U", unformat_ip46_address, &as_addr, IP46_TYPE_ANY)) {
      vec_add1(as_array, as_addr);
    } else if (unformat(line_input, "del")) {
      del = 1;
    } else {
      vec_free(as_array);
      return clib_error_return (0, "parse error: '%U'",
                                format_unformat_error, line_input);
    }
  }

  if (!vec_len(as_array)) {
    vec_free(as_array);
    return clib_error_return (0, "No AS address provided");
  }

  lb_garbage_collection();
  clib_warning("vip index is %d", vip_index);

  if (del) {
    if ((ret = lb_vip_del_ass(vip_index, as_array, vec_len(as_array)))) {
      vec_free(as_array);
      return clib_error_return (0, "lb_vip_del_ass error %d", ret);
    }
  } else {
    if ((ret = lb_vip_add_ass(vip_index, as_array, vec_len(as_array)))) {
      vec_free(as_array);
      return clib_error_return (0, "lb_vip_add_ass error %d", ret);
    }
  }

  vec_free(as_array);
  return 0;
}

VLIB_CLI_COMMAND (lb_as_command, static) =
{
  .path = "lb as",
  .short_help = "lb as <vip-prefix> [<address> [<address> [...]]] [del]",
  .function = lb_as_command_fn,
};

static clib_error_t *
lb_conf_command_fn (vlib_main_t * vm,
              unformat_input_t * input, vlib_cli_command_t * cmd)
{
  lb_main_t *lbm = &lb_main;
  unformat_input_t _line_input, *line_input = &_line_input;
  ip4_address_t ip4 = lbm->ip4_src_address;
  ip6_address_t ip6 = lbm->ip6_src_address;
  u32 per_cpu_sticky_buckets = lbm->per_cpu_sticky_buckets;
  u32 per_cpu_sticky_buckets_log2 = 0;
  u32 flow_timeout = lbm->flow_timeout;
  int ret;

  if (!unformat_user (input, unformat_line_input, line_input))
    return 0;

  while (unformat_check_input (line_input) != UNFORMAT_END_OF_INPUT)
  {
    if (unformat(line_input, "ip4-src-address %U", unformat_ip4_address, &ip4))
      ;
    else if (unformat(line_input, "ip6-src-address %U", unformat_ip6_address, &ip6))
      ;
    else if (unformat(line_input, "buckets %d", &per_cpu_sticky_buckets))
      ;
    else if (unformat(line_input, "buckets-log2 %d", &per_cpu_sticky_buckets_log2)) {
      if (per_cpu_sticky_buckets_log2 >= 32)
        return clib_error_return (0, "buckets-log2 value is too high");
      per_cpu_sticky_buckets = 1 << per_cpu_sticky_buckets_log2;
    } else if (unformat(line_input, "timeout %d", &flow_timeout))
      ;
    else
      return clib_error_return (0, "parse error: '%U'",
                                format_unformat_error, line_input);
  }

  unformat_free (line_input);

  lb_garbage_collection();

  if ((ret = lb_conf(&ip4, &ip6, per_cpu_sticky_buckets, flow_timeout)))
    return clib_error_return (0, "lb_conf error %d", ret);

  return NULL;
}

VLIB_CLI_COMMAND (lb_conf_command, static) =
{
  .path = "lb conf",
  .short_help = "lb conf [ip4-src-address <addr>] [ip6-src-address <addr>] [buckets <n>] [timeout <s>]",
  .function = lb_conf_command_fn,
};

static clib_error_t *
lb_show_command_fn (vlib_main_t * vm,
              unformat_input_t * input, vlib_cli_command_t * cmd)
{
  vlib_cli_output(vm, "%U", format_lb_main);
  return NULL;
}


VLIB_CLI_COMMAND (lb_show_command, static) =
{
  .path = "show lb",
  .short_help = "show lb",
  .function = lb_show_command_fn,
};

static clib_error_t *
lb_show_vips_command_fn (vlib_main_t * vm,
              unformat_input_t * input, vlib_cli_command_t * cmd)
{
  unformat_input_t line_input;
  lb_main_t *lbm = &lb_main;
  lb_vip_t *vip;
  u8 verbose = 0;

  if (!unformat_user (input, unformat_line_input, &line_input))
      return 0;

  if (unformat(&line_input, "verbose"))
    verbose = 1;

  pool_foreach(vip, lbm->vips, {
      vlib_cli_output(vm, "%U\n", verbose?format_lb_vip_detailed:format_lb_vip, vip);
  });

  unformat_free (&line_input);
  return NULL;
}

VLIB_CLI_COMMAND (lb_show_vips_command, static) =
{
  .path = "show lb vips",
  .short_help = "show lb vips [verbose]",
  .function = lb_show_vips_command_fn,
};
self, intf, pkts, output, timeout=None, stats_diff=None): if stats_diff: stats_snapshot = self.snapshot_stats(stats_diff) self.pg_send(intf, pkts) rx = output.get_capture(len(pkts)) outputs = [output] if not timeout: timeout = 1 for i in self.pg_interfaces: if i not in outputs: i.assert_nothing_captured(timeout=timeout) timeout = 0.1 if stats_diff: self.compare_stats_with_snapshot(stats_diff, stats_snapshot) return rx def get_testcase_doc_name(test): return getdoc(test.__class__).splitlines()[0] def get_test_description(descriptions, test): short_description = test.shortDescription() if descriptions and short_description: return short_description else: return str(test) class TestCaseInfo(object): def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path): self.logger = logger self.tempdir = tempdir self.vpp_pid = vpp_pid self.vpp_bin_path = vpp_bin_path self.core_crash_test = None class VppTestResult(unittest.TestResult): """ @property result_string String variable to store the test case result string. @property errors List variable containing 2-tuples of TestCase instances and strings holding formatted tracebacks. Each tuple represents a test which raised an unexpected exception. @property failures List variable containing 2-tuples of TestCase instances and strings holding formatted tracebacks. Each tuple represents a test where a failure was explicitly signalled using the TestCase.assert*() methods. """ failed_test_cases_info = set() core_crash_test_cases_info = set() current_test_case_info = None def __init__(self, stream=None, descriptions=None, verbosity=None, runner=None): """ :param stream File descriptor to store where to report test results. Set to the standard error stream by default. :param descriptions Boolean variable to store information if to use test case descriptions. :param verbosity Integer variable to store required verbosity level. """ super(VppTestResult, self).__init__(stream, descriptions, verbosity) self.stream = stream self.descriptions = descriptions self.verbosity = verbosity self.result_string = None self.runner = runner self.printed = [] def addSuccess(self, test): """ Record a test succeeded result :param test: """ if self.current_test_case_info: self.current_test_case_info.logger.debug( "--- addSuccess() %s.%s(%s) called" % (test.__class__.__name__, test._testMethodName, test._testMethodDoc) ) unittest.TestResult.addSuccess(self, test) self.result_string = colorize("OK", GREEN) self.send_result_through_pipe(test, PASS) def addSkip(self, test, reason): """ Record a test skipped. :param test: :param reason: """ if self.current_test_case_info: self.current_test_case_info.logger.debug( "--- addSkip() %s.%s(%s) called, reason is %s" % ( test.__class__.__name__, test._testMethodName, test._testMethodDoc, reason, ) ) unittest.TestResult.addSkip(self, test, reason) self.result_string = colorize("SKIP", YELLOW) if reason == "not enough cpus": self.send_result_through_pipe(test, SKIP_CPU_SHORTAGE) else: self.send_result_through_pipe(test, SKIP) def symlink_failed(self): if self.current_test_case_info: try: failed_dir = config.failed_dir link_path = os.path.join( failed_dir, "%s-FAILED" % os.path.basename(self.current_test_case_info.tempdir), ) self.current_test_case_info.logger.debug( "creating a link to the failed test" ) self.current_test_case_info.logger.debug( "os.symlink(%s, %s)" % (self.current_test_case_info.tempdir, link_path) ) if os.path.exists(link_path): self.current_test_case_info.logger.debug("symlink already exists") else: os.symlink(self.current_test_case_info.tempdir, link_path) except Exception as e: self.current_test_case_info.logger.error(e) def send_result_through_pipe(self, test, result): if hasattr(self, "test_framework_result_pipe"): pipe = self.test_framework_result_pipe if pipe: pipe.send((test.id(), result)) def log_error(self, test, err, fn_name): if self.current_test_case_info: if isinstance(test, unittest.suite._ErrorHolder): test_name = test.description else: test_name = "%s.%s(%s)" % ( test.__class__.__name__, test._testMethodName, test._testMethodDoc, ) self.current_test_case_info.logger.debug( "--- %s() %s called, err is %s" % (fn_name, test_name, err) ) self.current_test_case_info.logger.debug( "formatted exception is:\n%s" % "".join(format_exception(*err)) ) def add_error(self, test, err, unittest_fn, error_type): if error_type == FAIL: self.log_error(test, err, "addFailure") error_type_str = colorize("FAIL", RED) elif error_type == ERROR: self.log_error(test, err, "addError") error_type_str = colorize("ERROR", RED) else: raise Exception( "Error type %s cannot be used to record an " "error or a failure" % error_type ) unittest_fn(self, test, err) if self.current_test_case_info: self.result_string = "%s [ temp dir used by test case: %s ]" % ( error_type_str, self.current_test_case_info.tempdir, ) self.symlink_failed() self.failed_test_cases_info.add(self.current_test_case_info) if is_core_present(self.current_test_case_info.tempdir): if not self.current_test_case_info.core_crash_test: if isinstance(test, unittest.suite._ErrorHolder): test_name = str(test) else: test_name = "'{!s}' ({!s})".format( get_testcase_doc_name(test), test.id() ) self.current_test_case_info.core_crash_test = test_name self.core_crash_test_cases_info.add(self.current_test_case_info) else: self.result_string = "%s [no temp dir]" % error_type_str self.send_result_through_pipe(test, error_type) def addFailure(self, test, err): """ Record a test failed result :param test: :param err: error message """ self.add_error(test, err, unittest.TestResult.addFailure, FAIL) def addError(self, test, err): """ Record a test error result :param test: :param err: error message """ self.add_error(test, err, unittest.TestResult.addError, ERROR) def getDescription(self, test): """ Get test description :param test: :returns: test description """ return get_test_description(self.descriptions, test) def startTest(self, test): """ Start a test :param test: """ def print_header(test): if test.__class__ in self.printed: return test_doc = getdoc(test) if not test_doc: raise Exception("No doc string for test '%s'" % test.id()) test_title = test_doc.splitlines()[0].rstrip() test_title = colorize(test_title, GREEN) if test.is_tagged_run_solo(): test_title = colorize(f"SOLO RUN: {test_title}", YELLOW) # This block may overwrite the colorized title above, # but we want this to stand out and be fixed if test.has_tag(TestCaseTag.FIXME_VPP_WORKERS): test_title = colorize(f"FIXME with VPP workers: {test_title}", RED) if test.has_tag(TestCaseTag.FIXME_ASAN): test_title = colorize(f"FIXME with ASAN: {test_title}", RED) test.skip_fixme_asan() if is_distro_ubuntu2204 == True and test.has_tag( TestCaseTag.FIXME_UBUNTU2204 ): test_title = colorize(f"FIXME on Ubuntu-22.04: {test_title}", RED) test.skip_fixme_ubuntu2204() if is_distro_debian11 == True and test.has_tag(TestCaseTag.FIXME_DEBIAN11): test_title = colorize(f"FIXME on Debian-11: {test_title}", RED) test.skip_fixme_debian11() if "debug" in config.vpp_tag and test.has_tag(TestCaseTag.FIXME_VPP_DEBUG): test_title = colorize(f"FIXME on VPP Debug: {test_title}", RED) test.skip_fixme_vpp_debug() if hasattr(test, "vpp_worker_count"): if test.vpp_worker_count == 0: test_title += " [main thread only]" elif test.vpp_worker_count == 1: test_title += " [1 worker thread]" else: test_title += f" [{test.vpp_worker_count} worker threads]" if test.__class__.skipped_due_to_cpu_lack: test_title = colorize( f"{test_title} [skipped - not enough cpus, " f"required={test.__class__.get_cpus_required()}, " f"available={max_vpp_cpus}]", YELLOW, ) print(double_line_delim) print(test_title) print(double_line_delim) self.printed.append(test.__class__) print_header(test) self.start_test = time.time() unittest.TestResult.startTest(self, test) if self.verbosity > 0: self.stream.writeln("Starting " + self.getDescription(test) + " ...") self.stream.writeln(single_line_delim) def stopTest(self, test): """ Called when the given test has been run :param test: """ unittest.TestResult.stopTest(self, test) if self.verbosity > 0: self.stream.writeln(single_line_delim) self.stream.writeln( "%-73s%s" % (self.getDescription(test), self.result_string) ) self.stream.writeln(single_line_delim) else: self.stream.writeln( "%-68s %4.2f %s" % ( self.getDescription(test), time.time() - self.start_test, self.result_string, ) ) self.send_result_through_pipe(test, TEST_RUN) def printErrors(self): """ Print errors from running the test case """ if len(self.errors) > 0 or len(self.failures) > 0: self.stream.writeln() self.printErrorList("ERROR", self.errors) self.printErrorList("FAIL", self.failures) # ^^ that is the last output from unittest before summary if not self.runner.print_summary: devnull = unittest.runner._WritelnDecorator(open(os.devnull, "w")) self.stream = devnull self.runner.stream = devnull def printErrorList(self, flavour, errors): """ Print error list to the output stream together with error type and test case description. :param flavour: error type :param errors: iterable errors """ for test, err in errors: self.stream.writeln(double_line_delim) self.stream.writeln("%s: %s" % (flavour, self.getDescription(test))) self.stream.writeln(single_line_delim) self.stream.writeln("%s" % err) class VppTestRunner(unittest.TextTestRunner): """ A basic test runner implementation which prints results to standard error. """ @property def resultclass(self): """Class maintaining the results of the tests""" return VppTestResult def __init__( self, keep_alive_pipe=None, descriptions=True, verbosity=1, result_pipe=None, failfast=False, buffer=False, resultclass=None, print_summary=True, **kwargs, ): # ignore stream setting here, use hard-coded stdout to be in sync # with prints from VppTestCase methods ... super(VppTestRunner, self).__init__( sys.stdout, descriptions, verbosity, failfast, buffer, resultclass, **kwargs ) KeepAliveReporter.pipe = keep_alive_pipe self.orig_stream = self.stream self.resultclass.test_framework_result_pipe = result_pipe self.print_summary = print_summary def _makeResult(self): return self.resultclass(self.stream, self.descriptions, self.verbosity, self) def run(self, test): """ Run the tests :param test: """ faulthandler.enable() # emit stack trace to stderr if killed by signal result = super(VppTestRunner, self).run(test) if not self.print_summary: self.stream = self.orig_stream result.stream = self.orig_stream return result class Worker(Thread): def __init__(self, executable_args, logger, env=None, *args, **kwargs): super(Worker, self).__init__(*args, **kwargs) self.logger = logger self.args = executable_args if hasattr(self, "testcase") and self.testcase.debug_all: if self.testcase.debug_gdbserver: self.args = [ "/usr/bin/gdbserver", "localhost:{port}".format(port=self.testcase.gdbserver_port), ] + args elif self.testcase.debug_gdb and hasattr(self, "wait_for_gdb"): self.args.append(self.wait_for_gdb) self.app_bin = executable_args[0] self.app_name = os.path.basename(self.app_bin) if hasattr(self, "role"): self.app_name += " {role}".format(role=self.role) self.process = None self.result = None env = {} if env is None else env self.env = copy.deepcopy(env) def wait_for_enter(self): if not hasattr(self, "testcase"): return if self.testcase.debug_all and self.testcase.debug_gdbserver: print() print(double_line_delim) print( "Spawned GDB Server for '{app}' with PID: {pid}".format( app=self.app_name, pid=self.process.pid ) ) elif self.testcase.debug_all and self.testcase.debug_gdb: print() print(double_line_delim) print( "Spawned '{app}' with PID: {pid}".format( app=self.app_name, pid=self.process.pid ) ) else: return print(single_line_delim) print("You can debug '{app}' using:".format(app=self.app_name)) if self.testcase.debug_gdbserver: print( "sudo gdb " + self.app_bin + " -ex 'target remote localhost:{port}'".format( port=self.testcase.gdbserver_port ) ) print( "Now is the time to attach gdb by running the above " "command, set up breakpoints etc., then resume from " "within gdb by issuing the 'continue' command" ) self.testcase.gdbserver_port += 1 elif self.testcase.debug_gdb: print( "sudo gdb " + self.app_bin + " -ex 'attach {pid}'".format(pid=self.process.pid) ) print( "Now is the time to attach gdb by running the above " "command and set up breakpoints etc., then resume from" " within gdb by issuing the 'continue' command" ) print(single_line_delim) input("Press ENTER to continue running the testcase...") def run(self): executable = self.args[0] if not os.path.exists(executable) or not os.access( executable, os.F_OK | os.X_OK ): # Exit code that means some system file did not exist, # could not be opened, or had some other kind of error. self.result = os.EX_OSFILE raise EnvironmentError( "executable '%s' is not found or executable." % executable ) self.logger.debug( "Running executable '{app}': '{cmd}'".format( app=self.app_name, cmd=" ".join(self.args) ) ) env = os.environ.copy() env.update(self.env) env["CK_LOG_FILE_NAME"] = "-" self.process = subprocess.Popen( ["stdbuf", "-o0", "-e0"] + self.args, shell=False, env=env, preexec_fn=os.setpgrp, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) self.wait_for_enter() out, err = self.process.communicate() self.logger.debug("Finished running `{app}'".format(app=self.app_name)) self.logger.info("Return code is `%s'" % self.process.returncode) self.logger.info(single_line_delim) self.logger.info( "Executable `{app}' wrote to stdout:".format(app=self.app_name) ) self.logger.info(single_line_delim) self.logger.info(out.decode("utf-8")) self.logger.info(single_line_delim) self.logger.info( "Executable `{app}' wrote to stderr:".format(app=self.app_name) ) self.logger.info(single_line_delim) self.logger.info(err.decode("utf-8")) self.logger.info(single_line_delim) self.result = self.process.returncode if __name__ == "__main__": pass