-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 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 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
+ 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()
def tearDown(self):
super(TestSpan, self).tearDown()
self.reset_packet_infos()
def tearDown(self):
super(TestSpan, self).tearDown()
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):
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(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)
def _remove_tag(self, packet, vlan, tag_type):
self.assertEqual(packet.type, tag_type)
- size = packet_sizes[(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))
+ 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)
+ )
if do_dot1:
p = self.sub_if.add_dot1_layer(p)
self.extend_packet(p, size)
if do_dot1:
p = self.sub_if.add_dot1_layer(p)
self.extend_packet(p, size)
- 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]
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):
self.assertEqual(pkts1.sort(), pkts2.sort())
def test_device_span(self):
# Create bi-directional cross-connects between pg0 and pg1
self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index)
# Create bi-directional cross-connects between pg0 and pg1
self.xconnect(self.pg0.sw_if_index, self.pg1.sw_if_index)
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
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):
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):
# 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
# 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.pg0.add_stream(pkts)
# Enable SPAN on pg0 (mirrored to pg2)
self.vapi.sw_interface_span_enable_disable(
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
- 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):
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):
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
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.pg0.add_stream(pkts)
# Enable SPAN on pg0 sub if (mirrored to vxlan)
self.vapi.sw_interface_span_enable_disable(
self.logger.info(self.vapi.ppcli("show interface span"))
# Enable packet capturing and start packet sending
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.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):
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):
- gre_if = VppGreInterface(self, self.pg2.local_ip4,
- self.pg2.remote_ip4,
- type=2,
- session=543)
+ 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),
+ )
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
# Create incoming packet streams for packet-generator interfaces
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.pg0.add_stream(pkts)
# Enable SPAN on pg0 sub if (mirrored to gre-erspan)
self.vapi.sw_interface_span_enable_disable(
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(
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):
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):
- gre_if = VppGreInterface(self, self.pg2.local_ip4,
- self.pg2.remote_ip4,
- type=1)
+ 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),
+ )
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
# Create incoming packet streams for packet-generator interfaces
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.pg0.add_stream(pkts)
# Enable SPAN on pg0 sub if (mirrored to gre sub if)
self.vapi.sw_interface_span_enable_disable(
- 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(
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):
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):
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
# Create incoming packet streams for packet-generator interfaces
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.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.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):
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):
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index, is_add=1)
# Create incoming packet streams for packet-generator interfaces
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)
- 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.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):
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):
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
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.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.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.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):
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):
self.xconnect(self.sub_if.sw_if_index, self.pg1.sw_if_index)
# Create incoming packet streams for packet-generator interfaces
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)
- 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.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.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.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):
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):
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(
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(
- 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.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.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.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(
self.bridge(self.pg2.sw_if_index, is_add=0)
self.vapi.sw_interface_set_l2_bridge(
- 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
+ )