#!/usr/bin/env python
-from abc import *
-import os
import subprocess
import unittest
import tempfile
from inspect import getdoc
from hook import StepHook, PollHook
from vpp_pg_interface import VppPGInterface
+from vpp_lo_interface import VppLoInterface
from vpp_papi_provider import VppPapiProvider
from scapy.packet import Raw
from log import *
Help process information about the next packet.
Set variables to default values.
- @property index
- Integer variable to store the index of the packet.
- @property src
- Integer variable to store the index of the source packet generator
- interface of the packet.
- @property dst
- Integer variable to store the index of the destination packet generator
- interface of the packet.
- @property data
- Object variable to store the copy of the former packet.
-
-
"""
+ #: 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 the copy of the former packet.
data = None
class VppTestCase(unittest.TestCase):
- """
- Subclass of the python unittest.TestCase class.
-
- This subclass is a base class for test cases that are implemented as classes
- It provides methods to create and run test case.
-
+ """This subclass is a base class for VPP test cases that are implemented as
+ classes. It provides methods to create and run test case.
"""
@property
cls.set_debug_flags(d)
cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
- cls.vpp_cmdline = [cls.vpp_bin, "unix", "nodaemon",
+ debug_cli = ""
+ if cls.step or cls.debug_gdb or cls.debug_gdbserver:
+ debug_cli = "cli-listen localhost:5002"
+ cls.vpp_cmdline = [cls.vpp_bin, "unix", "{", "nodaemon", debug_cli, "}",
"api-segment", "{", "prefix", cls.shm_prefix, "}"]
if cls.plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
cmdline = cls.vpp_cmdline
if cls.debug_gdbserver:
- cmdline = ['gdbserver', 'localhost:7777'] + cls.vpp_cmdline
+ gdbserver = '/usr/bin/gdbserver'
+ if not os.path.isfile(gdbserver) or \
+ not os.access(gdbserver, os.X_OK):
+ raise Exception("gdbserver binary '%s' does not exist or is "
+ "not executable" % gdbserver)
+
+ cmdline = [gdbserver, 'localhost:7777'] + cls.vpp_cmdline
cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
- cls.vpp = subprocess.Popen(cmdline,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE,
- bufsize=1)
+ try:
+ cls.vpp = subprocess.Popen(cmdline,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ bufsize=1)
+ except Exception as e:
+ cls.logger.critical("Couldn't start vpp: %s" % e)
+ raise
+
cls.wait_for_enter()
@classmethod
cls.packet_infos = {}
cls.verbose = 0
print(double_line_delim)
- print(colorize(getdoc(cls), YELLOW))
+ print(colorize(getdoc(cls).splitlines()[0], YELLOW))
print(double_line_delim)
# need to catch exceptions here because if we raise, then the cleanup
# doesn't get called and we might end with a zombie vpp
target=pump_output, args=(cls.vpp.stderr, cls.vpp_stderr_queue))
cls.vpp_stderr_reader_thread.start()
except:
- cls.vpp.terminate()
- del cls.vpp
+ if hasattr(cls, 'vpp'):
+ cls.vpp.terminate()
+ del cls.vpp
raise
@classmethod
def tearDown(self):
""" Show various debug prints after each test """
if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show int"))
- self.logger.info(self.vapi.cli("show trace"))
- self.logger.info(self.vapi.cli("show hardware"))
- self.logger.info(self.vapi.cli("show error"))
- self.logger.info(self.vapi.cli("show run"))
+ self.logger.info(self.vapi.ppcli("show int"))
+ self.logger.debug(self.vapi.cli("show trace"))
+ self.logger.info(self.vapi.ppcli("show hardware"))
+ self.logger.info(self.vapi.ppcli("show error"))
+ self.logger.info(self.vapi.ppcli("show run"))
def setUp(self):
""" Clear trace before running each test"""
cls.pg_interfaces = result
return result
+ @classmethod
+ def create_loopback_interfaces(cls, interfaces):
+ """
+ Create loopback interfaces
+
+ :param interfaces: iterable indexes of the interfaces
+
+ """
+ result = []
+ for i in interfaces:
+ intf = VppLoInterface(cls, i)
+ setattr(cls, intf.name, intf)
+ result.append(intf)
+ cls.lo_interfaces = result
+ return result
+
@staticmethod
def extend_packet(packet, size):
"""