cf9356d642
- Required for Ubuntu 24.04 LTS jobs - temporarily disable TestIpsecEsp1 and TestIpsecAhAll tests until a patch can be added to fix them Type: test Change-Id: I1ae7b170117182c3252629bbbb770775e2c496c9 Signed-off-by: Benoît Ganne <bganne@cisco.com> Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
610 lines
22 KiB
Python
610 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import unittest
|
|
|
|
from scapy.packet import Raw
|
|
from scapy.layers.l2 import Ether, GRE
|
|
from scapy.contrib.erspan import ERSPAN
|
|
from scapy.layers.inet import IP, UDP
|
|
from scapy.layers.vxlan import VXLAN
|
|
|
|
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
|
|
from collections import namedtuple
|
|
from vpp_papi import VppEnum
|
|
from config import config
|
|
|
|
|
|
Tag = namedtuple("Tag", ["dot1", "vlan"])
|
|
DOT1AD = 0x88A8
|
|
DOT1Q = 0x8100
|
|
|
|
|
|
@unittest.skipIf(
|
|
"vxlan" in config.excluded_plugins, "Exclude tests requiring VXLAN plugin"
|
|
)
|
|
class TestSpan(VppTestCase):
|
|
"""SPAN Test Case"""
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(TestSpan, cls).setUpClass()
|
|
# Test variables
|
|
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.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]
|
|
|
|
# setup all interfaces
|
|
for i in cls.pg_interfaces:
|
|
i.admin_up()
|
|
i.config_ip4()
|
|
i.resolve_arp()
|
|
|
|
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.add_vpp_config()
|
|
self.reset_packet_infos()
|
|
|
|
def tearDown(self):
|
|
super(TestSpan, self).tearDown()
|
|
|
|
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(
|
|
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)
|
|
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_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
|
|
"""
|
|
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
|
|
|
|
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):
|
|
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)
|
|
)
|
|
if do_dot1:
|
|
p = self.sub_if.add_dot1_layer(p)
|
|
self.extend_packet(p, size)
|
|
pkts.append(p)
|
|
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)),
|
|
)
|
|
|
|
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"""
|
|
|
|
# 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()
|
|
|
|
# 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)
|
|
|
|
# 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"""
|
|
|
|
self.sub_if.admin_up()
|
|
|
|
self.bridge(self.pg2.sw_if_index)
|
|
# 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)
|
|
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)
|
|
pg2_expected = len(pkts)
|
|
pg1_pkts = self.pg1.get_capture(pg2_expected)
|
|
pg2_pkts = self.pg2.get_capture(pg2_expected)
|
|
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"""
|
|
|
|
self.sub_if.admin_up()
|
|
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)
|
|
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)
|
|
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"""
|
|
|
|
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.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)
|
|
|
|
# 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
|
|
)
|
|
|
|
# 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.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
|
|
)
|
|
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"""
|
|
|
|
self.sub_if.admin_up()
|
|
self.vlan_sub_if.admin_up()
|
|
|
|
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)
|
|
|
|
# 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.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)
|
|
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.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.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"""
|
|
|
|
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 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.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)
|
|
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.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(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)
|