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.pg_streams = []
cls.packet_infos = {}
cls.verbose = 0
+ cls.vpp_dead = False
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
try:
cls.run_vpp()
- cls.vpp_dead = False
+ cls.vpp_stdout_queue = Queue()
+ cls.vpp_stdout_reader_thread = Thread(target=pump_output, args=(
+ cls.vpp.stdout, cls.vpp_stdout_queue))
+ cls.vpp_stdout_reader_thread.start()
+ cls.vpp_stderr_queue = Queue()
+ cls.vpp_stderr_reader_thread = Thread(target=pump_output, args=(
+ cls.vpp.stderr, cls.vpp_stderr_queue))
+ cls.vpp_stderr_reader_thread.start()
cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix)
if cls.step:
- cls.vapi.register_hook(StepHook(cls))
+ hook = StepHook(cls)
else:
- cls.vapi.register_hook(PollHook(cls))
+ hook = PollHook(cls)
+ cls.vapi.register_hook(hook)
time.sleep(0.1)
+ hook.poll_vpp()
try:
cls.vapi.connect()
except:
"VPP-API connection failed, did you forget "
"to 'continue' VPP from within gdb?", RED))
raise
- cls.vpp_stdout_queue = Queue()
- cls.vpp_stdout_reader_thread = Thread(
- target=pump_output, args=(cls.vpp.stdout, cls.vpp_stdout_queue))
- cls.vpp_stdout_reader_thread.start()
- cls.vpp_stderr_queue = Queue()
- cls.vpp_stderr_reader_thread = Thread(
- target=pump_output, args=(cls.vpp.stderr, cls.vpp_stderr_queue))
- cls.vpp_stderr_reader_thread.start()
except:
- if hasattr(cls, 'vpp'):
- cls.vpp.terminate()
- del cls.vpp
- raise
+ t, v, tb = sys.exc_info()
+ try:
+ cls.quit()
+ except:
+ pass
+ raise t, v, tb
@classmethod
def quit(cls):
" and finish running the testcase...")
if hasattr(cls, 'vpp'):
- cls.vapi.disconnect()
+ if hasattr(cls, 'vapi'):
+ cls.vapi.disconnect()
cls.vpp.poll()
if cls.vpp.returncode is None:
cls.vpp.terminate()
def tearDown(self):
""" Show various debug prints after each test """
if not self.vpp_dead:
- self.logger.info(self.vapi.ppcli("show int"))
self.logger.debug(self.vapi.cli("show trace"))
+ self.logger.info(self.vapi.ppcli("show int"))
self.logger.info(self.vapi.ppcli("show hardware"))
self.logger.info(self.vapi.ppcli("show error"))
self.logger.info(self.vapi.ppcli("show run"))