#!/usr/bin/env python3 from __future__ import print_function import gc import logging import sys import os import select import signal import subprocess import unittest import tempfile import time import faulthandler import random import copy import psutil import platform 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 import scapy.compat from scapy.packet import Raw 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 import vpp_papi from vpp_papi.vpp_stats import VPPStats from vpp_papi.vpp_transport_shmem import VppTransportShmemIOError 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 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 class BoolEnvironmentVariable(object): def __init__(self, env_var_name, default='n', true_values=None): self.name = env_var_name self.default = default self.true_values = true_values if true_values is not None else \ ("y", "yes", "1") def __bool__(self): return os.getenv(self.name, self.default).lower() in self.true_values if sys.version_info[0] == 2: __nonzero__ = __bool__ def __repr__(self): return 'BoolEnvironmentVariable(%r, default=%r, true_values=%r)' % \ (self.name, self.default, self.true_values) debug_framework = BoolEnvironmentVariable('TEST_DEBUG') if 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 = 'running %s.%s ' % (testcase, method_name) msg = "VPP subprocess died %sunexpectedly with return code: %d%s." % ( in_msg, self.rv, ' [%s]' % (self.signal_name if self.signal_name is not None else '')) 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 """ 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 testclass.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 testclass.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_skip_aarch64_set(): return BoolEnvironmentVariable('SKIP_AARCH64') is_skip_aarch64_set = _is_skip_aarch64_set() def _is_platform_aarch64(): return platform.machine() == 'aarch64' is_platform_aarch64 = _is_platform_aarch64() def _running_extended_tests(): return BoolEnvironmentVariable("EXTENDED_TESTS") running_extended_tests = _running_extended_tests() def _running_gcov_tests(): return BoolEnvironmentVariable("GCOV_TESTS") running_gcov_tests = _running_gcov_tests() 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 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, test.vpp_bin, 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 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) class VppTestCase(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_punt_config = [] extra_vpp_plugin_config = [] logger = null_logger vapi_response_timeout = 5 @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 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 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 else: raise Exception("Unrecognized DEBUG option: '%s'" % d) if dl == "gdb-all" or dl == "gdbserver-all": cls.debug_all = True @staticmethod def get_least_used_cpu(): cpu_usage_list = [set(range(psutil.cpu_count()))] vpp_processes = [p for p in psutil.process_iter(attrs=['pid', 'name']) if 'vpp_main' == p.info['name']] for vpp_process in vpp_processes: for cpu_usage_set in cpu_usage_list: try: cpu_num = vpp_process.cpu_num() if cpu_num in cpu_usage_set: cpu_usage_set_index = cpu_usage_list.index( cpu_usage_set) if cpu_usage_set_index == len(cpu_usage_list) - 1: cpu_usage_list.append({cpu_num}) else: cpu_usage_list[cpu_usage_set_index + 1].add( cpu_num) cpu_usage_set.remove(cpu_num) break except psutil.NoSuchProcess: pass for cpu_usage_set in cpu_usage_list: if len(cpu_usage_set) > 0: min_usage_set = cpu_usage_set break return random.choice(tuple(min_usage_set)) @classmethod def setUpConstants(cls): """ Set-up the test case class based on environment variables """ cls.step = BoolEnvironmentVariable('STEP') d = os.getenv("DEBUG", None) # inverted case to handle '' == True c = os.getenv("CACHE_OUTPUT", "1") cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True cls.set_debug_flags(d) cls.vpp_bin = os.getenv('VPP_BIN', "vpp") cls.plugin_path = os.getenv('VPP_PLUGIN_PATH') cls.test_plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH') cls.extern_plugin_path = os.getenv('EXTERN_PLUGINS') plugin_path = None if cls.plugin_path is not None: if cls.extern_plugin_path is not None: plugin_path = "%s:%s" % ( cls.plugin_path, cls.extern_plugin_path) else: plugin_path = cls.plugin_path elif cls.extern_plugin_path is not None: plugin_path = cls.extern_plugin_path debug_cli = "" if cls.step or cls.debug_gdb or cls.debug_gdbserver: debug_cli = "cli-listen localhost:5002" coredump_size = None size = os.getenv("COREDUMP_SIZE") if size is not None: coredump_size = "coredump-size %s" % size if coredump_size is None: coredump_size = "coredump-size unlimited" cpu_core_number = cls.get_least_used_cpu() if not hasattr(cls, "worker_config"): cls.worker_config = os.getenv("VPP_WORKER_CONFIG", "") if cls.worker_config != "": if cls.has_tag(TestCaseTag.FIXME_VPP_WORKERS): cls.worker_config = "" default_variant = os.getenv("VARIANT") if default_variant is not None: default_variant = "defaults { %s 100 }" % default_variant else: default_variant = "" api_fuzzing = os.getenv("API_FUZZ") if api_fuzzing is None: api_fuzzing = 'off' cls.vpp_cmdline = [cls.vpp_bin, "unix", "{", "nodaemon", debug_cli, "full-coredump", coredump_size, "runtime-dir", cls.tempdir, "}", "api-trace", "{", "on", "}", "api-segment", "{", "prefix", cls.shm_prefix, "}", "cpu", "{", "main-core", str(cpu_core_number), cls.worker_config, "}", "physmem", "{", "max-size", "32m", "}", "statseg", "{", "socket-name", cls.stats_sock, "}", "socksvr", "{", "socket-name", cls.api_sock, "}", "node { ", default_variant, "}", "api-fuzz {", api_fuzzing, "}", "plugins", "{", "plugin", "dpdk_plugin.so", "{", "disable", "}", "plugin", "rdma_plugin.so", "{", "disable", "}", "plugin", "lisp_unittest_plugin.so", "{", "enable", "}", "plugin", "unittest_plugin.so", "{", "enable", "}"] + cls.extra_vpp_plugin_config + ["}", ] if cls.extra_vpp_punt_config is not None: cls.vpp_cmdline.extend(cls.extra_vpp_punt_config) if plugin_path is not None: cls.vpp_cmdline.extend(["plugin_path", plugin_path]) if cls.test_plugin_path is not None: 
/*
 * 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.
 */

#ifndef VNET_CONTROL_H_
#define VNET_CONTROL_H_

#include <vnet/vnet.h>
#include <vnet/lisp-cp/gid_dictionary.h>
#include <vnet/lisp-cp/lisp_types.h>

#define NUMBER_OF_RETRIES                   1
#define PENDING_MREQ_EXPIRATION_TIME        3.0	/* seconds */
#define PENDING_MREQ_QUEUE_LEN              5

#define PENDING_MREG_EXPIRATION_TIME        3.0	/* seconds */
#define RLOC_PROBING_INTERVAL               60.0

/* when map-registration is enabled "quick registration" takes place first.
   In this mode ETR sends map-register messages at an increased frequency
   until specified message count is reached */
#define QUICK_MAP_REGISTER_MSG_COUNT        3
#define QUICK_MAP_REGISTER_INTERVAL         3.0

/* normal map-register period */
#define MAP_REGISTER_INTERVAL               60.0

/* 15 minutes */
#define MAP_REGISTER_DEFAULT_TTL            900

typedef struct
{
  gid_address_t src;
  gid_address_t dst;
  u32 retries_num;
  f64 time_to_expire;
  u8 is_smr_invoked;
  u64 *nonces;
  u8 to_be_removed;
} pending_map_request_t;

typedef struct
{
  gid_address_t leid;
  gid_address_t reid;
  u8 is_src_dst;
  locator_pair_t *locator_pairs;
} fwd_entry_t;

typedef struct
{
  gid_address_t leid;
  gid_address_t reid;
} lisp_adjacency_t;

typedef enum
{
  IP4_MISS_PACKET,
  IP6_MISS_PACKET
} miss_packet_type_t;

/* map-server/map-resolver structure */
typedef struct
{
  u8 is_down;
  f64 last_update;
  ip_address_t address;
  char *key;
} lisp_msmr_t;

typedef struct
{
  /* headers */
  u8 data[100];
  u32 length;
  miss_packet_type_t type;
} miss_packet_t;

typedef enum
{
  MR_MODE_DST_ONLY = 0,
  MR_MODE_SRC_DST,
  _MR_MODE_MAX
} map_request_mode_t;

typedef struct
{
  /* LISP feature status */
  u8 is_enabled;

  /* eid table */
  gid_dictionary_t mapping_index_by_gid;

  /* pool of mappings */
  mapping_t *mapping_pool;

  /* hash map of secret keys by mapping index */
  u8 *key_by_mapping_index;

  /* pool of locators */
  locator_t *locator_pool;

  /* pool of locator-sets */
  locator_set_t *locator_set_pool;

  /* vector of locator-set vectors composed of and indexed by locator index */
  u32 **locator_to_locator_sets;

  /* hash map of locators by name */
  uword *locator_set_index_by_name;

  /* vector of eid index vectors supported and indexed by locator-set index */
  u32 **locator_set_to_eids;

  /* vectors of indexes for local locator-sets and mappings */
  u32 *local_mappings_indexes;
  u32 *local_locator_set_indexes;

  /* hash map of forwarding entries by mapping index */
  u32 *fwd_entry_by_mapping_index;

  /* forwarding entries pool */
  fwd_entry_t *fwd_entry_pool;

  /* hash map keyed by nonce of pending map-requests */
  uword *pending_map_requests_by_nonce;

  /* pool of pending map requests */
  pending_map_request_t *pending_map_requests_pool;

  /* hash map of sent map register messages */
  uword *map_register_messages_by_nonce;

  /* vector of map-resolvers */
  lisp_msmr_t *map_resolvers;

  /* vector of map-servers */
  lisp_msmr_t *map_servers;

  /* map resolver address currently being used for sending requests.
   * This has to be an actual address and not an index to map_resolvers vector
   * since the vector may be modified during request resend/retry procedure
   * and break things :-) */
  ip_address_t active_map_resolver;

  u8 do_map_resolver_election;

  /* map-request  locator set index */
  u32 mreq_itr_rlocs;

  /* vni to vrf hash tables */
  uword *table_id_by_vni;
  uword *vni_by_table_id;

  /* vni to bd-index hash tables */
  uword *bd_id_by_vni;
  uword *vni_by_bd_id;

  /* track l2 and l3 interfaces that have been created for vni */
  uword *l2_dp_intf_by_vni;

  /* Proxy ETR map index */
  u32 pitr_map_index;

  /* LISP PITR mode */
  u8 lisp_pitr;

  /* map request mode */
  u8 map_request_mode;

  /* enable/disable map registering */
  u8 map_registering;

  /* enable/disable rloc-probing */
  u8 rloc_probing;

  /* timing wheel for mappping timeouts */
  timing_wheel_t wheel;

  /* commodity */
  ip4_main_t *im4;
  ip6_main_t *im6;
  vlib_main_t *vlib_main;
  vnet_main_t *vnet_main;
} lisp_cp_main_t;

/* lisp-gpe control plane */
lisp_cp_main_t lisp_control_main;

extern vlib_node_registration_t lisp_cp_input_node;
extern vlib_node_registration_t lisp_cp_lookup_ip4_node;
extern vlib_node_registration_t lisp_cp_lookup_ip6_node;

clib_error_t *lisp_cp_init ();

always_inline lisp_cp_main_t *
vnet_lisp_cp_get_main ()
{
  return &lisp_control_main;
}

typedef struct
{
  u8 is_add;
  union
  {
    u8 *name;
    u32 index;
  };
  locator_t *locators;
  u8 local;
} vnet_lisp_add_del_locator_set_args_t;

int
vnet_lisp_add_del_locator_set (vnet_lisp_add_del_locator_set_args_t * a,
			       u32 * ls_index);
