X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=test%2Futil.py;h=3e0267a3f4db2b509d8eb1216d2588a12448ff4f;hb=refs%2Fchanges%2F72%2F8972%2F1;hp=c72a3965d9ef5d7c3e34c4a0c0b18a8b948d3b25;hpb=f56b77a0764222cc45a9df572df901067a273356;p=vpp.git diff --git a/test/util.py b/test/util.py index c72a3965d9e..3e0267a3f4d 100644 --- a/test/util.py +++ b/test/util.py @@ -1,139 +1,231 @@ -## @package util -# Module with common functions that should be used by the test cases. -# -# The module provides a set of tools for setup the test environment - -from scapy.layers.l2 import Ether, ARP -from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6NDOptSrcLLAddr - - -## Util class -# -# Test cases that want to use methods defined in Util class should -# inherit this class. -# -# class Example(Util, VppTestCase): -# pass -class Util(object): - - ## Class method to send ARP Request for each VPP IPv4 address in - # order to determine VPP interface MAC address to IPv4 bindings. - # - # Resolved MAC address is saved to the VPP_MACS dictionary with interface - # index as a key. ARP Request is sent from MAC in MY_MACS dictionary with - # interface index as a key. - # @param cls The class pointer. - # @param args List variable to store indices of VPP interfaces. - @classmethod - def resolve_arp(cls, args): - for i in args: - ip = cls.VPP_IP4S[i] - cls.log("Sending ARP request for %s on port %u" % (ip, i)) - arp_req = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) / - ARP(op=ARP.who_has, pdst=ip, - psrc=cls.MY_IP4S[i], hwsrc=cls.MY_MACS[i])) - cls.pg_add_stream(i, arp_req) - cls.pg_enable_capture([i]) - - cls.cli(2, "trace add pg-input 1") - cls.pg_start() - arp_reply = cls.pg_get_capture(i)[0] - if arp_reply[ARP].op == ARP.is_at: - cls.log("VPP pg%u MAC address is %s " % (i, arp_reply[ARP].hwsrc)) - cls.VPP_MACS[i] = arp_reply[ARP].hwsrc - else: - cls.log("No ARP received on port %u" % i) - cls.cli(2, "show trace") - ## @var ip - # - ## @var arp_req - # - ## @var arp_reply - # - ## @var VPP_MACS - # - - ## Class method to send ND request for each VPP IPv6 address in - # order to determine VPP MAC address to IPv6 bindings. - # - # Resolved MAC address is saved to the VPP_MACS dictionary with interface - # index as a key. ND Request is sent from MAC in MY_MACS dictionary with - # interface index as a key. - # @param cls The class pointer. - # @param args List variable to store indices of VPP interfaces. - @classmethod - def resolve_icmpv6_nd(cls, args): - for i in args: - ip = cls.VPP_IP6S[i] - cls.log("Sending ICMPv6ND_NS request for %s on port %u" % (ip, i)) - nd_req = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) / - IPv6(src=cls.MY_IP6S[i], dst=ip) / - ICMPv6ND_NS(tgt=ip) / - ICMPv6NDOptSrcLLAddr(lladdr=cls.MY_MACS[i])) - cls.pg_add_stream(i, nd_req) - cls.pg_enable_capture([i]) - - cls.cli(2, "trace add pg-input 1") - cls.pg_start() - nd_reply = cls.pg_get_capture(i)[0] - icmpv6_na = nd_reply['ICMPv6 Neighbor Discovery - Neighbor Advertisement'] - dst_ll_addr = icmpv6_na['ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address'] - cls.VPP_MACS[i] = dst_ll_addr.lladdr - ## @var ip - # - ## @var nd_req - # - ## @var nd_reply - # - ## @var icmpv6_na - # - ## @var dst_ll_addr - # - ## @var VPP_MACS - # - - ## Class method to configure IPv4 addresses on VPP interfaces. - # - # Set dictionary variables MY_IP4S and VPP_IP4S to IPv4 addresses - # calculated using interface VPP interface index as a parameter. - # /24 IPv4 prefix is used, with VPP interface address host part set - # to .1 and MY address set to .2. - # Used IPv4 prefix scheme: 172.16.{VPP-interface-index}.0/24. - # @param cls The class pointer. - # @param args List variable to store indices of VPP interfaces. - @classmethod - def config_ip4(cls, args): - for i in args: - cls.MY_IP4S[i] = "172.16.%u.2" % i - cls.VPP_IP4S[i] = "172.16.%u.1" % i - cls.api("sw_interface_add_del_address pg%u %s/24" % (i, cls.VPP_IP4S[i])) - cls.log("My IPv4 address is %s" % (cls.MY_IP4S[i])) - ## @var MY_IP4S - # Dictionary variable to store host IPv4 addresses connected to packet - # generator interfaces. - ## @var VPP_IP4S - # Dictionary variable to store VPP IPv4 addresses of the packet - # generator interfaces. - - ## Class method to configure IPv6 addresses on VPP interfaces. - # - # Set dictionary variables MY_IP6S and VPP_IP6S to IPv6 addresses - # calculated using interface VPP interface index as a parameter. - # /64 IPv6 prefix is used, with VPP interface address host part set - # to ::1 and MY address set to ::2. - # Used IPv6 prefix scheme: fd10:{VPP-interface-index}::0/64. - # @param cls The class pointer. - # @param args List variable to store indices of VPP interfaces. - @classmethod - def config_ip6(cls, args): - for i in args: - cls.MY_IP6S[i] = "fd10:%u::2" % i - cls.VPP_IP6S[i] = "fd10:%u::1" % i - cls.api("sw_interface_add_del_address pg%u %s/64" % (i, cls.VPP_IP6S[i])) - cls.log("My IPv6 address is %s" % (cls.MY_IP6S[i])) - ## @var MY_IP6S - # Dictionary variable to store host IPv6 addresses connected to packet - # generator interfaces. - ## @var VPP_IP6S - # Dictionary variable to store VPP IPv6 addresses of the packet - # generator interfaces. +""" test framework utilities """ + +import socket +import sys +from abc import abstractmethod, ABCMeta +from cStringIO import StringIO +from scapy.layers.inet6 import in6_mactoifaceid + +from scapy.layers.l2 import Ether +from scapy.packet import Raw +from scapy.layers.inet import IP, UDP, TCP +from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest +from scapy.packet import Packet +from socket import inet_pton, AF_INET, AF_INET6 + + +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 + 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. """ + + @property + def mac(self): + """ MAC address """ + return self._mac + + @property + def bin_mac(self): + """ MAC address """ + return mactobinary(self._mac) + + @property + def ip4(self): + """ IPv4 address - string """ + return self._ip4 + + @property + def ip4n(self): + """ 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 - string """ + return self._ip6 + + @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) + + +class L4_Conn(): + """ L4 'connection' tied to two VPP interfaces """ + def __init__(self, testcase, if1, if2, af, l4proto, port1, port2): + self.testcase = testcase + self.ifs = [None, None] + self.ifs[0] = if1 + self.ifs[1] = if2 + self.address_family = af + self.l4proto = l4proto + self.ports = [None, None] + self.ports[0] = port1 + self.ports[1] = port2 + self + + def pkt(self, side, l4args={}, payload="x"): + is_ip6 = 1 if self.address_family == AF_INET6 else 0 + s0 = side + s1 = 1-side + src_if = self.ifs[s0] + dst_if = self.ifs[s1] + layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4), + IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)] + merged_l4args = {'sport': self.ports[s0], 'dport': self.ports[s1]} + merged_l4args.update(l4args) + p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) / + layer_3[is_ip6] / + self.l4proto(**merged_l4args) / + Raw(payload)) + return p + + def send(self, side, flags=None, payload=""): + l4args = {} + if flags is not None: + l4args['flags'] = flags + self.ifs[side].add_stream(self.pkt(side, + l4args=l4args, payload=payload)) + self.ifs[1-side].enable_capture() + self.testcase.pg_start() + + def recv(self, side): + p = self.ifs[side].wait_for_packet(1) + return p + + def send_through(self, side, flags=None, payload=""): + self.send(side, flags, payload) + p = self.recv(1-side) + return p + + def send_pingpong(self, side, flags1=None, flags2=None): + p1 = self.send_through(side, flags1) + p2 = self.send_through(1-side, flags2) + return [p1, p2] + + +class L4_CONN_SIDE: + L4_CONN_SIDE_ZERO = 0 + L4_CONN_SIDE_ONE = 1