from vpp_papi import VppEnum
-Tag = namedtuple('Tag', ['dot1', 'vlan'])
+Tag = namedtuple("Tag", ["dot1", "vlan"])
DOT1AD = 0x88A8
DOT1Q = 0x8100
class TestSpan(VppTestCase):
- """ SPAN Test Case """
+ """SPAN Test Case"""
@classmethod
def setUpClass(cls):
super(TestSpan, cls).setUpClass()
# Test variables
- cls.pkts_per_burst = 257 # Number of packets per burst
+ cls.pkts_per_burst = 257 # Number of packets per burst
# create 3 pg interfaces
cls.create_pg_interfaces(range(3))
def setUp(self):
super(TestSpan, self).setUp()
- self.vxlan = VppVxlanTunnel(self, src=self.pg2.local_ip4,
- dst=self.pg2.remote_ip4, vni=1111)
+ self.vxlan = VppVxlanTunnel(
+ self, src=self.pg2.local_ip4, dst=self.pg2.remote_ip4, vni=1111
+ )
self.vxlan.add_vpp_config()
self.reset_packet_infos()
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(rx_sw_if_index=sw_if_index,
- bd_id=self.bd_id, enable=is_add)
+ self.vapi.sw_interface_set_l2_bridge(
+ rx_sw_if_index=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)
for i in range(0, self.pkts_per_burst):
payload = "span test"
size = packet_sizes[int((i / 2) % len(packet_sizes))]
- p = (Ether(src=src_if.local_mac, dst=dst_mac) /
- IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
- UDP(sport=10000 + src_if.sw_if_index * 1000 + i, dport=1234) /
- Raw(payload))
+ p = (
+ Ether(src=src_if.local_mac, dst=dst_mac)
+ / IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4)
+ / UDP(sport=10000 + src_if.sw_if_index * 1000 + i, dport=1234)
+ / Raw(payload)
+ )
if do_dot1:
p = self.sub_if.add_dot1_layer(p)
self.extend_packet(p, size)
return pkts
def verify_capture(self, cap1, cap2):
- self.assertEqual(len(cap1), len(cap2),
- "Different number of sent and mirrored packets :"
- "%u != %u" % (len(cap1), len(cap2)))
+ self.assertEqual(
+ len(cap1),
+ len(cap2),
+ "Different number of sent and mirrored packets :"
+ "%u != %u" % (len(cap1), len(cap2)),
+ )
pkts1 = [(pkt[Ether] / pkt[IP] / pkt[UDP]) for pkt in cap1]
pkts2 = [(pkt[Ether] / pkt[IP] / pkt[UDP]) for pkt in cap2]
self.assertEqual(pkts1.sort(), pkts2.sort())
def test_device_span(self):
- """ SPAN device rx mirror """
+ """SPAN device rx mirror"""
# Create bi-directional cross-connects between pg0 and pg1
self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index)
# 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.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
# 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.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)
self.verify_capture(pg1_pkts, pg2_pkts)
def test_span_l2_rx(self):
- """ SPAN l2 rx mirror """
+ """SPAN l2 rx mirror"""
self.sub_if.admin_up()
# Create bi-directional cross-connects between pg0 subif 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)
+ 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.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
# 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.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)
self.verify_capture(pg1_pkts, pg2_pkts)
def test_span_l2_rx_dst_vxlan(self):
- """ SPAN l2 rx mirror into vxlan """
+ """SPAN l2 rx mirror into vxlan"""
self.sub_if.admin_up()
- self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index,
- flags=1)
+ self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index, flags=1)
self.bridge(self.vxlan.sw_if_index, is_add=1)
# Create bi-directional cross-connects between pg0 subif 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)
+ 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.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.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.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)
self.verify_capture(pg1_pkts, pg2_pkts)
def test_span_l2_rx_dst_gre_erspan(self):
- """ SPAN l2 rx mirror into gre-erspan """
+ """SPAN l2 rx mirror into gre-erspan"""
self.sub_if.admin_up()
- gre_if = VppGreInterface(self, self.pg2.local_ip4,
- self.pg2.remote_ip4,
- session=543,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_ERSPAN))
+ gre_if = VppGreInterface(
+ self,
+ self.pg2.local_ip4,
+ self.pg2.remote_ip4,
+ session=543,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_ERSPAN),
+ )
gre_if.add_vpp_config()
gre_if.admin_up()
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)
+ 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 gre-erspan)
self.vapi.sw_interface_span_enable_disable(
- self.sub_if.sw_if_index, gre_if.sw_if_index, is_l2=1)
+ self.sub_if.sw_if_index, gre_if.sw_if_index, is_l2=1
+ )
# Enable packet capturing and start packet sending
self.pg_enable_capture(self.pg_interfaces)
pg1_pkts = self.pg1.get_capture(n_pkts)
pg2_pkts = self.pg2.get_capture(n_pkts)
- def decap(p): return self.decap_erspan(p, session=543)
+ def decap(p):
+ return self.decap_erspan(p, session=543)
+
pg2_decaped = [decap(p) for p in pg2_pkts]
self.bridge(gre_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_if.sw_if_index, state=0, is_l2=1)
+ self.sub_if.sw_if_index, gre_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)
self.verify_capture(pg1_pkts, pg2_decaped)
def test_span_l2_rx_dst_gre_subif_vtr(self):
- """ SPAN l2 rx mirror into gre-subif+vtr """
+ """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,
- type=(VppEnum.vl_api_gre_tunnel_type_t.
- GRE_API_TUNNEL_TYPE_TEB))
+ gre_if = VppGreInterface(
+ self,
+ self.pg2.local_ip4,
+ self.pg2.remote_ip4,
+ type=(VppEnum.vl_api_gre_tunnel_type_t.GRE_API_TUNNEL_TYPE_TEB),
+ )
gre_if.add_vpp_config()
gre_if.admin_up()
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)
+ 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 gre sub if)
self.vapi.sw_interface_span_enable_disable(
- self.sub_if.sw_if_index, gre_sub_if.sw_if_index, is_l2=1)
+ 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)
pg1_pkts = self.pg1.get_capture(n_pkts)
pg2_pkts = self.pg2.get_capture(n_pkts)
- def decap(p): return self.remove_tags(
- self.decap_gre(p), [Tag(dot1=DOT1Q, vlan=500)])
+ def decap(p):
+ return self.remove_tags(self.decap_gre(p), [Tag(dot1=DOT1Q, vlan=500)])
+
pg2_decaped = [decap(p) for p in pg2_pkts]
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)
+ 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)
self.verify_capture(pg1_pkts, pg2_decaped)
def test_span_l2_rx_dst_1q_vtr(self):
- """ SPAN l2 rx mirror into 1q subif+vtr """
+ """SPAN l2 rx mirror into 1q subif+vtr"""
self.sub_if.admin_up()
self.vlan_sub_if.admin_up()
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)
+ 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.vlan_sub_if.sw_if_index, is_l2=1)
+ self.sub_if.sw_if_index, self.vlan_sub_if.sw_if_index, is_l2=1
+ )
# Enable packet capturing and start packet sending
self.pg_enable_capture(self.pg_interfaces)
n_pkts = len(pkts)
pg1_pkts = self.pg1.get_capture(n_pkts)
pg2_pkts = self.pg2.get_capture(n_pkts)
- pg2_untagged = [self.remove_tags(p, [Tag(dot1=DOT1Q, vlan=300)])
- for p in pg2_pkts]
+ pg2_untagged = [
+ self.remove_tags(p, [Tag(dot1=DOT1Q, vlan=300)]) for p in pg2_pkts
+ ]
self.bridge(self.vlan_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.vlan_sub_if.sw_if_index, state=0,
- is_l2=1)
+ self.sub_if.sw_if_index, self.vlan_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)
self.verify_capture(pg1_pkts, pg2_untagged)
def test_span_l2_rx_dst_1ad_vtr(self):
- """ SPAN l2 rx mirror into 1ad subif+vtr """
+ """SPAN l2 rx mirror into 1ad subif+vtr"""
self.sub_if.admin_up()
self.qinq_sub_if.admin_up()
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)
+ 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.qinq_sub_if.sw_if_index, is_l2=1)
+ self.sub_if.sw_if_index, self.qinq_sub_if.sw_if_index, is_l2=1
+ )
# Enable packet capturing and start packet sending
self.pg_enable_capture(self.pg_interfaces)
n_pkts = len(pkts)
pg1_pkts = self.pg1.get_capture(n_pkts)
pg2_pkts = self.pg2.get_capture(n_pkts)
- pg2_untagged = [self.remove_tags(p, [Tag(dot1=DOT1AD, vlan=400),
- Tag(dot1=DOT1Q, vlan=500)])
- for p in pg2_pkts]
+ pg2_untagged = [
+ self.remove_tags(p, [Tag(dot1=DOT1AD, vlan=400), Tag(dot1=DOT1Q, vlan=500)])
+ for p in pg2_pkts
+ ]
self.bridge(self.qinq_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.qinq_sub_if.sw_if_index, state=0,
- is_l2=1)
+ self.sub_if.sw_if_index, self.qinq_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)
self.verify_capture(pg1_pkts, pg2_untagged)
def test_l2_tx_span(self):
- """ SPAN l2 tx mirror """
+ """SPAN l2 tx mirror"""
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)
+ pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True)
self.pg0.add_stream(pkts)
# Enable SPAN on pg1 (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.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.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.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)
self.verify_capture(pg1_pkts, pg2_pkts)
def test_l2_rx_tx_span(self):
- """ SPAN l2 rx tx mirror """
+ """SPAN l2 rx tx mirror"""
self.sub_if.admin_up()
self.bridge(self.pg2.sw_if_index)
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index)
# Create incoming packet streams for packet-generator interfaces
- pg0_pkts = self.create_stream(
- self.pg0, self.pg_if_packet_sizes, do_dot1=True)
+ pg0_pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes, do_dot1=True)
self.pg0.add_stream(pg0_pkts)
- pg1_pkts = self.create_stream(
- self.pg1, self.pg_if_packet_sizes, do_dot1=False)
+ pg1_pkts = self.create_stream(self.pg1, self.pg_if_packet_sizes, do_dot1=False)
self.pg1.add_stream(pg1_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, state=3)
+ self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3
+ )
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
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.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)
self.verify_capture(pg0_pkts + pg1_pkts, pg2_pkts)
def test_l2_bcast_mirror(self):
- """ SPAN l2 broadcast mirror """
+ """SPAN l2 broadcast mirror"""
self.sub_if.admin_up()
self.bridge(self.pg2.sw_if_index)
# Create bi-directional cross-connects between pg0 and pg1
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=1)
+ rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=1
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=1)
+ rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=1
+ )
# Create incoming packet streams for packet-generator interfaces
pg0_pkts = self.create_stream(
- self.pg0, self.pg_if_packet_sizes, do_dot1=True, bcast=True)
+ self.pg0, self.pg_if_packet_sizes, do_dot1=True, bcast=True
+ )
self.pg0.add_stream(pg0_pkts)
pg1_pkts = self.create_stream(
- self.pg1, self.pg_if_packet_sizes, do_dot1=False, bcast=True)
+ self.pg1, self.pg_if_packet_sizes, do_dot1=False, bcast=True
+ )
self.pg1.add_stream(pg1_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, state=3)
+ self.sub_if.sw_if_index, self.pg2.sw_if_index, is_l2=1, state=3
+ )
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
self.bridge(self.pg2.sw_if_index, is_add=0)
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=0)
+ rx_sw_if_index=self.sub_if.sw_if_index, bd_id=99, enable=0
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=0)
+ rx_sw_if_index=self.pg1.sw_if_index, bd_id=99, enable=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.sub_if.sw_if_index, self.pg2.sw_if_index, state=0, is_l2=1
+ )
self.verify_capture(pg0_pkts + pg1_pkts, pg2_pkts)
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff