
Introduce an API which asserts empty capture for interface. Throw exception in old API if the capture does not exist, thus making it clear if the test expects packets to arrive or not. Improve performance by not doing sleeps after starting the packet generator, rather lazily deleting captures when needed. Fix wrong usage of packet.show() in various tests. Change-Id: I456cb23316eef99b3f35f80344fe595c4db9a21c Signed-off-by: Klement Sekera <ksekera@cisco.com>
88 lines
1.9 KiB
Python
88 lines
1.9 KiB
Python
import socket
|
|
import sys
|
|
from abc import abstractmethod, ABCMeta
|
|
from cStringIO import StringIO
|
|
|
|
|
|
def ppp(headline, packet):
|
|
""" Return string containing the output of scapy packet.show() call. """
|
|
o = StringIO()
|
|
old_stdout = sys.stdout
|
|
sys.stdout = o
|
|
print(headline)
|
|
packet.show()
|
|
sys.stdout = old_stdout
|
|
return o.getvalue()
|
|
|
|
|
|
def ppc(headline, capture, limit=10):
|
|
""" Return string containing ppp() printout for a capture.
|
|
|
|
:param headline: printed as first line of output
|
|
:param capture: packets to print
|
|
:param limit: limit the print to # of packets
|
|
"""
|
|
if not capture:
|
|
return headline
|
|
result = headline + "\n"
|
|
count = 1
|
|
for p in capture:
|
|
result.append(ppp("Packet #%s:" % count, p))
|
|
count += 1
|
|
if count >= limit:
|
|
break
|
|
if limit < len(capture):
|
|
result.append(
|
|
"Capture contains %s packets in total, of which %s were printed" %
|
|
(len(capture), limit))
|
|
|
|
|
|
class NumericConstant(object):
|
|
__metaclass__ = ABCMeta
|
|
|
|
desc_dict = {}
|
|
|
|
@abstractmethod
|
|
def __init__(self, value):
|
|
self._value = value
|
|
|
|
def __int__(self):
|
|
return self._value
|
|
|
|
def __long__(self):
|
|
return self._value
|
|
|
|
def __str__(self):
|
|
if self._value in self.desc_dict:
|
|
return self.desc_dict[self._value]
|
|
return ""
|
|
|
|
|
|
class Host(object):
|
|
""" Generic test host "connected" to VPPs interface. """
|
|
|
|
@property
|
|
def mac(self):
|
|
""" MAC address """
|
|
return self._mac
|
|
|
|
@property
|
|
def ip4(self):
|
|
""" IPv4 address """
|
|
return self._ip4
|
|
|
|
@property
|
|
def ip4n(self):
|
|
""" IPv4 address """
|
|
return socket.inet_pton(socket.AF_INET, self._ip4)
|
|
|
|
@property
|
|
def ip6(self):
|
|
""" IPv6 address """
|
|
return self._ip6
|
|
|
|
def __init__(self, mac=None, ip4=None, ip6=None):
|
|
self._mac = mac
|
|
self._ip4 = ip4
|
|
self._ip6 = ip6
|