test/test_ip.py - fdio/vpp - Gitiles

 #!/usr/bin/env python

 import logging
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)

 import unittest
 from framework import VppTestCase, VppTestRunner
 from util import Util

 from scapy.packet import Raw
 from scapy.layers.l2 import Ether, ARP, Dot1Q
 from scapy.layers.inet import IP, UDP


 class TestIPv4(Util, VppTestCase):
     """ IPv4 Test Case """

     @classmethod
     def setUpClass(cls):
         super(TestIPv4, cls).setUpClass()

         try:
             cls.create_interfaces_and_subinterfaces()

             # configure IPv4 on hardware interfaces
             cls.config_ip4(cls.interfaces)

             cls.config_ip4_on_software_interfaces(cls.interfaces)

             # resolve ARP using hardware interfaces
             cls.resolve_arp(cls.interfaces)

             # let VPP know MAC addresses of peer (sub)interfaces
             cls.resolve_arp_on_software_interfaces(cls.interfaces)

             # config 2M FIB enries
             cls.config_fib_entries(2000000)

         except Exception as e:
             super(TestIPv4, cls).tearDownClass()
             raise

     def tearDown(self):
         self.cli(2, "show int")
         self.cli(2, "show trace")
         self.cli(2, "show hardware")
         self.cli(2, "show ip arp")
         # self.cli(2, "show ip fib")  # 2M entries
         self.cli(2, "show error")
         self.cli(2, "show run")

     @classmethod
     def create_vlan_subif(cls, pg_index, vlan):
         cls.api("create_vlan_subif pg%u vlan %u" % (pg_index, vlan))

     @classmethod
     def create_dot1ad_subif(cls, pg_index, sub_id, outer_vlan_id, inner_vlan_id):
         cls.api("create_subif pg%u sub_id %u outer_vlan_id %u inner_vlan_id %u dot1ad"
                  % (pg_index, sub_id, outer_vlan_id, inner_vlan_id))

     class SoftInt(object):
         pass

     class HardInt(SoftInt):
         pass

     class Subint(SoftInt):
         def __init__(self, sub_id):
             self.sub_id = sub_id

     class Dot1QSubint(Subint):
         def __init__(self, sub_id, vlan=None):
             if vlan is None:
                 vlan = sub_id
             super(TestIPv4.Dot1QSubint, self).__init__(sub_id)
             self.vlan = vlan

     class Dot1ADSubint(Subint):
         def __init__(self, sub_id, outer_vlan, inner_vlan):
             super(TestIPv4.Dot1ADSubint, self).__init__(sub_id)
             self.outer_vlan = outer_vlan
             self.inner_vlan = inner_vlan

     @classmethod
     def create_interfaces_and_subinterfaces(cls):
         cls.interfaces = range(3)

         cls.create_interfaces(cls.interfaces)

         # Make vpp_api_test see interfaces created using debug CLI (in function create_interfaces)
         cls.api("sw_interface_dump")

         cls.INT_DETAILS = dict()

         cls.INT_DETAILS[0] = cls.HardInt()

         cls.INT_DETAILS[1] = cls.Dot1QSubint(100)
         cls.create_vlan_subif(1, cls.INT_DETAILS[1].vlan)

         # FIXME: Wrong packet format/wrong layer on output of interface 2
         #self.INT_DETAILS[2] = self.Dot1ADSubint(10, 200, 300)
         #self.create_dot1ad_subif(2, self.INT_DETAILS[2].sub_id, self.INT_DETAILS[2].outer_vlan, self.INT_DETAILS[2].inner_vlan)

         # Use dor1q for now
         cls.INT_DETAILS[2] = cls.Dot1QSubint(200)
         cls.create_vlan_subif(2, cls.INT_DETAILS[2].vlan)

         for i in cls.interfaces:
             det = cls.INT_DETAILS[i]
             if isinstance(det, cls.Subint):
                 cls.api("sw_interface_set_flags pg%u.%u admin-up" % (i, det.sub_id))

     # IP adresses on subinterfaces
     MY_SOFT_IP4S = {}
     VPP_SOFT_IP4S = {}

     @classmethod
     def config_ip4_on_software_interfaces(cls, args):
         for i in args:
             cls.MY_SOFT_IP4S[i] = "172.17.%u.2" % i
             cls.VPP_SOFT_IP4S[i] = "172.17.%u.1" % i
             if isinstance(cls.INT_DETAILS[i], cls.Subint):
                 interface = "pg%u.%u" % (i, cls.INT_DETAILS[i].sub_id)
             else:
                 interface = "pg%u" % i
             cls.api("sw_interface_add_del_address %s %s/24" % (interface, cls.VPP_SOFT_IP4S[i]))
             cls.log("My subinterface IPv4 address is %s" % (cls.MY_SOFT_IP4S[i]))

     # let VPP know MAC addresses of peer (sub)interfaces
     @classmethod
     def resolve_arp_on_software_interfaces(cls, args):
         for i in args:
             ip = cls.VPP_SOFT_IP4S[i]
             cls.log("Sending ARP request for %s on port %u" % (ip, i))
             packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) /
                       ARP(op=ARP.who_has, pdst=ip, psrc=cls.MY_SOFT_IP4S[i], hwsrc=cls.MY_MACS[i]))

             cls.add_dot1_layers(i, packet)

             cls.pg_add_stream(i, packet)
             cls.pg_enable_capture([i])

             cls.cli(2, "trace add pg-input 1")
             cls.pg_start()

             # We don't need to read output

     @classmethod
     def config_fib_entries(cls, count):
         n_int = len(cls.interfaces)
         for i in cls.interfaces:
             cls.api("ip_add_del_route 10.0.0.1/32 via %s count %u" % (cls.VPP_SOFT_IP4S[i], count / n_int))

     @classmethod
     def add_dot1_layers(cls, i, packet):
         assert(type(packet) is Ether)
         payload = packet.payload
         det = cls.INT_DETAILS[i]
         if isinstance(det, cls.Dot1QSubint):
             packet.remove_payload()
             packet.add_payload(Dot1Q(vlan=det.sub_id) / payload)
         elif isinstance(det, cls.Dot1ADSubint):
             packet.remove_payload()
             packet.add_payload(Dot1Q(vlan=det.outer_vlan) / Dot1Q(vlan=det.inner_vlan) / payload)
             packet.type = 0x88A8

     def remove_dot1_layers(self, i, packet):
         self.assertEqual(type(packet), Ether)
         payload = packet.payload
         det = self.INT_DETAILS[i]
         if isinstance(det, self.Dot1QSubint):
             self.assertEqual(type(payload), Dot1Q)
             self.assertEqual(payload.vlan, self.INT_DETAILS[i].vlan)
             payload = payload.payload
         elif isinstance(det, self.Dot1ADSubint):  # TODO: change 88A8 type
             self.assertEqual(type(payload), Dot1Q)
             self.assertEqual(payload.vlan, self.INT_DETAILS[i].outer_vlan)
             payload = payload.payload
             self.assertEqual(type(payload), Dot1Q)
             self.assertEqual(payload.vlan, self.INT_DETAILS[i].inner_vlan)
             payload = payload.payload
         packet.remove_payload()
         packet.add_payload(payload)

     def create_stream(self, pg_id):
         pg_targets = [None] * 3
         pg_targets[0] = [1, 2]
         pg_targets[1] = [0, 2]
         pg_targets[2] = [0, 1]
         pkts = []
         for i in range(0, 257):
             target_pg_id = pg_targets[pg_id][i % 2]
             info = self.create_packet_info(pg_id, target_pg_id)
             payload = self.info_to_payload(info)
             p = (Ether(dst=self.VPP_MACS[pg_id], src=self.MY_MACS[pg_id]) /
                  IP(src=self.MY_SOFT_IP4S[pg_id], dst=self.MY_SOFT_IP4S[target_pg_id]) /
                  UDP(sport=1234, dport=1234) /
                  Raw(payload))
             info.data = p.copy()
             self.add_dot1_layers(pg_id, p)
             if not isinstance(self.INT_DETAILS[pg_id], self.Subint):
                 packet_sizes = [64, 512, 1518, 9018]
             else:
                 packet_sizes = [64, 512, 1518+4, 9018+4]
             size = packet_sizes[(i / 2) % len(packet_sizes)]
             self.extend_packet(p, size)
             pkts.append(p)
         return pkts

     def verify_capture(self, o, capture):
         last_info = {}
         for i in self.interfaces:
             last_info[i] = None
         for packet in capture:
             self.remove_dot1_layers(o, packet)  # Check VLAN tags and Ethernet header
             self.assertTrue(Dot1Q not in packet)
             try:
                 ip = packet[IP]
                 udp = packet[UDP]
                 payload_info = self.payload_to_info(str(packet[Raw]))
                 packet_index = payload_info.index
                 src_pg = payload_info.src
                 dst_pg = payload_info.dst
                 self.assertEqual(dst_pg, o)
                 self.log("Got packet on port %u: src=%u (id=%u)" % (o, src_pg, packet_index), 2)
                 next_info = self.get_next_packet_info_for_interface2(src_pg, dst_pg, last_info[src_pg])
                 last_info[src_pg] = next_info
                 self.assertTrue(next_info is not None)
                 self.assertEqual(packet_index, next_info.index)
                 saved_packet = next_info.data
                 # Check standard fields
                 self.assertEqual(ip.src, saved_packet[IP].src)
                 self.assertEqual(ip.dst, saved_packet[IP].dst)
                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
             except:
                 self.log("Unexpected or invalid packet:")
                 packet.show()
                 raise
         for i in self.interfaces:
             remaining_packet = self.get_next_packet_info_for_interface2(i, o, last_info[i])
             self.assertTrue(remaining_packet is None, "Port %u: Packet expected from source %u didn't arrive" % (o, i))

     def test_fib(self):
         """ IPv4 FIB test """

         for i in self.interfaces:
             pkts = self.create_stream(i)
             self.pg_add_stream(i, pkts)

         self.pg_enable_capture(self.interfaces)
         self.pg_start()

         for i in self.interfaces:
             out = self.pg_get_capture(i)
             self.log("Verifying capture %u" % i)
             self.verify_capture(i, out)


 if __name__ == '__main__':
     unittest.main(testRunner = VppTestRunner)
	#!/usr/bin/env python

	import logging
	logging.getLogger("scapy.runtime").setLevel(logging.ERROR)

	import unittest
	from framework import VppTestCase, VppTestRunner
	from util import Util

	from scapy.packet import Raw
	from scapy.layers.l2 import Ether, ARP, Dot1Q
	from scapy.layers.inet import IP, UDP


	class TestIPv4(Util, VppTestCase):
	""" IPv4 Test Case """

	@classmethod
	def setUpClass(cls):
	super(TestIPv4, cls).setUpClass()

	try:
	cls.create_interfaces_and_subinterfaces()

	# configure IPv4 on hardware interfaces
	cls.config_ip4(cls.interfaces)

	cls.config_ip4_on_software_interfaces(cls.interfaces)

	# resolve ARP using hardware interfaces
	cls.resolve_arp(cls.interfaces)

	# let VPP know MAC addresses of peer (sub)interfaces
	cls.resolve_arp_on_software_interfaces(cls.interfaces)

	# config 2M FIB enries
	cls.config_fib_entries(2000000)

	except Exception as e:
	super(TestIPv4, cls).tearDownClass()
	raise

	def tearDown(self):
	self.cli(2, "show int")
	self.cli(2, "show trace")
	self.cli(2, "show hardware")
	self.cli(2, "show ip arp")
	# self.cli(2, "show ip fib") # 2M entries
	self.cli(2, "show error")
	self.cli(2, "show run")

	@classmethod
	def create_vlan_subif(cls, pg_index, vlan):
	cls.api("create_vlan_subif pg%u vlan %u" % (pg_index, vlan))

	@classmethod
	def create_dot1ad_subif(cls, pg_index, sub_id, outer_vlan_id, inner_vlan_id):
	cls.api("create_subif pg%u sub_id %u outer_vlan_id %u inner_vlan_id %u dot1ad"
	% (pg_index, sub_id, outer_vlan_id, inner_vlan_id))

	class SoftInt(object):
	pass

	class HardInt(SoftInt):
	pass

	class Subint(SoftInt):
	def __init__(self, sub_id):
	self.sub_id = sub_id

	class Dot1QSubint(Subint):
	def __init__(self, sub_id, vlan=None):
	if vlan is None:
	vlan = sub_id
	super(TestIPv4.Dot1QSubint, self).__init__(sub_id)
	self.vlan = vlan

	class Dot1ADSubint(Subint):
	def __init__(self, sub_id, outer_vlan, inner_vlan):
	super(TestIPv4.Dot1ADSubint, self).__init__(sub_id)
	self.outer_vlan = outer_vlan
	self.inner_vlan = inner_vlan

	@classmethod
	def create_interfaces_and_subinterfaces(cls):
	cls.interfaces = range(3)

	cls.create_interfaces(cls.interfaces)

	# Make vpp_api_test see interfaces created using debug CLI (in function create_interfaces)
	cls.api("sw_interface_dump")

	cls.INT_DETAILS = dict()

	cls.INT_DETAILS[0] = cls.HardInt()

	cls.INT_DETAILS[1] = cls.Dot1QSubint(100)
	cls.create_vlan_subif(1, cls.INT_DETAILS[1].vlan)

	# FIXME: Wrong packet format/wrong layer on output of interface 2
	#self.INT_DETAILS[2] = self.Dot1ADSubint(10, 200, 300)
	#self.create_dot1ad_subif(2, self.INT_DETAILS[2].sub_id, self.INT_DETAILS[2].outer_vlan, self.INT_DETAILS[2].inner_vlan)

	# Use dor1q for now
	cls.INT_DETAILS[2] = cls.Dot1QSubint(200)
	cls.create_vlan_subif(2, cls.INT_DETAILS[2].vlan)

	for i in cls.interfaces:
	det = cls.INT_DETAILS[i]
	if isinstance(det, cls.Subint):
	cls.api("sw_interface_set_flags pg%u.%u admin-up" % (i, det.sub_id))

	# IP adresses on subinterfaces
	MY_SOFT_IP4S = {}
	VPP_SOFT_IP4S = {}

	@classmethod
	def config_ip4_on_software_interfaces(cls, args):
	for i in args:
	cls.MY_SOFT_IP4S[i] = "172.17.%u.2" % i
	cls.VPP_SOFT_IP4S[i] = "172.17.%u.1" % i
	if isinstance(cls.INT_DETAILS[i], cls.Subint):
	interface = "pg%u.%u" % (i, cls.INT_DETAILS[i].sub_id)
	else:
	interface = "pg%u" % i
	cls.api("sw_interface_add_del_address %s %s/24" % (interface, cls.VPP_SOFT_IP4S[i]))
	cls.log("My subinterface IPv4 address is %s" % (cls.MY_SOFT_IP4S[i]))

	# let VPP know MAC addresses of peer (sub)interfaces
	@classmethod
	def resolve_arp_on_software_interfaces(cls, args):
	for i in args:
	ip = cls.VPP_SOFT_IP4S[i]
	cls.log("Sending ARP request for %s on port %u" % (ip, i))
	packet = (Ether(dst="ff:ff:ff:ff:ff:ff", src=cls.MY_MACS[i]) /
	ARP(op=ARP.who_has, pdst=ip, psrc=cls.MY_SOFT_IP4S[i], hwsrc=cls.MY_MACS[i]))

	cls.add_dot1_layers(i, packet)

	cls.pg_add_stream(i, packet)
	cls.pg_enable_capture([i])

	cls.cli(2, "trace add pg-input 1")
	cls.pg_start()

	# We don't need to read output

	@classmethod
	def config_fib_entries(cls, count):
	n_int = len(cls.interfaces)
	for i in cls.interfaces:
	cls.api("ip_add_del_route 10.0.0.1/32 via %s count %u" % (cls.VPP_SOFT_IP4S[i], count / n_int))

	@classmethod
	def add_dot1_layers(cls, i, packet):
	assert(type(packet) is Ether)
	payload = packet.payload
	det = cls.INT_DETAILS[i]
	if isinstance(det, cls.Dot1QSubint):
	packet.remove_payload()
	packet.add_payload(Dot1Q(vlan=det.sub_id) / payload)
	elif isinstance(det, cls.Dot1ADSubint):
	packet.remove_payload()
	packet.add_payload(Dot1Q(vlan=det.outer_vlan) / Dot1Q(vlan=det.inner_vlan) / payload)
	packet.type = 0x88A8

	def remove_dot1_layers(self, i, packet):
	self.assertEqual(type(packet), Ether)
	payload = packet.payload
	det = self.INT_DETAILS[i]
	if isinstance(det, self.Dot1QSubint):
	self.assertEqual(type(payload), Dot1Q)
	self.assertEqual(payload.vlan, self.INT_DETAILS[i].vlan)
	payload = payload.payload
	elif isinstance(det, self.Dot1ADSubint): # TODO: change 88A8 type
	self.assertEqual(type(payload), Dot1Q)
	self.assertEqual(payload.vlan, self.INT_DETAILS[i].outer_vlan)
	payload = payload.payload
	self.assertEqual(type(payload), Dot1Q)
	self.assertEqual(payload.vlan, self.INT_DETAILS[i].inner_vlan)
	payload = payload.payload
	packet.remove_payload()
	packet.add_payload(payload)

	def create_stream(self, pg_id):
	pg_targets = [None] * 3
	pg_targets[0] = [1, 2]
	pg_targets[1] = [0, 2]
	pg_targets[2] = [0, 1]
	pkts = []
	for i in range(0, 257):
	target_pg_id = pg_targets[pg_id][i % 2]
	info = self.create_packet_info(pg_id, target_pg_id)
	payload = self.info_to_payload(info)
	p = (Ether(dst=self.VPP_MACS[pg_id], src=self.MY_MACS[pg_id]) /
	IP(src=self.MY_SOFT_IP4S[pg_id], dst=self.MY_SOFT_IP4S[target_pg_id]) /
	UDP(sport=1234, dport=1234) /
	Raw(payload))
	info.data = p.copy()
	self.add_dot1_layers(pg_id, p)
	if not isinstance(self.INT_DETAILS[pg_id], self.Subint):
	packet_sizes = [64, 512, 1518, 9018]
	else:
	packet_sizes = [64, 512, 1518+4, 9018+4]
	size = packet_sizes[(i / 2) % len(packet_sizes)]
	self.extend_packet(p, size)
	pkts.append(p)
	return pkts

	def verify_capture(self, o, capture):
	last_info = {}
	for i in self.interfaces:
	last_info[i] = None
	for packet in capture:
	self.remove_dot1_layers(o, packet) # Check VLAN tags and Ethernet header
	self.assertTrue(Dot1Q not in packet)
	try:
	ip = packet[IP]
	udp = packet[UDP]
	payload_info = self.payload_to_info(str(packet[Raw]))
	packet_index = payload_info.index
	src_pg = payload_info.src
	dst_pg = payload_info.dst
	self.assertEqual(dst_pg, o)
	self.log("Got packet on port %u: src=%u (id=%u)" % (o, src_pg, packet_index), 2)
	next_info = self.get_next_packet_info_for_interface2(src_pg, dst_pg, last_info[src_pg])
	last_info[src_pg] = next_info
	self.assertTrue(next_info is not None)
	self.assertEqual(packet_index, next_info.index)
	saved_packet = next_info.data
	# Check standard fields
	self.assertEqual(ip.src, saved_packet[IP].src)
	self.assertEqual(ip.dst, saved_packet[IP].dst)
	self.assertEqual(udp.sport, saved_packet[UDP].sport)
	self.assertEqual(udp.dport, saved_packet[UDP].dport)
	except:
	self.log("Unexpected or invalid packet:")
	packet.show()
	raise
	for i in self.interfaces:
	remaining_packet = self.get_next_packet_info_for_interface2(i, o, last_info[i])
	self.assertTrue(remaining_packet is None, "Port %u: Packet expected from source %u didn't arrive" % (o, i))

	def test_fib(self):
	""" IPv4 FIB test """

	for i in self.interfaces:
	pkts = self.create_stream(i)
	self.pg_add_stream(i, pkts)

	self.pg_enable_capture(self.interfaces)
	self.pg_start()

	for i in self.interfaces:
	out = self.pg_get_capture(i)
	self.log("Verifying capture %u" % i)
	self.verify_capture(i, out)


	if __name__ == '__main__':
	unittest.main(testRunner = VppTestRunner)