import unittest
from scapy.packet import Raw
-from scapy.layers.l2 import Ether
+from scapy.layers.l2 import Ether, Dot1Q, GRE
from scapy.layers.inet import IP, UDP
+from scapy.layers.vxlan import VXLAN
from framework import VppTestCase, VppTestRunner
from util import Host, ppp
+from vpp_sub_interface import VppDot1QSubint, VppDot1ADSubint
+from vpp_gre_interface import VppGreInterface, VppGre6Interface
+from vpp_papi_provider import L2_VTR_OP
+from collections import namedtuple
+
+Tag = namedtuple('Tag', ['dot1', 'vlan'])
+DOT1AD = 0x88A8
+DOT1Q = 0x8100
class TestSpan(VppTestCase):
""" SPAN Test Case """
- # Test variables
- hosts_nr = 10 # Number of hosts
- pkts_per_burst = 257 # Number of packets per burst
-
@classmethod
def setUpClass(cls):
super(TestSpan, cls).setUpClass()
-
- def setUp(self):
- super(TestSpan, self).setUp()
-
+ # Test variables
+ cls.hosts_nr = 10 # Number of hosts
+ cls.pkts_per_burst = 257 # Number of packets per burst
# create 3 pg interfaces
- self.create_pg_interfaces(range(3))
+ cls.create_pg_interfaces(range(3))
+ cls.bd_id = 55
+ cls.sub_if = VppDot1QSubint(cls, cls.pg0, 100)
+ cls.dst_sub_if = VppDot1QSubint(cls, cls.pg2, 300)
+ cls.dst_sub_if.set_vtr(L2_VTR_OP.L2_POP_1, tag=300)
# packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
- self.flows = dict()
- self.flows[self.pg0] = [self.pg1]
+ cls.flows = dict()
+ cls.flows[cls.pg0] = [cls.pg1]
# packet sizes
- self.pg_if_packet_sizes = [64, 512] # , 1518, 9018]
+ cls.pg_if_packet_sizes = [64, 512] # , 1518, 9018]
- self.interfaces = list(self.pg_interfaces)
+ cls.interfaces = list(cls.pg_interfaces)
# Create host MAC and IPv4 lists
- # self.MY_MACS = dict()
- # self.MY_IP4S = dict()
- self.create_host_lists(TestSpan.hosts_nr)
-
- # Create bi-directional cross-connects between pg0 and pg1
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg0.sw_if_index, self.pg1.sw_if_index, enable=1)
- self.vapi.sw_interface_set_l2_xconnect(
- self.pg1.sw_if_index, self.pg0.sw_if_index, enable=1)
+ # cls.MY_MACS = dict()
+ # cls.MY_IP4S = dict()
+ cls.create_host_lists(cls.hosts_nr)
# setup all interfaces
- for i in self.interfaces:
+ for i in cls.interfaces:
i.admin_up()
i.config_ip4()
i.resolve_arp()
- # Enable SPAN on pg0 (mirrored to pg2)
- self.vapi.sw_interface_span_enable_disable(
- self.pg0.sw_if_index, self.pg2.sw_if_index)
+ cls.vxlan = cls.vapi.vxlan_add_del_tunnel(
+ src_addr=cls.pg2.local_ip4n,
+ dst_addr=cls.pg2.remote_ip4n,
+ vni=1111,
+ is_add=1)
+
+ def setUp(self):
+ super(TestSpan, self).setUp()
+ self.reset_packet_infos()
def tearDown(self):
super(TestSpan, self).tearDown()
+ if not self.vpp_dead:
+ self.logger.info(self.vapi.ppcli("show interface span"))
+
+ def xconnect(self, a, b, is_add=1):
+ self.vapi.sw_interface_set_l2_xconnect(a, b, enable=is_add)
+ self.vapi.sw_interface_set_l2_xconnect(b, a, enable=is_add)
+
+ def bridge(self, sw_if_index, is_add=1):
+ self.vapi.sw_interface_set_l2_bridge(
+ sw_if_index, bd_id=self.bd_id, enable=is_add)
+
+ def _remove_tag(self, packet, vlan, tag_type):
+ self.assertEqual(packet.type, tag_type)
+ payload = packet.payload
+ self.assertEqual(payload.vlan, vlan)
+ inner_type = payload.type
+ payload = payload.payload
+ packet.remove_payload()
+ packet.add_payload(payload)
+ packet.type = inner_type
+
+ def remove_tags(self, packet, tags):
+ for t in tags:
+ self._remove_tag(packet, t.vlan, t.dot1)
+ return packet
+ def decap_gre(self, pkt):
+ """
+ Decapsulate the original payload frame by removing GRE header
+ """
+ self.assertEqual(pkt[Ether].src, self.pg2.local_mac)
+ self.assertEqual(pkt[Ether].dst, self.pg2.remote_mac)
+
+ self.assertEqual(pkt[IP].src, self.pg2.local_ip4)
+ self.assertEqual(pkt[IP].dst, self.pg2.remote_ip4)
+
+ return pkt[GRE].payload
+
+ def decap_vxlan(self, pkt):
+ """
+ Decapsulate the original payload frame by removing VXLAN header
+ """
+ self.assertEqual(pkt[Ether].src, self.pg2.local_mac)
+ self.assertEqual(pkt[Ether].dst, self.pg2.remote_mac)
+
+ self.assertEqual(pkt[IP].src, self.pg2.local_ip4)
+ self.assertEqual(pkt[IP].dst, self.pg2.remote_ip4)
+
+ return pkt[VXLAN].payload
+
+ @classmethod
def create_host_lists(self, count):
""" Method to create required number of MAC and IPv4 addresses.
Create required number of host MAC addresses and distribute them among
"172.17.1%02x.%u" % (pg_if.sw_if_index, j))
hosts.append(host)
- def create_stream(self, src_if, packet_sizes):
+ def create_stream(self, src_if, packet_sizes, do_dot1=False):
pkts = []
- for i in range(0, TestSpan.pkts_per_burst):
+ for i in range(0, self.pkts_per_burst):
dst_if = self.flows[src_if][0]
pkt_info = self.create_packet_info(src_if, dst_if)
payload = self.info_to_payload(pkt_info)
IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
UDP(sport=1234, dport=1234) /
Raw(payload))
+ if do_dot1:
+ p = self.sub_if.add_dot1_layer(p)
pkt_info.data = p.copy()
size = packet_sizes[(i / 2) % len(packet_sizes)]
self.extend_packet(p, size)
"Port %u: Packet expected from source %u didn't"
" arrive" % (dst_sw_if_index, i.sw_if_index))
- def test_span(self):
- """ SPAN test
+ def test_device_span(self):
+ """ SPAN device rx mirror test
Test scenario:
1. config
burst of packets per interface
"""
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index)
# Create incoming packet streams for packet-generator interfaces
pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
self.pg0.add_stream(pkts)
+ # Enable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.pg0.sw_if_index, self.pg2.sw_if_index)
+
+ self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.pg1.get_capture(),
self.pg2.get_capture(pg2_expected))
+ # Disable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.pg0.sw_if_index, self.pg2.sw_if_index, state=0)
+ self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
+ def test_span_l2_rx(self):
+ """ SPAN l2 rx mirror test """
+
+ self.sub_if.admin_up()
+
+ self.bridge(self.pg2.sw_if_index)
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index)
+ # Create incoming packet streams for packet-generator interfaces
+ pkts = self.create_stream(
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ self.pg0.add_stream(pkts)
+
+ # Enable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1)
+
+ self.logger.info(self.vapi.ppcli("show interface span"))
+ # Enable packet capturing and start packet sending
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Verify packets outgoing packet streams on mirrored interface (pg2)
+ self.logger.info("Verifying capture on interfaces %s and %s" %
+ (self.pg1.name, self.pg2.name))
+ pg2_expected = self.get_packet_count_for_if_idx(self.pg1.sw_if_index)
+ pg1_pkts = self.pg1.get_capture()
+ pg2_pkts = self.pg2.get_capture(pg2_expected)
+ self.verify_capture(
+ self.pg1,
+ pg1_pkts,
+ pg2_pkts)
+
+ self.bridge(self.pg2.sw_if_index, is_add=0)
+ # Disable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1)
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
+ def test_span_l2_rx_dst_vxlan(self):
+ """ SPAN l2 rx mirror into vxlan test """
+
+ self.sub_if.admin_up()
+ self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index,
+ admin_up_down=1)
+
+ self.bridge(self.vxlan.sw_if_index, is_add=1)
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index)
+ # Create incoming packet streams for packet-generator interfaces
+ pkts = self.create_stream(
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ self.pg0.add_stream(pkts)
+
+ # Enable SPAN on pg0 sub if (mirrored to vxlan)
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.vxlan.sw_if_index, is_l2=1)
+
+ self.logger.info(self.vapi.ppcli("show interface span"))
+ # Enable packet capturing and start packet sending
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Verify packets outgoing packet streams on mirrored interface (pg2)
+ self.logger.info("Verifying capture on interfaces %s and %s" %
+ (self.pg1.name, self.pg2.name))
+ pg2_expected = self.get_packet_count_for_if_idx(self.pg1.sw_if_index)
+ pg1_pkts = self.pg1.get_capture()
+ pg2_pkts = [self.decap_vxlan(p)
+ for p in self.pg2.get_capture(pg2_expected)]
+ self.verify_capture(
+ self.pg1,
+ pg1_pkts,
+ pg2_pkts)
+
+ self.bridge(self.vxlan.sw_if_index, is_add=0)
+ # Disable SPAN on pg0 sub if (mirrored to vxlan)
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.vxlan.sw_if_index, state=0, is_l2=1)
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
+ def test_span_l2_rx_dst_gre_subif_vtr(self):
+ """ SPAN l2 rx mirror into gre-subif+vtr """
+
+ self.sub_if.admin_up()
+
+ gre_if = VppGreInterface(self, self.pg2.local_ip4,
+ self.pg2.remote_ip4,
+ is_teb=1)
+
+ gre_if.add_vpp_config()
+ gre_if.admin_up()
+
+ gre_sub_if = VppDot1QSubint(self, gre_if, 500)
+ gre_sub_if.set_vtr(L2_VTR_OP.L2_POP_1, tag=500)
+ gre_sub_if.admin_up()
+
+ self.bridge(gre_sub_if.sw_if_index)
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
+
+ # Create incoming packet streams for packet-generator interfaces
+ pkts = self.create_stream(
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ self.pg0.add_stream(pkts)
+
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, gre_sub_if.sw_if_index, is_l2=1)
+
+ # Enable packet capturing and start packet sending
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Verify packets outgoing packet streams on mirrored interface (pg2)
+ self.logger.info("Verifying capture on interfaces %s and %s" %
+ (self.pg1.name, self.pg2.name))
+ pg2_expected = self.get_packet_count_for_if_idx(self.pg1.sw_if_index)
+ pg1_pkts = self.pg1.get_capture()
+ pg2_pkts = self.pg2.get_capture(pg2_expected)
+ pg2_decaped = [self.remove_tags(self.decap_gre(
+ p), [Tag(dot1=DOT1Q, vlan=500)]) for p in pg2_pkts]
+ self.verify_capture(
+ self.pg1,
+ pg1_pkts,
+ pg2_decaped)
+
+ self.bridge(gre_sub_if.sw_if_index, is_add=0)
+ # Disable SPAN on pg0 sub if
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, gre_sub_if.sw_if_index, state=0,
+ is_l2=1)
+ gre_if.remove_vpp_config()
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
+ def test_span_l2_rx_dst_vtr(self):
+ """ SPAN l2 rx mirror into subif+vtr """
+
+ self.sub_if.admin_up()
+ self.dst_sub_if.admin_up()
+
+ self.bridge(self.dst_sub_if.sw_if_index)
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
+
+ # Create incoming packet streams for packet-generator interfaces
+ pkts = self.create_stream(
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ self.pg0.add_stream(pkts)
+
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.dst_sub_if.sw_if_index, is_l2=1)
+
+ # Enable packet capturing and start packet sending
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Verify packets outgoing packet streams on mirrored interface (pg2)
+ self.logger.info("Verifying capture on interfaces %s and %s" %
+ (self.pg1.name, self.pg2.name))
+ pg2_expected = self.get_packet_count_for_if_idx(self.pg1.sw_if_index)
+ pg1_pkts = self.pg1.get_capture()
+ pg2_pkts = self.pg2.get_capture(pg2_expected)
+ pg2_untagged = [self.remove_tags(p, [Tag(dot1=DOT1Q, vlan=300)])
+ for p in pg2_pkts]
+ self.verify_capture(
+ self.pg1,
+ pg1_pkts,
+ pg2_untagged)
+
+ self.bridge(self.dst_sub_if.sw_if_index, is_add=0)
+ # Disable SPAN on pg0 sub if (mirrored to vxlan)
+ self.vapi.sw_interface_span_enable_disable(
+ self.sub_if.sw_if_index, self.dst_sub_if.sw_if_index, state=0,
+ is_l2=1)
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
+ def test_l2_tx_span(self):
+ """ SPAN l2 tx mirror test """
+
+ self.sub_if.admin_up()
+ self.bridge(self.pg2.sw_if_index)
+ # Create bi-directional cross-connects between pg0 and pg1
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index)
+ # Create incoming packet streams for packet-generator interfaces
+ pkts = self.create_stream(
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ self.pg0.add_stream(pkts)
+
+ # Enable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.pg1.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=2)
+
+ self.logger.info(self.vapi.ppcli("show interface span"))
+ # Enable packet capturing and start packet sending
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ # Verify packets outgoing packet streams on mirrored interface (pg2)
+ self.logger.info("Verifying capture on interfaces %s and %s" %
+ (self.pg1.name, self.pg2.name))
+ pg2_expected = self.get_packet_count_for_if_idx(self.pg1.sw_if_index)
+ pg1_pkts = self.pg1.get_capture()
+ pg2_pkts = self.pg2.get_capture(pg2_expected)
+ self.verify_capture(
+ self.pg1,
+ pg1_pkts,
+ pg2_pkts)
+
+ self.bridge(self.pg2.sw_if_index, is_add=0)
+ # Disable SPAN on pg0 (mirrored to pg2)
+ self.vapi.sw_interface_span_enable_disable(
+ self.pg1.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1)
+ self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=0)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff