import unittest
from scapy.packet import Raw
-from scapy.layers.l2 import Ether, Dot1Q, GRE
+from scapy.layers.l2 import Ether, Dot1Q, 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 vpp_sub_interface import VppDot1QSubint, VppDot1ADSubint
-from vpp_gre_interface import VppGreInterface, VppGre6Interface
-from vpp_papi_provider import L2_VTR_OP
+from vpp_sub_interface import L2_VTR_OP, VppDot1QSubint, VppDot1ADSubint
+from vpp_gre_interface import VppGreInterface
from collections import namedtuple
+from vpp_papi import VppEnum
+
Tag = namedtuple('Tag', ['dot1', 'vlan'])
DOT1AD = 0x88A8
def setUpClass(cls):
super(TestSpan, cls).setUpClass()
# Test variables
- cls.hosts_nr = 10 # Number of hosts
cls.pkts_per_burst = 257 # Number of packets per burst
# create 3 pg interfaces
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)
+ cls.vlan_sub_if = VppDot1QSubint(cls, cls.pg2, 300)
+ cls.vlan_sub_if.set_vtr(L2_VTR_OP.L2_POP_1, tag=300)
+
+ cls.qinq_sub_if = VppDot1ADSubint(cls, cls.pg2, 33, 400, 500)
+ cls.qinq_sub_if.set_vtr(L2_VTR_OP.L2_POP_2, outer=500, inner=400)
+
# packet flows mapping pg0 -> pg1, pg2 -> pg3, etc.
cls.flows = dict()
cls.flows[cls.pg0] = [cls.pg1]
+ cls.flows[cls.pg1] = [cls.pg0]
# packet sizes
- cls.pg_if_packet_sizes = [64, 512] # , 1518, 9018]
-
- cls.interfaces = list(cls.pg_interfaces)
-
- # Create host MAC and IPv4 lists
- # cls.MY_MACS = dict()
- # cls.MY_IP4S = dict()
- cls.create_host_lists(cls.hosts_nr)
+ cls.pg_if_packet_sizes = [64, 512, 1518] # , 9018]
# setup all interfaces
- for i in cls.interfaces:
+ for i in cls.pg_interfaces:
i.admin_up()
i.config_ip4()
i.resolve_arp()
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)
+ src_address=cls.pg2.local_ip4n, dst_address=cls.pg2.remote_ip4n,
+ is_add=1, vni=1111)
def setUp(self):
super(TestSpan, self).setUp()
def tearDown(self):
super(TestSpan, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.ppcli("show interface span"))
+
+ def show_commands_at_teardown(self):
+ 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)
+ 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)
return pkt[GRE].payload
+ def decap_erspan(self, pkt, session):
+ """
+ Decapsulate the original payload frame by removing ERSPAN 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)
+
+ self.assertEqual(pkt[ERSPAN].ver, 1)
+ self.assertEqual(pkt[ERSPAN].vlan, 0)
+ self.assertEqual(pkt[ERSPAN].cos, 0)
+ self.assertEqual(pkt[ERSPAN].en, 3)
+ self.assertEqual(pkt[ERSPAN].t, 0)
+ self.assertEqual(pkt[ERSPAN].session_id, session)
+ self.assertEqual(pkt[ERSPAN].reserved, 0)
+ self.assertEqual(pkt[ERSPAN].index, 0)
+
+ return pkt[ERSPAN].payload
+
def decap_vxlan(self, pkt):
"""
Decapsulate the original payload frame by removing VXLAN header
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
- interfaces. Create host IPv4 address for every host MAC address too.
-
- :param count: Number of hosts to create MAC and IPv4 addresses for.
- """
- # mapping between packet-generator index and lists of test hosts
- self.hosts_by_pg_idx = dict()
-
- for pg_if in self.pg_interfaces:
- # self.MY_MACS[i.sw_if_index] = []
- # self.MY_IP4S[i.sw_if_index] = []
- self.hosts_by_pg_idx[pg_if.sw_if_index] = []
- hosts = self.hosts_by_pg_idx[pg_if.sw_if_index]
- for j in range(0, count):
- host = Host(
- "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j),
- "172.17.1%02x.%u" % (pg_if.sw_if_index, j))
- hosts.append(host)
-
- def create_stream(self, src_if, packet_sizes, do_dot1=False):
+ def create_stream(self, src_if, packet_sizes, do_dot1=False, bcast=False):
pkts = []
+ dst_if = self.flows[src_if][0]
+ dst_mac = src_if.remote_mac
+ if bcast:
+ dst_mac = "ff:ff:ff:ff:ff:ff"
+
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)
- p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+ payload = "span test"
+ 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=1234, dport=1234) /
+ 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)
- pkt_info.data = p.copy()
- size = packet_sizes[(i / 2) % len(packet_sizes)]
self.extend_packet(p, size)
pkts.append(p)
return pkts
- def verify_capture(self, dst_if, capture_pg1, capture_pg2):
- last_info = dict()
- for i in self.interfaces:
- last_info[i.sw_if_index] = None
- dst_sw_if_index = dst_if.sw_if_index
- self.assertEqual(
- len(capture_pg1),
- len(capture_pg2),
- "Different number of outgoing and mirrored packets : %u != %u" %
- (len(capture_pg1),
- len(capture_pg2)))
- for pkt_pg1, pkt_pg2 in zip(capture_pg1, capture_pg2):
- try:
- ip1 = pkt_pg1[IP]
- udp1 = pkt_pg1[UDP]
- raw1 = pkt_pg1[Raw]
-
- if pkt_pg1[Ether] != pkt_pg2[Ether]:
- self.logger.error("Different ethernet header of "
- "outgoing and mirrored packet")
- raise
- if ip1 != pkt_pg2[IP]:
- self.logger.error(
- "Different ip header of outgoing and mirrored packet")
- raise
- if udp1 != pkt_pg2[UDP]:
- self.logger.error(
- "Different udp header of outgoing and mirrored packet")
- raise
- if raw1 != pkt_pg2[Raw]:
- self.logger.error(
- "Different raw data of outgoing and mirrored packet")
- raise
-
- payload_info = self.payload_to_info(str(raw1))
- packet_index = payload_info.index
- self.assertEqual(payload_info.dst, dst_sw_if_index)
- self.logger.debug(
- "Got packet on port %s: src=%u (id=%u)" %
- (dst_if.name, payload_info.src, packet_index))
- next_info = self.get_next_packet_info_for_interface2(
- payload_info.src, dst_sw_if_index,
- last_info[payload_info.src])
- last_info[payload_info.src] = 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(ip1.src, saved_packet[IP].src)
- self.assertEqual(ip1.dst, saved_packet[IP].dst)
- self.assertEqual(udp1.sport, saved_packet[UDP].sport)
- self.assertEqual(udp1.dport, saved_packet[UDP].dport)
- except:
- self.logger.error("Unexpected or invalid packets:")
- self.logger.error(ppp("pg1 packet:", pkt_pg1))
- self.logger.error(ppp("pg2 packet:", pkt_pg2))
- raise
- for i in self.interfaces:
- remaining_packet = self.get_next_packet_info_for_interface2(
- i, dst_sw_if_index, last_info[i.sw_if_index])
- self.assertTrue(remaining_packet is None,
- "Port %u: Packet expected from source %u didn't"
- " arrive" % (dst_sw_if_index, i.sw_if_index))
+ 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)))
+
+ 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 test
-
- Test scenario:
- 1. config
- 3 interfaces, pg0 l2xconnected with pg1
- 2. sending l2 eth packets between 2 interfaces (pg0, pg1) and
- mirrored to pg2
- 64B, 512B, 1518B, 9018B (ether_size)
- burst of packets per interface
- """
+ """ 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)
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)
- self.verify_capture(
- self.pg1,
- self.pg1.get_capture(),
- self.pg2.get_capture(pg2_expected))
+ n_pkts = len(pkts)
+ pg1_pkts = self.pg1.get_capture(n_pkts)
+ pg2_pkts = self.pg2.get_capture(n_pkts)
# 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)
+ self.verify_capture(pg1_pkts, pg2_pkts)
+
def test_span_l2_rx(self):
- """ SPAN l2 rx mirror test """
+ """ SPAN l2 rx mirror """
self.sub_if.admin_up()
self.bridge(self.pg2.sw_if_index)
- # Create bi-directional cross-connects between pg0 and pg1
+ # 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.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_expected = len(pkts)
+ pg1_pkts = self.pg1.get_capture(pg2_expected)
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)
+ self.verify_capture(pg1_pkts, pg2_pkts)
+
def test_span_l2_rx_dst_vxlan(self):
- """ SPAN l2 rx mirror into vxlan test """
+ """ SPAN l2 rx mirror into vxlan """
self.sub_if.admin_up()
self.vapi.sw_interface_set_flags(self.vxlan.sw_if_index,
- admin_up_down=1)
+ flags=1)
self.bridge(self.vxlan.sw_if_index, is_add=1)
- # Create bi-directional cross-connects between pg0 and pg1
+ # 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.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)
+ n_pkts = len(pkts)
+ pg1_pkts = self.pg1.get_capture(n_pkts)
+ pg2_pkts = [self.decap_vxlan(p) for p in self.pg2.get_capture(n_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)
+ self.verify_capture(pg1_pkts, pg2_pkts)
+
+ def test_span_l2_rx_dst_gre_erspan(self):
+ """ 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.add_vpp_config()
+ gre_if.admin_up()
+
+ self.bridge(gre_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)
+
+ # 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)
+
+ # 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)
+ n_pkts = len(pkts)
+ 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)
+ 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)
+ 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 """
gre_if = VppGreInterface(self, self.pg2.local_ip4,
self.pg2.remote_ip4,
- is_teb=1)
+ type=(VppEnum.vl_api_gre_tunnel_type_t.
+ GRE_API_TUNNEL_TYPE_TEB))
gre_if.add_vpp_config()
gre_if.admin_up()
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.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)
+ n_pkts = len(pkts)
+ 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)])
+ 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)
- def test_span_l2_rx_dst_vtr(self):
- """ SPAN l2 rx mirror into subif+vtr """
+ self.verify_capture(pg1_pkts, pg2_decaped)
+
+ def test_span_l2_rx_dst_1q_vtr(self):
+ """ SPAN l2 rx mirror into 1q subif+vtr """
self.sub_if.admin_up()
- self.dst_sub_if.admin_up()
+ self.vlan_sub_if.admin_up()
- self.bridge(self.dst_sub_if.sw_if_index)
+ self.bridge(self.vlan_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)
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)
+ 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)
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)
+ 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]
- self.verify_capture(
- self.pg1,
- pg1_pkts,
- pg2_untagged)
- self.bridge(self.dst_sub_if.sw_if_index, is_add=0)
+ 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.dst_sub_if.sw_if_index, state=0,
+ 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 """
+
+ self.sub_if.admin_up()
+ self.qinq_sub_if.admin_up()
+
+ self.bridge(self.qinq_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.qinq_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)
+ 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]
+
+ 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.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 test """
+ """ SPAN l2 tx mirror """
self.sub_if.admin_up()
self.bridge(self.pg2.sw_if_index)
self.pg0, self.pg_if_packet_sizes, do_dot1=True)
self.pg0.add_stream(pkts)
- # Enable SPAN on pg0 (mirrored to pg2)
+ # 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.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()
+ n_pkts = len(pkts)
+ pg1_pkts = self.pg1.get_capture(n_pkts)
+ pg2_pkts = self.pg2.get_capture(n_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)
+
+ self.verify_capture(pg1_pkts, pg2_pkts)
+
+ def test_l2_rx_tx_span(self):
+ """ SPAN l2 rx 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
+ 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)
+ 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.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)
+ pg0_expected = len(pg1_pkts)
+ pg1_expected = len(pg0_pkts)
+ pg2_expected = pg0_expected + pg1_expected
+
+ pg0_pkts = self.pg0.get_capture(pg0_expected)
+ pg1_pkts = self.pg1.get_capture(pg1_expected)
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.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 """
+
+ 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)
+ self.vapi.sw_interface_set_l2_bridge(
+ 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.add_stream(pg0_pkts)
+ pg1_pkts = self.create_stream(
+ 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.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)
+ pg0_expected = len(pg1_pkts)
+ pg1_expected = len(pg0_pkts)
+ pg2_expected = pg0_expected + pg1_expected
+
+ pg0_pkts = self.pg0.get_capture(pg0_expected)
+ pg1_pkts = self.pg1.get_capture(pg1_expected)
+ pg2_pkts = self.pg2.get_capture(pg2_expected)
+
+ 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)
+ self.vapi.sw_interface_set_l2_bridge(
+ 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.verify_capture(pg0_pkts + pg1_pkts, pg2_pkts)
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff