X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Futil.py;h=d6aa9a426771314905c077de31b67d3a0da7fb73;hb=3cdc25ffbaa572639f99e197172c568e4324bc03;hp=643377f5740c8916b986554f80feebf7bde65cd0;hpb=7bb873a4cc068a6cc3c9d0e1d32987c5f8003904;p=vpp.git diff --git a/test/util.py b/test/util.py index 643377f5740..d6aa9a42677 100644 --- a/test/util.py +++ b/test/util.py @@ -1,6 +1,10 @@ +""" test framework utilities """ + import socket import sys +from abc import abstractmethod, ABCMeta from cStringIO import StringIO +from scapy.layers.inet6 import in6_mactoifaceid def ppp(headline, packet): @@ -14,6 +18,68 @@ def ppp(headline, packet): 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 + tail = "" + if limit < len(capture): + tail = "\nPrint limit reached, %s out of %s packets printed" % ( + len(capture), limit) + limit = len(capture) + body = "".join([ppp("Packet #%s:" % count, p) + for count, p in zip(range(0, limit), capture)]) + return "%s\n%s%s" % (headline, body, tail) + + +def ip4_range(ip4, s, e): + tmp = ip4.rsplit('.', 1)[0] + return ("%s.%d" % (tmp, i) for i in range(s, e)) + + +def ip4n_range(ip4n, s, e): + ip4 = socket.inet_ntop(socket.AF_INET, ip4n) + return (socket.inet_pton(socket.AF_INET, ip) + for ip in ip4_range(ip4, s, e)) + + +def mactobinary(mac): + """ Convert the : separated format into binary packet data for the API """ + return mac.replace(':', '').decode('hex') + + +def mk_ll_addr(mac): + euid = in6_mactoifaceid(mac) + addr = "fe80::" + euid + return addr + + +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. """ @@ -22,22 +88,78 @@ class Host(object): """ MAC address """ return self._mac + @property + def bin_mac(self): + """ MAC address """ + return mactobinary(self._mac) + @property def ip4(self): - """ IPv4 address """ + """ IPv4 address - string """ return self._ip4 @property def ip4n(self): - """ IPv4 address """ + """ IPv4 address of remote host - raw, suitable as API parameter.""" return socket.inet_pton(socket.AF_INET, self._ip4) @property def ip6(self): - """ IPv6 address """ + """ IPv6 address - string """ return self._ip6 - def __init__(self, mac=None, ip4=None, ip6=None): + @property + def ip6n(self): + """ IPv6 address of remote host - raw, suitable as API parameter.""" + return socket.inet_pton(socket.AF_INET6, self._ip6) + + @property + def ip6_ll(self): + """ IPv6 link-local address - string """ + return self._ip6_ll + + @property + def ip6n_ll(self): + """ IPv6 link-local address of remote host - + raw, suitable as API parameter.""" + return socket.inet_pton(socket.AF_INET6, self._ip6_ll) + + def __eq__(self, other): + if isinstance(other, Host): + return (self.mac == other.mac and + self.ip4 == other.ip4 and + self.ip6 == other.ip6 and + self.ip6_ll == other.ip6_ll) + else: + return False + + def __ne__(self, other): + return not self.__eq__(other) + + def __repr__(self): + return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (self.mac, + self.ip4, + self.ip6, + self.ip6_ll) + + def __hash__(self): + return hash(self.__repr__()) + + def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None): self._mac = mac self._ip4 = ip4 self._ip6 = ip6 + self._ip6_ll = ip6_ll + + +class ForeignAddressFactory(object): + count = 0 + prefix_len = 24 + net_template = '10.10.10.{}' + net = net_template.format(0) + '/' + str(prefix_len) + + def get_ip4(self): + if self.count > 255: + raise Exception("Network host address exhaustion") + self.count += 1 + return self.net_template.format(self.count)