int
vnet_lisp_add_del_locator (vnet_lisp_add_del_locator_set_args_t * a,
			   locator_set_t * ls, u32 * ls_index);

typedef struct
{
  u8 is_add;
  gid_address_t eid;
  u32 locator_set_index;

  u32 ttl;
  u8 action;
  u8 authoritative;

  u8 local;
  u8 is_static;
  u8 *key;
  u8 key_id;
} vnet_lisp_add_del_mapping_args_t;

int
vnet_lisp_map_cache_add_del (vnet_lisp_add_del_mapping_args_t * a,
			     u32 * map_index);
int
vnet_lisp_add_del_local_mapping (vnet_lisp_add_del_mapping_args_t * a,
				 u32 * map_index_result);

int
vnet_lisp_add_del_mapping (gid_address_t * deid, locator_t * dlocs, u8 action,
			   u8 authoritative, u32 ttl, u8 is_add, u8 is_static,
			   u32 * res_map_index);

typedef struct
{
  gid_address_t reid;
  gid_address_t leid;
  u8 is_add;
} vnet_lisp_add_del_adjacency_args_t;

int vnet_lisp_add_del_adjacency (vnet_lisp_add_del_adjacency_args_t * a);

typedef struct
{
  u8 is_add;
  ip_address_t address;
} vnet_lisp_add_del_map_resolver_args_t;

int
vnet_lisp_add_del_map_resolver (vnet_lisp_add_del_map_resolver_args_t * a);
int vnet_lisp_add_del_map_server (ip_address_t * addr, u8 is_add);

clib_error_t *vnet_lisp_enable_disable (u8 is_enabled);
u8 vnet_lisp_enable_disable_status (void);

int vnet_lisp_pitr_set_locator_set (u8 * locator_set_name, u8 is_add);

typedef struct
{
  u8 is_add;
  u8 *locator_set_name;
} vnet_lisp_add_del_mreq_itr_rloc_args_t;

int
vnet_lisp_add_del_mreq_itr_rlocs (vnet_lisp_add_del_mreq_itr_rloc_args_t * a);

int vnet_lisp_clear_all_remote_adjacencies (void);

int vnet_lisp_eid_table_map (u32 vni, u32 vrf, u8 is_l2, u8 is_add);
int vnet_lisp_add_del_map_table_key (gid_address_t * eid, char *key,
				     u8 is_add);
int vnet_lisp_set_map_request_mode (u8 mode);
u8 vnet_lisp_get_map_request_mode (void);
lisp_adjacency_t *vnet_lisp_adjacencies_get_by_vni (u32 vni);
int vnet_lisp_rloc_probe_enable_disable (u8 is_enable);
int vnet_lisp_map_register_enable_disable (u8 is_enable);
u8 vnet_lisp_map_register_state_get (void);
u8 vnet_lisp_rloc_probe_state_get (void);

#endif /* VNET_CONTROL_H_ */

/*
 * fd.io coding-style-patch-verification: ON
 *
 * Local Variables:
 * eval: (c-set-style "gnu")
 * End:
 */
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 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) self.send_result_through_pipe(test, SKIP) def symlink_failed(self): if self.current_test_case_info: try: failed_dir = os.getenv('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): test_doc = getdoc(test) if not test_doc: raise Exception("No doc string for test '%s'" % test.id()) test_title = test_doc.splitlines()[0] test_title_colored = colorize(test_title, GREEN) if test.is_tagged_run_solo(): # long live PEP-8 and 80 char width limitation... c = YELLOW test_title_colored = colorize("SOLO RUN: " + test_title, c) # 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): c = RED w = "FIXME with VPP workers: " test_title_colored = colorize(w + test_title, c) if not hasattr(test.__class__, '_header_printed'): print(double_line_delim) print(test_title_colored) print(double_line_delim) test.__class__._header_printed = True 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}'" .format(app=' '.join(self.args))) env = os.environ.copy() env.update(self.env) env["CK_LOG_FILE_NAME"] = "-" self.process = subprocess.Popen( 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