#!/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 from abc import ABC, abstractmethod from struct import pack, unpack import scapy.compat from scapy.packet import Raw, Packet 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 cpu_config import available_cpus, num_cpus, max_vpp_cpus 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 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 = ' 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 """ 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 spl
from vpp_interface import VppInterface
class VppVhostInterface(VppInterface):
"""VPP vhost interface."""
def __init__(self, test, sock_filename, is_server=0, renumber=0,
disable_mrg_rxbuf=0, disable_indirect_desc=0, enable_gso=0,
enable_packed_ring=0, enable_event_idx=0,
custom_dev_instance=0xFFFFFFFF, use_custom_mac=0,
mac_address='', tag=''):
""" Create VPP Vhost interface """
super(VppVhostInterface, self).__init__(test)
self.is_server = is_server
self.sock_filename = sock_filename
self.renumber = renumber
self.disable_mrg_rxbuf = disable_mrg_rxbuf
self.disable_indirect_desc = disable_indirect_desc
self.enable_gso = enable_gso
self.enable_packed_ring = enable_packed_ring
self.enable_event_idx = enable_event_idx
self.custom_dev_instance = custom_dev_instance
self.use_custom_mac = use_custom_mac
self.mac_address = mac_address
self.tag = tag
def add_vpp_config(self):
r = self.test.vapi.create_vhost_user_if_v2(self.is_server,
self.sock_filename,
self.renumber,
self.disable_mrg_rxbuf,
self.disable_indirect_desc,
self.enable_gso,
self.enable_packed_ring,
self.enable_event_idx,
self.custom_dev_instance,
self.use_custom_mac,
self.mac_address,
self.tag)
self.set_sw_if_index(r.sw_if_index)
def remove_vpp_config(self):
self.test.vapi.delete_vhost_user_if(self.sw_if_index)
def is_interface_config_in_dump(self, dump):
for i in dump:
if i.sw_if_index == self.sw_if_index:
return True
else:
return False