diff options
Diffstat (limited to 'test/test_span.py')
| -rw-r--r-- | test/test_span.py | 216 |
1 files changed, 123 insertions, 93 deletions
diff --git a/test/test_span.py b/test/test_span.py index ecefe153706..8eea1b0661d 100644 --- a/test/test_span.py +++ b/test/test_span.py @@ -3,12 +3,12 @@ import unittest from scapy.packet import Raw -from scapy.layers.l2 import Ether, Dot1Q, GRE, ERSPAN +from scapy.layers.l2 import Ether, GRE, ERSPAN from scapy.layers.inet import IP, UDP from scapy.layers.vxlan import VXLAN -from framework import VppTestCase, VppTestRunner -from util import Host, ppp +from framework import VppTestCase +from asfframework import VppTestRunner from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint from vpp_gre_interface import VppGreInterface from vpp_vxlan_tunnel import VppVxlanTunnel @@ -16,19 +16,19 @@ from collections import namedtuple 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)) @@ -56,8 +56,9 @@ class TestSpan(VppTestCase): 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() @@ -72,8 +73,9 @@ class TestSpan(VppTestCase): 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) @@ -145,10 +147,12 @@ class TestSpan(VppTestCase): 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) @@ -156,9 +160,12 @@ class TestSpan(VppTestCase): 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] @@ -166,7 +173,7 @@ class TestSpan(VppTestCase): 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) @@ -176,7 +183,8 @@ class TestSpan(VppTestCase): # 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 @@ -190,13 +198,14 @@ class TestSpan(VppTestCase): # 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() @@ -204,13 +213,13 @@ class TestSpan(VppTestCase): # 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 @@ -225,29 +234,29 @@ class TestSpan(VppTestCase): # 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 @@ -262,20 +271,23 @@ class TestSpan(VppTestCase): 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() @@ -285,13 +297,13 @@ class TestSpan(VppTestCase): 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) @@ -302,28 +314,33 @@ class TestSpan(VppTestCase): 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() @@ -337,13 +354,13 @@ class TestSpan(VppTestCase): 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) @@ -354,22 +371,24 @@ class TestSpan(VppTestCase): 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() @@ -379,12 +398,12 @@ class TestSpan(VppTestCase): 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) @@ -394,20 +413,21 @@ class TestSpan(VppTestCase): 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() @@ -417,12 +437,12 @@ class TestSpan(VppTestCase): 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) @@ -432,34 +452,35 @@ class TestSpan(VppTestCase): 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 @@ -473,13 +494,14 @@ class TestSpan(VppTestCase): 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) @@ -487,16 +509,15 @@ class TestSpan(VppTestCase): 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 @@ -515,34 +536,40 @@ class TestSpan(VppTestCase): 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 @@ -560,15 +587,18 @@ class TestSpan(VppTestCase): 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) |