#!/usr/bin/env python from __future__ import print_function import gc import sys import os import select import signal 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 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 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 if os.name == 'posix' and sys.version_info[0] < 3: # using subprocess32 is recommended by python official documentation # @ https://docs.python.org/2/library/subprocess.html import subprocess32 as subprocess else: import subprocess # Python2/3 compatible try: input = raw_input except NameError: pass PASS = 0 FAIL = 1 ERROR = 2 SKIP = 3 TEST_RUN = 4 debug_framework = False if os.getenv('TEST_DEBUG', "0") == "1": debug_framework = True 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: 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.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.debug( "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.splitlines(True) if len(stderr_fragment) > 0: split[0] = "%s%s" % (stderr_fragment, split[0]) if len(split) > 0 and split[-1].endswith(b"\n"): limit = N
# Minimal clang-format version is 8
BasedOnStyle: GNU
UseTab: Always
SpaceAfterCStyleCast: true
SortIncludes: false
AlignConsecutiveMacros: true
BreakBeforeTernaryOperators: false
BreakBeforeBinaryOperators: None
ContinuationIndentWidth: 2
ForEachMacros:
- 'clib_bitmap_foreach'
- 'pool_foreach'
- 'pool_foreach_index'
- 'vec_foreach'
- 'vec_foreach_backwards'
- 'vec_foreach_index'
- 'vec_foreach_index_backwards'
- 'vlib_foreach_rx_tx'
- 'foreach_vlib_main'
StatementMacros:
- 'CLIB_MULTIARCH_FN'
- 'VLIB_NODE_FN'
- 'VNET_DEVICE_CLASS_TX_FN'
- '__clib_section'
- '__clib_aligned'