from framework import tag_fixme_vpp_workers
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto, INVALID_INDEX
-from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsRoute, \
- VppMplsIpBind, VppIpMRoute, VppMRoutePath, \
- VppIpTable, VppMplsTable, \
- VppMplsLabel, MplsLspMode, find_mpls_route, \
- FibPathProto, FibPathType, FibPathFlags, VppMplsLabel, MplsLspMode
+from vpp_ip_route import (
+ VppIpRoute,
+ VppRoutePath,
+ VppMplsRoute,
+ VppMplsIpBind,
+ VppIpMRoute,
+ VppMRoutePath,
+ VppIpTable,
+ VppMplsTable,
+ VppMplsLabel,
+ MplsLspMode,
+ find_mpls_route,
+ FibPathProto,
+ FibPathType,
+ FibPathFlags,
+ VppMplsLabel,
+ MplsLspMode,
+)
from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface
from vpp_papi import VppEnum
from scapy.packet import Raw
from scapy.layers.l2 import Ether, ARP
from scapy.layers.inet import IP, UDP, ICMP, icmptypes, icmpcodes
-from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, ICMPv6EchoRequest, \
- ICMPv6PacketTooBig
+from scapy.layers.inet6 import (
+ IPv6,
+ ICMPv6TimeExceeded,
+ ICMPv6EchoRequest,
+ ICMPv6PacketTooBig,
+)
from scapy.contrib.mpls import MPLS
NUM_PKTS = 67
@tag_fixme_vpp_workers
class TestMPLS(VppTestCase):
- """ MPLS Test Case """
+ """MPLS Test Case"""
@classmethod
def setUpClass(cls):
# the default of 64 matches the IP packet TTL default
def create_stream_labelled_ip4(
- self,
- src_if,
- mpls_labels,
- ping=0,
- ip_itf=None,
- dst_ip=None,
- chksum=None,
- ip_ttl=64,
- n=257):
+ self,
+ src_if,
+ mpls_labels,
+ ping=0,
+ ip_itf=None,
+ dst_ip=None,
+ chksum=None,
+ ip_ttl=64,
+ n=257,
+ ):
self.reset_packet_infos()
pkts = []
for i in range(0, n):
p = Ether(dst=src_if.local_mac, src=src_if.remote_mac)
for ii in range(len(mpls_labels)):
- p = p / MPLS(label=mpls_labels[ii].value,
- ttl=mpls_labels[ii].ttl,
- cos=mpls_labels[ii].exp)
+ p = p / MPLS(
+ label=mpls_labels[ii].value,
+ ttl=mpls_labels[ii].ttl,
+ cos=mpls_labels[ii].exp,
+ )
if not ping:
if not dst_ip:
- p = (p / IP(src=src_if.local_ip4,
- dst=src_if.remote_ip4,
- ttl=ip_ttl) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = (
+ p
+ / IP(src=src_if.local_ip4, dst=src_if.remote_ip4, ttl=ip_ttl)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
else:
- p = (p / IP(src=src_if.local_ip4, dst=dst_ip, ttl=ip_ttl) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = (
+ p
+ / IP(src=src_if.local_ip4, dst=dst_ip, ttl=ip_ttl)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
else:
- p = (p / IP(src=ip_itf.remote_ip4,
- dst=ip_itf.local_ip4,
- ttl=ip_ttl) /
- ICMP())
+ p = (
+ p
+ / IP(src=ip_itf.remote_ip4, dst=ip_itf.local_ip4, ttl=ip_ttl)
+ / ICMP()
+ )
if chksum:
p[IP].chksum = chksum
pkts.append(p)
return pkts
- def create_stream_ip4(self, src_if, dst_ip, ip_ttl=64,
- ip_dscp=0, payload_size=None):
+ def create_stream_ip4(
+ self, src_if, dst_ip, ip_ttl=64, ip_dscp=0, payload_size=None
+ ):
self.reset_packet_infos()
pkts = []
for i in range(0, 257):
info = self.create_packet_info(src_if, src_if)
payload = self.info_to_payload(info)
- p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- IP(src=src_if.remote_ip4, dst=dst_ip,
- ttl=ip_ttl, tos=ip_dscp) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = (
+ Ether(dst=src_if.local_mac, src=src_if.remote_mac)
+ / IP(src=src_if.remote_ip4, dst=dst_ip, ttl=ip_ttl, tos=ip_dscp)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
info.data = p.copy()
if payload_size:
self.extend_packet(p, payload_size)
for i in range(0, 257):
info = self.create_packet_info(src_if, src_if)
payload = self.info_to_payload(info)
- p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
- IPv6(src=src_if.remote_ip6, dst=dst_ip,
- hlim=ip_ttl, tc=ip_dscp) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = (
+ Ether(dst=src_if.local_mac, src=src_if.remote_mac)
+ / IPv6(src=src_if.remote_ip6, dst=dst_ip, hlim=ip_ttl, tc=ip_dscp)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
info.data = p.copy()
pkts.append(p)
return pkts
- def create_stream_labelled_ip6(self, src_if, mpls_labels,
- hlim=64, dst_ip=None,
- ping=0, ip_itf=None):
+ def create_stream_labelled_ip6(
+ self, src_if, mpls_labels, hlim=64, dst_ip=None, ping=0, ip_itf=None
+ ):
if dst_ip is None:
dst_ip = src_if.remote_ip6
self.reset_packet_infos()
p = p / MPLS(label=l.value, ttl=l.ttl, cos=l.exp)
if ping:
- p = p / (IPv6(src=ip_itf.remote_ip6,
- dst=ip_itf.local_ip6) /
- ICMPv6EchoRequest())
+ p = p / (
+ IPv6(src=ip_itf.remote_ip6, dst=ip_itf.local_ip6)
+ / ICMPv6EchoRequest()
+ )
else:
- p = p / (IPv6(src=src_if.remote_ip6, dst=dst_ip, hlim=hlim) /
- UDP(sport=1234, dport=1234) /
- Raw(payload))
+ p = p / (
+ IPv6(src=src_if.remote_ip6, dst=dst_ip, hlim=hlim)
+ / UDP(sport=1234, dport=1234)
+ / Raw(payload)
+ )
info.data = p.copy()
pkts.append(p)
return pkts
- def verify_capture_ip4(self, src_if, capture, sent, ping_resp=0,
- ip_ttl=None, ip_dscp=0):
+ def verify_capture_ip4(
+ self, src_if, capture, sent, ping_resp=0, ip_ttl=None, ip_dscp=0
+ ):
try:
capture = verify_filter(capture, sent)
except:
raise
- def verify_capture_labelled_ip4(self, src_if, capture, sent,
- mpls_labels, ip_ttl=None):
+ def verify_capture_labelled_ip4(
+ self, src_if, capture, sent, mpls_labels, ip_ttl=None
+ ):
try:
capture = verify_filter(capture, sent)
except:
raise
- def verify_capture_labelled_ip6(self, src_if, capture, sent,
- mpls_labels, ip_ttl=None):
+ def verify_capture_labelled_ip6(
+ self, src_if, capture, sent, mpls_labels, ip_ttl=None
+ ):
try:
capture = verify_filter(capture, sent)
except:
raise
- def verify_capture_labelled(self, src_if, capture, sent,
- mpls_labels):
+ def verify_capture_labelled(self, src_if, capture, sent, mpls_labels):
try:
capture = verify_filter(capture, sent)
except:
raise
- def verify_capture_ip6(self, src_if, capture, sent,
- ip_hlim=None, ip_dscp=0,
- ping_resp=0):
+ def verify_capture_ip6(
+ self, src_if, capture, sent, ip_hlim=None, ip_dscp=0, ping_resp=0
+ ):
try:
self.assertEqual(len(capture), len(sent))
if not ping_resp:
self.assertEqual(rx_ip.src, tx_ip.src)
self.assertEqual(rx_ip.dst, tx_ip.dst)
- self.assertEqual(rx_ip.tc, ip_dscp)
+ self.assertEqual(rx_ip.tc, ip_dscp)
# IP processing post pop has decremented the TTL
if not ip_hlim:
self.assertEqual(rx_ip.hlim + 1, tx_ip.hlim)
except:
raise
- def verify_capture_fragmented_labelled_ip4(self, src_if, capture, sent,
- mpls_labels, ip_ttl=None):
+ def verify_capture_fragmented_labelled_ip4(
+ self, src_if, capture, sent, mpls_labels, ip_ttl=None
+ ):
try:
capture = verify_filter(capture, sent)
except:
raise
- def verify_capture_fragmented_labelled_ip6(self, src_if, capture, sent,
- mpls_labels, ip_ttl=None):
+ def verify_capture_fragmented_labelled_ip6(
+ self, src_if, capture, sent, mpls_labels, ip_ttl=None
+ ):
try:
capture = verify_filter(capture, sent)
raise
def test_swap(self):
- """ MPLS label swap tests """
+ """MPLS label swap tests"""
#
# A simple MPLS xconnect - eos label in label out
#
- route_32_eos = VppMplsRoute(self, 32, 1,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(33)])])
+ route_32_eos = VppMplsRoute(
+ self,
+ 32,
+ 1,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(33)]
+ )
+ ],
+ )
route_32_eos.add_vpp_config()
self.assertTrue(
- find_mpls_route(self, 0, 32, 1,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(33)])]))
+ find_mpls_route(
+ self,
+ 0,
+ 32,
+ 1,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(33)],
+ )
+ ],
+ )
+ )
#
# a stream that matches the route for 10.0.0.1
# PG0 is in the default table
#
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(32, ttl=32, exp=1)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(32, ttl=32, exp=1)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(33, ttl=31, exp=1)])
+ self.verify_capture_labelled(
+ self.pg0, rx, tx, [VppMplsLabel(33, ttl=31, exp=1)]
+ )
- self.assertEqual(route_32_eos.get_stats_to()['packets'], 257)
+ self.assertEqual(route_32_eos.get_stats_to()["packets"], 257)
#
# A simple MPLS xconnect - non-eos label in label out
#
- route_32_neos = VppMplsRoute(self, 32, 0,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(33)])])
+ route_32_neos = VppMplsRoute(
+ self,
+ 32,
+ 0,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(33)]
+ )
+ ],
+ )
route_32_neos.add_vpp_config()
#
# a stream that matches the route for 10.0.0.1
# PG0 is in the default table
#
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(32, ttl=21, exp=7),
- VppMplsLabel(99)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(32, ttl=21, exp=7), VppMplsLabel(99)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(33, ttl=20, exp=7),
- VppMplsLabel(99)])
- self.assertEqual(route_32_neos.get_stats_to()['packets'], 257)
+ self.verify_capture_labelled(
+ self.pg0, rx, tx, [VppMplsLabel(33, ttl=20, exp=7), VppMplsLabel(99)]
+ )
+ self.assertEqual(route_32_neos.get_stats_to()["packets"], 257)
#
# A simple MPLS xconnect - non-eos label in label out, uniform mode
#
route_42_neos = VppMplsRoute(
- self, 42, 0,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(43, MplsLspMode.UNIFORM)])])
+ self,
+ 42,
+ 0,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(43, MplsLspMode.UNIFORM)],
+ )
+ ],
+ )
route_42_neos.add_vpp_config()
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(42, ttl=21, exp=7),
- VppMplsLabel(99)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(42, ttl=21, exp=7), VppMplsLabel(99)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(43, ttl=20, exp=7),
- VppMplsLabel(99)])
+ self.verify_capture_labelled(
+ self.pg0, rx, tx, [VppMplsLabel(43, ttl=20, exp=7), VppMplsLabel(99)]
+ )
#
# An MPLS xconnect - EOS label in IP out
#
- route_33_eos = VppMplsRoute(self, 33, 1,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[])])
+ route_33_eos = VppMplsRoute(
+ self,
+ 33,
+ 1,
+ [VppRoutePath(self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[])],
+ )
route_33_eos.add_vpp_config()
tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(33)])
#
# disposed packets have an invalid IPv4 checksum
#
- tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(33)],
- dst_ip=self.pg0.remote_ip4,
- n=65,
- chksum=1)
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(33)], dst_ip=self.pg0.remote_ip4, n=65, chksum=1
+ )
self.send_and_assert_no_replies(self.pg0, tx, "Invalid Checksum")
#
# An MPLS xconnect - EOS label in IP out, uniform mode
#
route_3333_eos = VppMplsRoute(
- self, 3333, 1,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(3, MplsLspMode.UNIFORM)])])
+ self,
+ 3333,
+ 1,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(3, MplsLspMode.UNIFORM)],
+ )
+ ],
+ )
route_3333_eos.add_vpp_config()
tx = self.create_stream_labelled_ip4(
- self.pg0,
- [VppMplsLabel(3333, ttl=55, exp=3)])
+ self.pg0, [VppMplsLabel(3333, ttl=55, exp=3)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
self.verify_capture_ip4(self.pg0, rx, tx, ip_ttl=54, ip_dscp=0x60)
tx = self.create_stream_labelled_ip4(
- self.pg0,
- [VppMplsLabel(3333, ttl=66, exp=4)])
+ self.pg0, [VppMplsLabel(3333, ttl=66, exp=4)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
self.verify_capture_ip4(self.pg0, rx, tx, ip_ttl=65, ip_dscp=0x80)
# An MPLS xconnect - EOS label in IPv6 out
#
route_333_eos = VppMplsRoute(
- self, 333, 1,
- [VppRoutePath(self.pg0.remote_ip6,
- self.pg0.sw_if_index,
- labels=[])],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)
+ self,
+ 333,
+ 1,
+ [VppRoutePath(self.pg0.remote_ip6, self.pg0.sw_if_index, labels=[])],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
route_333_eos.add_vpp_config()
tx = self.create_stream_labelled_ip6(self.pg0, [VppMplsLabel(333)])
#
# disposed packets have an TTL expired
#
- tx = self.create_stream_labelled_ip6(self.pg0,
- [VppMplsLabel(333, ttl=64)],
- dst_ip=self.pg1.remote_ip6,
- hlim=1)
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(333, ttl=64)], dst_ip=self.pg1.remote_ip6, hlim=1
+ )
rx = self.send_and_expect_some(self.pg0, tx, self.pg0)
self.verify_capture_ip6_icmp(self.pg0, rx, tx)
# An MPLS xconnect - EOS label in IPv6 out w imp-null
#
route_334_eos = VppMplsRoute(
- self, 334, 1,
- [VppRoutePath(self.pg0.remote_ip6,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(3)])],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)
+ self,
+ 334,
+ 1,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip6, self.pg0.sw_if_index, labels=[VppMplsLabel(3)]
+ )
+ ],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
route_334_eos.add_vpp_config()
- tx = self.create_stream_labelled_ip6(self.pg0,
- [VppMplsLabel(334, ttl=64)])
+ tx = self.create_stream_labelled_ip6(self.pg0, [VppMplsLabel(334, ttl=64)])
rx = self.send_and_expect(self.pg0, tx, self.pg0)
self.verify_capture_ip6(self.pg0, rx, tx)
# An MPLS xconnect - EOS label in IPv6 out w imp-null in uniform mode
#
route_335_eos = VppMplsRoute(
- self, 335, 1,
- [VppRoutePath(self.pg0.remote_ip6,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(3, MplsLspMode.UNIFORM)])],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)
+ self,
+ 335,
+ 1,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip6,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(3, MplsLspMode.UNIFORM)],
+ )
+ ],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
route_335_eos.add_vpp_config()
tx = self.create_stream_labelled_ip6(
- self.pg0,
- [VppMplsLabel(335, ttl=27, exp=4)])
+ self.pg0, [VppMplsLabel(335, ttl=27, exp=4)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
self.verify_capture_ip6(self.pg0, rx, tx, ip_hlim=26, ip_dscp=0x80)
#
# disposed packets have an TTL expired
#
- tx = self.create_stream_labelled_ip6(self.pg0, [VppMplsLabel(334)],
- dst_ip=self.pg1.remote_ip6,
- hlim=0)
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(334)], dst_ip=self.pg1.remote_ip6, hlim=0
+ )
rx = self.send_and_expect_some(self.pg0, tx, self.pg0)
self.verify_capture_ip6_icmp(self.pg0, rx, tx)
# An MPLS xconnect - non-EOS label in IP out - an invalid configuration
# so this traffic should be dropped.
#
- route_33_neos = VppMplsRoute(self, 33, 0,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[])])
+ route_33_neos = VppMplsRoute(
+ self,
+ 33,
+ 0,
+ [VppRoutePath(self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[])],
+ )
route_33_neos.add_vpp_config()
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(33),
- VppMplsLabel(99)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(33), VppMplsLabel(99)]
+ )
self.send_and_assert_no_replies(
- self.pg0, tx,
- "MPLS non-EOS packets popped and forwarded")
+ self.pg0, tx, "MPLS non-EOS packets popped and forwarded"
+ )
#
# A recursive EOS x-connect, which resolves through another x-connect
# in pipe mode
#
- route_34_eos = VppMplsRoute(self, 34, 1,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_via_label=32,
- labels=[VppMplsLabel(44),
- VppMplsLabel(45)])])
+ route_34_eos = VppMplsRoute(
+ self,
+ 34,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ 0xFFFFFFFF,
+ nh_via_label=32,
+ labels=[VppMplsLabel(44), VppMplsLabel(45)],
+ )
+ ],
+ )
route_34_eos.add_vpp_config()
self.logger.info(self.vapi.cli("sh mpls fib 34"))
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(34, ttl=3)])
+ tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(34, ttl=3)])
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(33),
- VppMplsLabel(44),
- VppMplsLabel(45, ttl=2)])
+ self.verify_capture_labelled(
+ self.pg0,
+ rx,
+ tx,
+ [VppMplsLabel(33), VppMplsLabel(44), VppMplsLabel(45, ttl=2)],
+ )
- self.assertEqual(route_34_eos.get_stats_to()['packets'], 257)
- self.assertEqual(route_32_neos.get_stats_via()['packets'], 257)
+ self.assertEqual(route_34_eos.get_stats_to()["packets"], 257)
+ self.assertEqual(route_32_neos.get_stats_via()["packets"], 257)
#
# A recursive EOS x-connect, which resolves through another x-connect
# in uniform mode
#
route_35_eos = VppMplsRoute(
- self, 35, 1,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_via_label=42,
- labels=[VppMplsLabel(44)])])
+ self,
+ 35,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0", 0xFFFFFFFF, nh_via_label=42, labels=[VppMplsLabel(44)]
+ )
+ ],
+ )
route_35_eos.add_vpp_config()
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(35, ttl=3)])
+ tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(35, ttl=3)])
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(43, ttl=2),
- VppMplsLabel(44, ttl=2)])
+ self.verify_capture_labelled(
+ self.pg0, rx, tx, [VppMplsLabel(43, ttl=2), VppMplsLabel(44, ttl=2)]
+ )
#
# A recursive non-EOS x-connect, which resolves through another
# x-connect
#
- route_34_neos = VppMplsRoute(self, 34, 0,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_via_label=32,
- labels=[VppMplsLabel(44),
- VppMplsLabel(46)])])
+ route_34_neos = VppMplsRoute(
+ self,
+ 34,
+ 0,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ 0xFFFFFFFF,
+ nh_via_label=32,
+ labels=[VppMplsLabel(44), VppMplsLabel(46)],
+ )
+ ],
+ )
route_34_neos.add_vpp_config()
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(34, ttl=45),
- VppMplsLabel(99)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34, ttl=45), VppMplsLabel(99)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
# it's the 2nd (counting from 0) label in the stack that is swapped
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(33),
- VppMplsLabel(44),
- VppMplsLabel(46, ttl=44),
- VppMplsLabel(99)])
+ self.verify_capture_labelled(
+ self.pg0,
+ rx,
+ tx,
+ [
+ VppMplsLabel(33),
+ VppMplsLabel(44),
+ VppMplsLabel(46, ttl=44),
+ VppMplsLabel(99),
+ ],
+ )
#
# an recursive IP route that resolves through the recursive non-eos
# x-connect
#
- ip_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_via_label=34,
- labels=[VppMplsLabel(55)])])
+ ip_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ [
+ VppRoutePath(
+ "0.0.0.0", 0xFFFFFFFF, nh_via_label=34, labels=[VppMplsLabel(55)]
+ )
+ ],
+ )
ip_10_0_0_1.add_vpp_config()
tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(33),
- VppMplsLabel(44),
- VppMplsLabel(46),
- VppMplsLabel(55)])
- self.assertEqual(ip_10_0_0_1.get_stats_to()['packets'], 257)
+ self.verify_capture_labelled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [VppMplsLabel(33), VppMplsLabel(44), VppMplsLabel(46), VppMplsLabel(55)],
+ )
+ self.assertEqual(ip_10_0_0_1.get_stats_to()["packets"], 257)
ip_10_0_0_1.remove_vpp_config()
route_34_neos.remove_vpp_config()
route_32_eos.remove_vpp_config()
def test_bind(self):
- """ MPLS Local Label Binding test """
+ """MPLS Local Label Binding test"""
#
# Add a non-recursive route with a single out label
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(45)])])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(45)]
+ )
+ ],
+ )
route_10_0_0_1.add_vpp_config()
# bind a local label to the route
binding.add_vpp_config()
# non-EOS stream
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(44),
- VppMplsLabel(99)])
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(44), VppMplsLabel(99)]
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(45, ttl=63),
- VppMplsLabel(99)])
+ self.verify_capture_labelled(
+ self.pg0, rx, tx, [VppMplsLabel(45, ttl=63), VppMplsLabel(99)]
+ )
# EOS stream
tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(44)])
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled(self.pg0, rx, tx,
- [VppMplsLabel(45, ttl=63)])
+ self.verify_capture_labelled(self.pg0, rx, tx, [VppMplsLabel(45, ttl=63)])
# IP stream
tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
route_10_0_0_1.remove_vpp_config()
def test_imposition(self):
- """ MPLS label imposition test """
+ """MPLS label imposition test"""
#
# Add a non-recursive route with a single out label
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32)])])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(32)]
+ )
+ ],
+ )
route_10_0_0_1.add_vpp_config()
#
#
# Add a non-recursive route with a 3 out labels
#
- route_10_0_0_2 = VppIpRoute(self, "10.0.0.2", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32),
- VppMplsLabel(33),
- VppMplsLabel(34)])])
+ route_10_0_0_2 = VppIpRoute(
+ self,
+ "10.0.0.2",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34)],
+ )
+ ],
+ )
route_10_0_0_2.add_vpp_config()
- tx = self.create_stream_ip4(self.pg0, "10.0.0.2",
- ip_ttl=44, ip_dscp=0xff)
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.2", ip_ttl=44, ip_dscp=0xFF)
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32),
- VppMplsLabel(33),
- VppMplsLabel(34)],
- ip_ttl=43)
+ self.verify_capture_labelled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [VppMplsLabel(32), VppMplsLabel(33), VppMplsLabel(34)],
+ ip_ttl=43,
+ )
#
# Add a non-recursive route with a single out label in uniform mode
#
route_10_0_0_3 = VppIpRoute(
- self, "10.0.0.3", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32,
- mode=MplsLspMode.UNIFORM)])])
+ self,
+ "10.0.0.3",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(32, mode=MplsLspMode.UNIFORM)],
+ )
+ ],
+ )
route_10_0_0_3.add_vpp_config()
- tx = self.create_stream_ip4(self.pg0, "10.0.0.3",
- ip_ttl=54, ip_dscp=0xbe)
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.3", ip_ttl=54, ip_dscp=0xBE)
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32, ttl=53, exp=5)])
+ self.verify_capture_labelled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(32, ttl=53, exp=5)]
+ )
#
# Add a IPv6 non-recursive route with a single out label in
# uniform mode
#
route_2001_3 = VppIpRoute(
- self, "2001::3", 128,
- [VppRoutePath(self.pg0.remote_ip6,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32,
- mode=MplsLspMode.UNIFORM)])])
+ self,
+ "2001::3",
+ 128,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip6,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(32, mode=MplsLspMode.UNIFORM)],
+ )
+ ],
+ )
route_2001_3.add_vpp_config()
- tx = self.create_stream_ip6(self.pg0, "2001::3",
- ip_ttl=54, ip_dscp=0xbe)
+ tx = self.create_stream_ip6(self.pg0, "2001::3", ip_ttl=54, ip_dscp=0xBE)
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip6(self.pg0, rx, tx,
- [VppMplsLabel(32, ttl=53, exp=5)])
+ self.verify_capture_labelled_ip6(
+ self.pg0, rx, tx, [VppMplsLabel(32, ttl=53, exp=5)]
+ )
#
# add a recursive path, with output label, via the 1 label route
#
- route_11_0_0_1 = VppIpRoute(self, "11.0.0.1", 32,
- [VppRoutePath("10.0.0.1",
- 0xffffffff,
- labels=[VppMplsLabel(44)])])
+ route_11_0_0_1 = VppIpRoute(
+ self,
+ "11.0.0.1",
+ 32,
+ [VppRoutePath("10.0.0.1", 0xFFFFFFFF, labels=[VppMplsLabel(44)])],
+ )
route_11_0_0_1.add_vpp_config()
#
#
tx = self.create_stream_ip4(self.pg0, "11.0.0.1")
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32),
- VppMplsLabel(44)])
+ self.verify_capture_labelled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(32), VppMplsLabel(44)]
+ )
- self.assertEqual(route_11_0_0_1.get_stats_to()['packets'], 257)
+ self.assertEqual(route_11_0_0_1.get_stats_to()["packets"], 257)
#
# add a recursive path, with 2 labels, via the 3 label route
#
- route_11_0_0_2 = VppIpRoute(self, "11.0.0.2", 32,
- [VppRoutePath("10.0.0.2",
- 0xffffffff,
- labels=[VppMplsLabel(44),
- VppMplsLabel(45)])])
+ route_11_0_0_2 = VppIpRoute(
+ self,
+ "11.0.0.2",
+ 32,
+ [
+ VppRoutePath(
+ "10.0.0.2", 0xFFFFFFFF, labels=[VppMplsLabel(44), VppMplsLabel(45)]
+ )
+ ],
+ )
route_11_0_0_2.add_vpp_config()
#
#
tx = self.create_stream_ip4(self.pg0, "11.0.0.2")
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32),
- VppMplsLabel(33),
- VppMplsLabel(34),
- VppMplsLabel(44),
- VppMplsLabel(45)])
-
- self.assertEqual(route_11_0_0_2.get_stats_to()['packets'], 257)
+ self.verify_capture_labelled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [
+ VppMplsLabel(32),
+ VppMplsLabel(33),
+ VppMplsLabel(34),
+ VppMplsLabel(44),
+ VppMplsLabel(45),
+ ],
+ )
+
+ self.assertEqual(route_11_0_0_2.get_stats_to()["packets"], 257)
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32),
- VppMplsLabel(33),
- VppMplsLabel(34),
- VppMplsLabel(44),
- VppMplsLabel(45)])
+ self.verify_capture_labelled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [
+ VppMplsLabel(32),
+ VppMplsLabel(33),
+ VppMplsLabel(34),
+ VppMplsLabel(44),
+ VppMplsLabel(45),
+ ],
+ )
- self.assertEqual(route_11_0_0_2.get_stats_to()['packets'], 514)
+ self.assertEqual(route_11_0_0_2.get_stats_to()["packets"], 514)
#
# cleanup
route_10_0_0_1.remove_vpp_config()
def test_imposition_fragmentation(self):
- """ MPLS label imposition fragmentation test """
+ """MPLS label imposition fragmentation test"""
#
# Add a ipv4 non-recursive route with a single out label
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32)])])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(32)]
+ )
+ ],
+ )
route_10_0_0_1.add_vpp_config()
- route_1000_1 = VppIpRoute(self, "1000::1", 128,
- [VppRoutePath(self.pg0.remote_ip6,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(32)])])
+ route_1000_1 = VppIpRoute(
+ self,
+ "1000::1",
+ 128,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip6, self.pg0.sw_if_index, labels=[VppMplsLabel(32)]
+ )
+ ],
+ )
route_1000_1.add_vpp_config()
#
# 5 fragments per packet (257*5=1285)
#
rx = self.send_and_expect(self.pg0, tx, self.pg0, 1285)
- self.verify_capture_fragmented_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(32)])
+ self.verify_capture_fragmented_labelled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(32)]
+ )
# packets with DF bit set generate ICMP
for t in tx:
- t[IP].flags = 'DF'
+ t[IP].flags = "DF"
rxs = self.send_and_expect_some(self.pg0, tx, self.pg0)
for rx in rxs:
self.assertEqual(icmptypes[rx[ICMP].type], "dest-unreach")
- self.assertEqual(icmpcodes[rx[ICMP].type][rx[ICMP].code],
- "fragmentation-needed")
+ self.assertEqual(
+ icmpcodes[rx[ICMP].type][rx[ICMP].code], "fragmentation-needed"
+ )
# the link MTU is 9000, the MPLS over head is 4 bytes
self.assertEqual(rx[ICMP].nexthopmtu, 9000 - 4)
- self.assertEqual(self.statistics.get_err_counter(
- "/err/mpls-frag/can't fragment this packet"),
- len(tx))
+ self.assertEqual(
+ self.statistics.get_err_counter(
+ "/err/mpls-frag/can't fragment this packet"
+ ),
+ len(tx),
+ )
#
# a stream that matches the route for 1000::1/128
# PG0 is in the default table
route_10_0_0_1.remove_vpp_config()
def test_tunnel_pipe(self):
- """ MPLS Tunnel Tests - Pipe """
+ """MPLS Tunnel Tests - Pipe"""
#
# Create a tunnel with two out labels
#
mpls_tun = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(44),
- VppMplsLabel(46)])])
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(44), VppMplsLabel(46)],
+ )
+ ],
+ )
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
#
# add an unlabelled route through the new tunnel
#
- route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
- [VppRoutePath("0.0.0.0",
- mpls_tun._sw_if_index)])
+ route_10_0_0_3 = VppIpRoute(
+ self, "10.0.0.3", 32, [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index)]
+ )
route_10_0_0_3.add_vpp_config()
self.vapi.cli("clear trace")
self.pg_start()
rx = self.pg0.get_capture()
- self.verify_capture_tunneled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(44),
- VppMplsLabel(46)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(44), VppMplsLabel(46)]
+ )
#
# add a labelled route through the new tunnel
#
- route_10_0_0_4 = VppIpRoute(self, "10.0.0.4", 32,
- [VppRoutePath("0.0.0.0",
- mpls_tun._sw_if_index,
- labels=[33])])
+ route_10_0_0_4 = VppIpRoute(
+ self,
+ "10.0.0.4",
+ 32,
+ [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index, labels=[33])],
+ )
route_10_0_0_4.add_vpp_config()
self.vapi.cli("clear trace")
self.pg_start()
rx = self.pg0.get_capture()
- self.verify_capture_tunneled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(44),
- VppMplsLabel(46),
- VppMplsLabel(33, ttl=255)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [VppMplsLabel(44), VppMplsLabel(46), VppMplsLabel(33, ttl=255)],
+ )
#
# change tunnel's MTU to a low value
mpls_tun.set_l3_mtu(1200)
# send IP into the tunnel to be fragmented
- tx = self.create_stream_ip4(self.pg0, "10.0.0.3",
- payload_size=1500)
- rx = self.send_and_expect(self.pg0, tx, self.pg0, len(tx)*2)
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.3", payload_size=1500)
+ rx = self.send_and_expect(self.pg0, tx, self.pg0, len(tx) * 2)
fake_tx = []
for p in tx:
fake_tx.append(p)
fake_tx.append(p)
- self.verify_capture_tunneled_ip4(self.pg0, rx, fake_tx,
- [VppMplsLabel(44),
- VppMplsLabel(46)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0, rx, fake_tx, [VppMplsLabel(44), VppMplsLabel(46)]
+ )
# send MPLS into the tunnel to be fragmented
- tx = self.create_stream_ip4(self.pg0, "10.0.0.4",
- payload_size=1500)
- rx = self.send_and_expect(self.pg0, tx, self.pg0, len(tx)*2)
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.4", payload_size=1500)
+ rx = self.send_and_expect(self.pg0, tx, self.pg0, len(tx) * 2)
fake_tx = []
for p in tx:
fake_tx.append(p)
fake_tx.append(p)
- self.verify_capture_tunneled_ip4(self.pg0, rx, fake_tx,
- [VppMplsLabel(44),
- VppMplsLabel(46),
- VppMplsLabel(33, ttl=255)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0,
+ rx,
+ fake_tx,
+ [VppMplsLabel(44), VppMplsLabel(46), VppMplsLabel(33, ttl=255)],
+ )
def test_tunnel_uniform(self):
- """ MPLS Tunnel Tests - Uniform """
+ """MPLS Tunnel Tests - Uniform"""
#
# Create a tunnel with a single out label
#
mpls_tun = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(44, ttl=32),
- VppMplsLabel(46, MplsLspMode.UNIFORM)])])
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[
+ VppMplsLabel(44, ttl=32),
+ VppMplsLabel(46, MplsLspMode.UNIFORM),
+ ],
+ )
+ ],
+ )
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
#
# add an unlabelled route through the new tunnel
#
- route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
- [VppRoutePath("0.0.0.0",
- mpls_tun._sw_if_index)])
+ route_10_0_0_3 = VppIpRoute(
+ self, "10.0.0.3", 32, [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index)]
+ )
route_10_0_0_3.add_vpp_config()
self.vapi.cli("clear trace")
self.pg_start()
rx = self.pg0.get_capture()
- self.verify_capture_tunneled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(44, ttl=32),
- VppMplsLabel(46, ttl=23)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(44, ttl=32), VppMplsLabel(46, ttl=23)]
+ )
#
# add a labelled route through the new tunnel
#
route_10_0_0_4 = VppIpRoute(
- self, "10.0.0.4", 32,
- [VppRoutePath("0.0.0.0",
- mpls_tun._sw_if_index,
- labels=[VppMplsLabel(33, ttl=47)])])
+ self,
+ "10.0.0.4",
+ 32,
+ [
+ VppRoutePath(
+ "0.0.0.0", mpls_tun._sw_if_index, labels=[VppMplsLabel(33, ttl=47)]
+ )
+ ],
+ )
route_10_0_0_4.add_vpp_config()
self.vapi.cli("clear trace")
self.pg_start()
rx = self.pg0.get_capture()
- self.verify_capture_tunneled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(44, ttl=32),
- VppMplsLabel(46, ttl=47),
- VppMplsLabel(33, ttl=47)])
+ self.verify_capture_tunneled_ip4(
+ self.pg0,
+ rx,
+ tx,
+ [
+ VppMplsLabel(44, ttl=32),
+ VppMplsLabel(46, ttl=47),
+ VppMplsLabel(33, ttl=47),
+ ],
+ )
def test_mpls_tunnel_many(self):
- """ MPLS Multiple Tunnels """
+ """MPLS Multiple Tunnels"""
for ii in range(100):
mpls_tun = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(44, ttl=32),
- VppMplsLabel(46, MplsLspMode.UNIFORM)])])
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[
+ VppMplsLabel(44, ttl=32),
+ VppMplsLabel(46, MplsLspMode.UNIFORM),
+ ],
+ )
+ ],
+ )
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
for ii in range(100):
mpls_tun = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(44, ttl=32),
- VppMplsLabel(46, MplsLspMode.UNIFORM)])],
- is_l2=1)
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[
+ VppMplsLabel(44, ttl=32),
+ VppMplsLabel(46, MplsLspMode.UNIFORM),
+ ],
+ )
+ ],
+ is_l2=1,
+ )
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
def test_v4_exp_null(self):
- """ MPLS V4 Explicit NULL test """
+ """MPLS V4 Explicit NULL test"""
#
# The first test case has an MPLS TTL of 0
# all packet should be dropped
#
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(0, ttl=0)])
- self.send_and_assert_no_replies(self.pg0, tx,
- "MPLS TTL=0 packets forwarded")
+ tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(0, ttl=0)])
+ self.send_and_assert_no_replies(self.pg0, tx, "MPLS TTL=0 packets forwarded")
#
# a stream with a non-zero MPLS TTL
self.verify_capture_ip4(self.pg1, rx, tx)
def test_v6_exp_null(self):
- """ MPLS V6 Explicit NULL test """
+ """MPLS V6 Explicit NULL test"""
#
# a stream with a non-zero MPLS TTL
self.verify_capture_ip6(self.pg0, rx, tx)
def test_deag(self):
- """ MPLS Deagg """
+ """MPLS Deagg"""
#
# A de-agg route - next-hop lookup in default table
#
- route_34_eos = VppMplsRoute(self, 34, 1,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_table_id=0)])
+ route_34_eos = VppMplsRoute(
+ self, 34, 1, [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=0)]
+ )
route_34_eos.add_vpp_config()
#
# ping an interface in the default table
# PG0 is in the default table
#
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(34)],
- ping=1,
- ip_itf=self.pg0)
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], ping=1, ip_itf=self.pg0
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg0)
self.verify_capture_ip4(self.pg0, rx, tx, ping_resp=1)
#
# A de-agg route - next-hop lookup in non-default table
#
- route_35_eos = VppMplsRoute(self, 35, 1,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_table_id=1)])
+ route_35_eos = VppMplsRoute(
+ self, 35, 1, [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1)]
+ )
route_35_eos.add_vpp_config()
route_356_eos = VppMplsRoute(
- self, 356, 1,
- [VppRoutePath("0::0",
- 0xffffffff,
- nh_table_id=1)],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)
+ self,
+ 356,
+ 1,
+ [VppRoutePath("0::0", 0xFFFFFFFF, nh_table_id=1)],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
route_356_eos.add_vpp_config()
#
# default table and egress unlabelled in the non-default
#
tx = self.create_stream_labelled_ip4(
- self.pg0, [VppMplsLabel(35)], ping=1, ip_itf=self.pg1)
+ self.pg0, [VppMplsLabel(35)], ping=1, ip_itf=self.pg1
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip4(self.pg1, rx, tx, ping_resp=1)
tx = self.create_stream_labelled_ip6(
- self.pg0, [VppMplsLabel(356)], ping=1, ip_itf=self.pg1)
+ self.pg0, [VppMplsLabel(356)], ping=1, ip_itf=self.pg1
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip6(self.pg1, rx, tx, ping_resp=1)
#
# Double pop
#
- route_36_neos = VppMplsRoute(self, 36, 0,
- [VppRoutePath("0.0.0.0",
- 0xffffffff)])
+ route_36_neos = VppMplsRoute(self, 36, 0, [VppRoutePath("0.0.0.0", 0xFFFFFFFF)])
route_36_neos.add_vpp_config()
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(36),
- VppMplsLabel(35)],
- ping=1, ip_itf=self.pg1)
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(36), VppMplsLabel(35)], ping=1, ip_itf=self.pg1
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip4(self.pg1, rx, tx, ping_resp=1)
route_34_eos.remove_vpp_config()
def test_interface_rx(self):
- """ MPLS Interface Receive """
+ """MPLS Interface Receive"""
#
# Add a non-recursive route that will forward the traffic
# post-interface-rx
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- table_id=1,
- paths=[VppRoutePath(self.pg1.remote_ip4,
- self.pg1.sw_if_index)])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ table_id=1,
+ paths=[VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
+ )
route_10_0_0_1.add_vpp_config()
#
# so as to have matched the route in table 1
#
route_34_eos = VppMplsRoute(
- self, 34, 1,
- [VppRoutePath("0.0.0.0",
- self.pg1.sw_if_index,
- type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX)])
+ self,
+ 34,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ self.pg1.sw_if_index,
+ type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
+ )
+ ],
+ )
route_34_eos.add_vpp_config()
#
# ping an interface in the default table
# PG0 is in the default table
#
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(34)],
- dst_ip="10.0.0.1")
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], dst_ip="10.0.0.1"
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip4(self.pg1, rx, tx)
def test_mcast_mid_point(self):
- """ MPLS Multicast Mid Point """
+ """MPLS Multicast Mid Point"""
#
# Add a non-recursive route that will forward the traffic
# post-interface-rx
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- table_id=1,
- paths=[VppRoutePath(self.pg1.remote_ip4,
- self.pg1.sw_if_index)])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ table_id=1,
+ paths=[VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index)],
+ )
route_10_0_0_1.add_vpp_config()
#
# and replicate to a interface-rx (like a bud node would)
#
route_3400_eos = VppMplsRoute(
- self, 3400, 1,
- [VppRoutePath(self.pg2.remote_ip4,
- self.pg2.sw_if_index,
- labels=[VppMplsLabel(3401)]),
- VppRoutePath(self.pg3.remote_ip4,
- self.pg3.sw_if_index,
- labels=[VppMplsLabel(3402)]),
- VppRoutePath("0.0.0.0",
- self.pg1.sw_if_index,
- type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX)],
- is_multicast=1)
+ self,
+ 3400,
+ 1,
+ [
+ VppRoutePath(
+ self.pg2.remote_ip4,
+ self.pg2.sw_if_index,
+ labels=[VppMplsLabel(3401)],
+ ),
+ VppRoutePath(
+ self.pg3.remote_ip4,
+ self.pg3.sw_if_index,
+ labels=[VppMplsLabel(3402)],
+ ),
+ VppRoutePath(
+ "0.0.0.0",
+ self.pg1.sw_if_index,
+ type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
+ ),
+ ],
+ is_multicast=1,
+ )
route_3400_eos.add_vpp_config()
#
# PG0 is in the default table
#
self.vapi.cli("clear trace")
- tx = self.create_stream_labelled_ip4(self.pg0,
- [VppMplsLabel(3400, ttl=64)],
- n=257,
- dst_ip="10.0.0.1")
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(3400, ttl=64)], n=257, dst_ip="10.0.0.1"
+ )
self.pg0.add_stream(tx)
self.pg_enable_capture(self.pg_interfaces)
self.verify_capture_ip4(self.pg1, rx, tx)
rx = self.pg2.get_capture(257)
- self.verify_capture_labelled(self.pg2, rx, tx,
- [VppMplsLabel(3401, ttl=63)])
+ self.verify_capture_labelled(self.pg2, rx, tx, [VppMplsLabel(3401, ttl=63)])
rx = self.pg3.get_capture(257)
- self.verify_capture_labelled(self.pg3, rx, tx,
- [VppMplsLabel(3402, ttl=63)])
+ self.verify_capture_labelled(self.pg3, rx, tx, [VppMplsLabel(3402, ttl=63)])
def test_mcast_head(self):
- """ MPLS Multicast Head-end """
+ """MPLS Multicast Head-end"""
MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
#
mpls_tun = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg2.remote_ip4,
- self.pg2.sw_if_index,
- labels=[VppMplsLabel(42)]),
- VppRoutePath(self.pg3.remote_ip4,
- self.pg3.sw_if_index,
- labels=[VppMplsLabel(43)])],
- is_multicast=1)
+ [
+ VppRoutePath(
+ self.pg2.remote_ip4, self.pg2.sw_if_index, labels=[VppMplsLabel(42)]
+ ),
+ VppRoutePath(
+ self.pg3.remote_ip4, self.pg3.sw_if_index, labels=[VppMplsLabel(43)]
+ ),
+ ],
+ is_multicast=1,
+ )
mpls_tun.add_vpp_config()
mpls_tun.admin_up()
#
# add an unlabelled route through the new tunnel
#
- route_10_0_0_3 = VppIpRoute(self, "10.0.0.3", 32,
- [VppRoutePath("0.0.0.0",
- mpls_tun._sw_if_index)])
+ route_10_0_0_3 = VppIpRoute(
+ self, "10.0.0.3", 32, [VppRoutePath("0.0.0.0", mpls_tun._sw_if_index)]
+ )
route_10_0_0_3.add_vpp_config()
self.vapi.cli("clear trace")
route_232_1_1_1 = VppIpMRoute(
self,
"0.0.0.0",
- "232.1.1.1", 32,
+ "232.1.1.1",
+ 32,
MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
- [VppMRoutePath(self.pg0.sw_if_index,
- MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT),
- VppMRoutePath(mpls_tun._sw_if_index,
- MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
+ [
+ VppMRoutePath(
+ self.pg0.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_ACCEPT
+ ),
+ VppMRoutePath(
+ mpls_tun._sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD
+ ),
+ ],
+ )
route_232_1_1_1.add_vpp_config()
self.logger.info(self.vapi.cli("sh ip mfib index 0"))
self.verify_capture_tunneled_ip4(self.pg0, rx, tx, [VppMplsLabel(43)])
def test_mcast_ip4_tail(self):
- """ MPLS IPv4 Multicast Tail """
+ """MPLS IPv4 Multicast Tail"""
MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
route_232_1_1_1 = VppIpMRoute(
self,
"0.0.0.0",
- "232.1.1.1", 32,
+ "232.1.1.1",
+ 32,
MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
table_id=1,
- paths=[VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD)])
+ paths=[
+ VppMRoutePath(
+ self.pg1.sw_if_index, MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD
+ )
+ ],
+ )
route_232_1_1_1.add_vpp_config()
#
# table 1
#
route_34_eos = VppMplsRoute(
- self, 34, 1,
- [VppRoutePath("0.0.0.0",
- 0xffffffff,
- nh_table_id=1,
- rpf_id=55)],
+ self,
+ 34,
+ 1,
+ [VppRoutePath("0.0.0.0", 0xFFFFFFFF, nh_table_id=1, rpf_id=55)],
is_multicast=1,
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP4)
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP4,
+ )
route_34_eos.add_vpp_config()
# Drop due to interface lookup miss
#
self.vapi.cli("clear trace")
- tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(34)],
- dst_ip="232.1.1.1", n=1)
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], dst_ip="232.1.1.1", n=1
+ )
self.send_and_assert_no_replies(self.pg0, tx, "RPF-ID drop none")
#
route_232_1_1_1.update_rpf_id(55)
self.logger.info(self.vapi.cli("sh ip mfib index 1 232.1.1.1"))
- tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(34)],
- dst_ip="232.1.1.1")
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], dst_ip="232.1.1.1"
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip4(self.pg1, rx, tx)
#
# disposed packets have an invalid IPv4 checksum
#
- tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(34)],
- dst_ip="232.1.1.1", n=65,
- chksum=1)
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], dst_ip="232.1.1.1", n=65, chksum=1
+ )
self.send_and_assert_no_replies(self.pg0, tx, "Invalid Checksum")
#
# set the RPF-ID of the entry to not match the input packet's
#
route_232_1_1_1.update_rpf_id(56)
- tx = self.create_stream_labelled_ip4(self.pg0, [VppMplsLabel(34)],
- dst_ip="232.1.1.1")
+ tx = self.create_stream_labelled_ip4(
+ self.pg0, [VppMplsLabel(34)], dst_ip="232.1.1.1"
+ )
self.send_and_assert_no_replies(self.pg0, tx, "RPF-ID drop 56")
def test_mcast_ip6_tail(self):
- """ MPLS IPv6 Multicast Tail """
+ """MPLS IPv6 Multicast Tail"""
MRouteItfFlags = VppEnum.vl_api_mfib_itf_flags_t
MRouteEntryFlags = VppEnum.vl_api_mfib_entry_flags_t
route_ff = VppIpMRoute(
self,
"::",
- "ff01::1", 32,
+ "ff01::1",
+ 32,
MRouteEntryFlags.MFIB_API_ENTRY_FLAG_NONE,
table_id=1,
- paths=[VppMRoutePath(self.pg1.sw_if_index,
- MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
- proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)])
+ paths=[
+ VppMRoutePath(
+ self.pg1.sw_if_index,
+ MRouteItfFlags.MFIB_API_ITF_FLAG_FORWARD,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
+ ],
+ )
route_ff.add_vpp_config()
#
# table 1
#
route_34_eos = VppMplsRoute(
- self, 34, 1,
- [VppRoutePath("::",
- 0xffffffff,
- nh_table_id=1,
- rpf_id=55)],
+ self,
+ 34,
+ 1,
+ [VppRoutePath("::", 0xFFFFFFFF, nh_table_id=1, rpf_id=55)],
is_multicast=1,
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6)
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_IP6,
+ )
route_34_eos.add_vpp_config()
#
# Drop due to interface lookup miss
#
- tx = self.create_stream_labelled_ip6(self.pg0, [VppMplsLabel(34)],
- dst_ip="ff01::1")
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(34)], dst_ip="ff01::1"
+ )
self.send_and_assert_no_replies(self.pg0, tx, "RPF Miss")
#
#
route_ff.update_rpf_id(55)
- tx = self.create_stream_labelled_ip6(self.pg0, [VppMplsLabel(34)],
- dst_ip="ff01::1")
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(34)], dst_ip="ff01::1"
+ )
rx = self.send_and_expect(self.pg0, tx, self.pg1)
self.verify_capture_ip6(self.pg1, rx, tx)
#
# disposed packets have hop-limit = 1
#
- tx = self.create_stream_labelled_ip6(self.pg0,
- [VppMplsLabel(34)],
- dst_ip="ff01::1",
- hlim=1)
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(34)], dst_ip="ff01::1", hlim=1
+ )
rx = self.send_and_expect_some(self.pg0, tx, self.pg0)
self.verify_capture_ip6_icmp(self.pg0, rx, tx)
# set the RPF-ID of the entry to not match the input packet's
#
route_ff.update_rpf_id(56)
- tx = self.create_stream_labelled_ip6(self.pg0,
- [VppMplsLabel(34)],
- dst_ip="ff01::1")
+ tx = self.create_stream_labelled_ip6(
+ self.pg0, [VppMplsLabel(34)], dst_ip="ff01::1"
+ )
self.send_and_assert_no_replies(self.pg0, tx, "RPF-ID drop 56")
def test_6pe(self):
- """ MPLS 6PE """
+ """MPLS 6PE"""
#
# Add a non-recursive route with a single out label
#
- route_10_0_0_1 = VppIpRoute(self, "10.0.0.1", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(45)])])
+ route_10_0_0_1 = VppIpRoute(
+ self,
+ "10.0.0.1",
+ 32,
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(45)]
+ )
+ ],
+ )
route_10_0_0_1.add_vpp_config()
# bind a local label to the route
# a labelled v6 route that resolves through the v4
#
route_2001_3 = VppIpRoute(
- self, "2001::3", 128,
- [VppRoutePath("10.0.0.1",
- INVALID_INDEX,
- labels=[VppMplsLabel(32)])])
+ self,
+ "2001::3",
+ 128,
+ [VppRoutePath("10.0.0.1", INVALID_INDEX, labels=[VppMplsLabel(32)])],
+ )
route_2001_3.add_vpp_config()
tx = self.create_stream_ip6(self.pg0, "2001::3")
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip6(self.pg0, rx, tx,
- [VppMplsLabel(45),
- VppMplsLabel(32)])
+ self.verify_capture_labelled_ip6(
+ self.pg0, rx, tx, [VppMplsLabel(45), VppMplsLabel(32)]
+ )
#
# and a v4 recursive via the v6
#
route_20_3 = VppIpRoute(
- self, "20.0.0.3", 32,
- [VppRoutePath("2001::3",
- INVALID_INDEX,
- labels=[VppMplsLabel(99)])])
+ self,
+ "20.0.0.3",
+ 32,
+ [VppRoutePath("2001::3", INVALID_INDEX, labels=[VppMplsLabel(99)])],
+ )
route_20_3.add_vpp_config()
tx = self.create_stream_ip4(self.pg0, "20.0.0.3")
rx = self.send_and_expect(self.pg0, tx, self.pg0)
- self.verify_capture_labelled_ip4(self.pg0, rx, tx,
- [VppMplsLabel(45),
- VppMplsLabel(32),
- VppMplsLabel(99)])
+ self.verify_capture_labelled_ip4(
+ self.pg0, rx, tx, [VppMplsLabel(45), VppMplsLabel(32), VppMplsLabel(99)]
+ )
def test_attached(self):
- """ Attach Routes with Local Label """
+ """Attach Routes with Local Label"""
#
# test that if a local label is associated with an attached/connected
# prefix, that we can reach hosts in the prefix.
#
- binding = VppMplsIpBind(self, 44,
- self.pg0._local_ip4_subnet,
- self.pg0.local_ip4_prefix_len)
+ binding = VppMplsIpBind(
+ self, 44, self.pg0._local_ip4_subnet, self.pg0.local_ip4_prefix_len
+ )
binding.add_vpp_config()
- tx = (Ether(src=self.pg1.remote_mac,
- dst=self.pg1.local_mac) /
- MPLS(label=44, ttl=64) /
- IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ tx = (
+ Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
+ / MPLS(label=44, ttl=64)
+ / IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4)
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
rxs = self.send_and_expect(self.pg0, [tx], self.pg0)
for rx in rxs:
# if there's an ARP then the label is linked to the glean
class TestMPLSDisabled(VppTestCase):
- """ MPLS disabled """
+ """MPLS disabled"""
@classmethod
def setUpClass(cls):
super(TestMPLSDisabled, self).tearDown()
def test_mpls_disabled(self):
- """ MPLS Disabled """
+ """MPLS Disabled"""
self.logger.info(self.vapi.cli("show mpls interface"))
self.logger.info(self.vapi.cli("show mpls interface pg1"))
self.logger.info(self.vapi.cli("show mpls interface pg0"))
- tx = (Ether(src=self.pg1.remote_mac,
- dst=self.pg1.local_mac) /
- MPLS(label=32, ttl=64) /
- IPv6(src="2001::1", dst=self.pg0.remote_ip6) /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ tx = (
+ Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac)
+ / MPLS(label=32, ttl=64)
+ / IPv6(src="2001::1", dst=self.pg0.remote_ip6)
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
#
# A simple MPLS xconnect - eos label in label out
#
- route_32_eos = VppMplsRoute(self, 32, 1,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[33])])
+ route_32_eos = VppMplsRoute(
+ self,
+ 32,
+ 1,
+ [VppRoutePath(self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[33])],
+ )
route_32_eos.add_vpp_config()
#
class TestMPLSPIC(VppTestCase):
- """ MPLS Prefix-Independent Convergence (PIC) edge convergence """
+ """MPLS Prefix-Independent Convergence (PIC) edge convergence"""
@classmethod
def setUpClass(cls):
super(TestMPLSPIC, self).tearDown()
def test_mpls_ibgp_pic(self):
- """ MPLS iBGP Prefix-Independent Convergence (PIC) edge convergence
+ """MPLS iBGP Prefix-Independent Convergence (PIC) edge convergence
1) setup many iBGP VPN routes via a pair of iBGP peers.
2) Check EMCP forwarding to these peers
#
# IGP+LDP core routes
#
- core_10_0_0_45 = VppIpRoute(self, "10.0.0.45", 32,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[45])])
+ core_10_0_0_45 = VppIpRoute(
+ self,
+ "10.0.0.45",
+ 32,
+ [VppRoutePath(self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[45])],
+ )
core_10_0_0_45.add_vpp_config()
- core_10_0_0_46 = VppIpRoute(self, "10.0.0.46", 32,
- [VppRoutePath(self.pg1.remote_ip4,
- self.pg1.sw_if_index,
- labels=[46])])
+ core_10_0_0_46 = VppIpRoute(
+ self,
+ "10.0.0.46",
+ 32,
+ [VppRoutePath(self.pg1.remote_ip4, self.pg1.sw_if_index, labels=[46])],
+ )
core_10_0_0_46.add_vpp_config()
#
pkts = []
for ii in range(NUM_PKTS):
dst = "192.168.1.%d" % ii
- vpn_routes.append(VppIpRoute(
- self, dst, 32,
- [VppRoutePath(
- "10.0.0.45",
- 0xffffffff,
- labels=[145],
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_HOST),
- VppRoutePath(
- "10.0.0.46",
- 0xffffffff,
- labels=[146],
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_HOST)],
- table_id=1))
+ vpn_routes.append(
+ VppIpRoute(
+ self,
+ dst,
+ 32,
+ [
+ VppRoutePath(
+ "10.0.0.45",
+ 0xFFFFFFFF,
+ labels=[145],
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_HOST,
+ ),
+ VppRoutePath(
+ "10.0.0.46",
+ 0xFFFFFFFF,
+ labels=[146],
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_HOST,
+ ),
+ ],
+ table_id=1,
+ )
+ )
vpn_routes[ii].add_vpp_config()
- pkts.append(Ether(dst=self.pg2.local_mac,
- src=self.pg2.remote_mac) /
- IP(src=self.pg2.remote_ip4, dst=dst) /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ pkts.append(
+ Ether(dst=self.pg2.local_mac, src=self.pg2.remote_mac)
+ / IP(src=self.pg2.remote_ip4, dst=dst)
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
#
# Send the packet stream (one pkt to each VPN route)
# with the split ratio, just as long as neither is 0
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
#
# use a test CLI command to stop the FIB walk process, this
self.pg_start()
rx0 = self.pg0.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# enable the FIB walk process to converge the FIB
self.pg_start()
rx0 = self.pg0.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# Add the IGP route back and we return to load-balancing
rx1 = self.pg1._get_capture(NUM_PKTS)
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
def test_mpls_ebgp_pic(self):
- """ MPLS eBGP Prefix-Independent Convergence (PIC) edge convergence
+ """MPLS eBGP Prefix-Independent Convergence (PIC) edge convergence
1) setup many eBGP VPN routes via a pair of eBGP peers.
2) Check EMCP forwarding to these peers
for ii in range(NUM_PKTS):
dst = "192.168.1.%d" % ii
local_label = 1600 + ii
- vpn_routes.append(VppIpRoute(
- self, dst, 32,
- [VppRoutePath(
- self.pg2.remote_ip4,
- 0xffffffff,
- nh_table_id=1,
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED),
- VppRoutePath(
- self.pg3.remote_ip4,
- 0xffffffff,
- nh_table_id=1,
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED)],
- table_id=1))
+ vpn_routes.append(
+ VppIpRoute(
+ self,
+ dst,
+ 32,
+ [
+ VppRoutePath(
+ self.pg2.remote_ip4,
+ 0xFFFFFFFF,
+ nh_table_id=1,
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED,
+ ),
+ VppRoutePath(
+ self.pg3.remote_ip4,
+ 0xFFFFFFFF,
+ nh_table_id=1,
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED,
+ ),
+ ],
+ table_id=1,
+ )
+ )
vpn_routes[ii].add_vpp_config()
- vpn_bindings.append(VppMplsIpBind(self, local_label, dst, 32,
- ip_table_id=1))
+ vpn_bindings.append(
+ VppMplsIpBind(self, local_label, dst, 32, ip_table_id=1)
+ )
vpn_bindings[ii].add_vpp_config()
- pkts.append(Ether(dst=self.pg0.local_mac,
- src=self.pg0.remote_mac) /
- MPLS(label=local_label, ttl=64) /
- IP(src=self.pg0.remote_ip4, dst=dst) /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ pkts.append(
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / MPLS(label=local_label, ttl=64)
+ / IP(src=self.pg0.remote_ip4, dst=dst)
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
#
# Send the packet stream (one pkt to each VPN route)
# with the split ratio, just as long as neither is 0
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
#
# use a test CLI command to stop the FIB walk process, this
self.pg_start()
rx0 = self.pg3.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# enable the FIB walk process to converge the FIB
self.pg_start()
rx0 = self.pg3.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# put the connected routes back
rx1 = self.pg3._get_capture(NUM_PKTS)
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
def test_mpls_v6_ebgp_pic(self):
- """ MPLSv6 eBGP Prefix-Independent Convergence (PIC) edge convergence
+ """MPLSv6 eBGP Prefix-Independent Convergence (PIC) edge convergence
1) setup many eBGP VPNv6 routes via a pair of eBGP peers
2) Check EMCP forwarding to these peers
for ii in range(NUM_PKTS):
dst = "3000::%d" % ii
local_label = 1600 + ii
- vpn_routes.append(VppIpRoute(
- self, dst, 128,
- [VppRoutePath(
- self.pg2.remote_ip6,
- 0xffffffff,
- nh_table_id=1,
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED),
- VppRoutePath(
- self.pg3.remote_ip6,
- 0xffffffff,
- nh_table_id=1,
- flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED)],
- table_id=1))
+ vpn_routes.append(
+ VppIpRoute(
+ self,
+ dst,
+ 128,
+ [
+ VppRoutePath(
+ self.pg2.remote_ip6,
+ 0xFFFFFFFF,
+ nh_table_id=1,
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED,
+ ),
+ VppRoutePath(
+ self.pg3.remote_ip6,
+ 0xFFFFFFFF,
+ nh_table_id=1,
+ flags=FibPathFlags.FIB_PATH_FLAG_RESOLVE_VIA_ATTACHED,
+ ),
+ ],
+ table_id=1,
+ )
+ )
vpn_routes[ii].add_vpp_config()
- vpn_bindings.append(VppMplsIpBind(self, local_label, dst, 128,
- ip_table_id=1))
+ vpn_bindings.append(
+ VppMplsIpBind(self, local_label, dst, 128, ip_table_id=1)
+ )
vpn_bindings[ii].add_vpp_config()
- pkts.append(Ether(dst=self.pg0.local_mac,
- src=self.pg0.remote_mac) /
- MPLS(label=local_label, ttl=64) /
- IPv6(src=self.pg0.remote_ip6, dst=dst) /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ pkts.append(
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / MPLS(label=local_label, ttl=64)
+ / IPv6(src=self.pg0.remote_ip6, dst=dst)
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
self.logger.info(self.vapi.cli("sh ip6 fib %s" % dst))
self.pg0.add_stream(pkts)
rx1 = self.pg3._get_capture(NUM_PKTS)
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
#
# use a test CLI command to stop the FIB walk process, this
self.pg_start()
rx0 = self.pg3.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# enable the FIB walk process to converge the FIB
self.pg_start()
rx0 = self.pg3.get_capture(NUM_PKTS)
- self.assertEqual(len(pkts), len(rx0),
- "Expected all (%s) packets across single path. "
- "rx0: %s." % (len(pkts), len(rx0)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0),
+ "Expected all (%s) packets across single path. "
+ "rx0: %s." % (len(pkts), len(rx0)),
+ )
#
# put the connected routes back
rx1 = self.pg3._get_capture(NUM_PKTS)
self.assertNotEqual(0, len(rx0))
self.assertNotEqual(0, len(rx1))
- self.assertEqual(len(pkts), len(rx0) + len(rx1),
- "Expected all (%s) packets across both ECMP paths. "
- "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)))
+ self.assertEqual(
+ len(pkts),
+ len(rx0) + len(rx1),
+ "Expected all (%s) packets across both ECMP paths. "
+ "rx0: %s rx1: %s." % (len(pkts), len(rx0), len(rx1)),
+ )
class TestMPLSL2(VppTestCase):
- """ MPLS-L2 """
+ """MPLS-L2"""
@classmethod
def setUpClass(cls):
self.assertEqual(arp.pdst, dip)
def test_vpws(self):
- """ Virtual Private Wire Service """
+ """Virtual Private Wire Service"""
#
# Create an MPLS tunnel that pushes 1 label
#
mpls_tun_1 = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(42, MplsLspMode.UNIFORM)])],
- is_l2=1)
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4,
+ self.pg0.sw_if_index,
+ labels=[VppMplsLabel(42, MplsLspMode.UNIFORM)],
+ )
+ ],
+ is_l2=1,
+ )
mpls_tun_1.add_vpp_config()
mpls_tun_1.admin_up()
# Create a label entry to for 55 that does L2 input to the tunnel
#
route_55_eos = VppMplsRoute(
- self, 55, 1,
- [VppRoutePath("0.0.0.0",
- mpls_tun_1.sw_if_index,
- type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
- proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)
+ self,
+ 55,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ mpls_tun_1.sw_if_index,
+ type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
+ ],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
route_55_eos.add_vpp_config()
#
# Cross-connect the tunnel with one of the customers L2 interfaces
#
- self.vapi.sw_interface_set_l2_xconnect(self.pg1.sw_if_index,
- mpls_tun_1.sw_if_index,
- enable=1)
- self.vapi.sw_interface_set_l2_xconnect(mpls_tun_1.sw_if_index,
- self.pg1.sw_if_index,
- enable=1)
+ self.vapi.sw_interface_set_l2_xconnect(
+ self.pg1.sw_if_index, mpls_tun_1.sw_if_index, enable=1
+ )
+ self.vapi.sw_interface_set_l2_xconnect(
+ mpls_tun_1.sw_if_index, self.pg1.sw_if_index, enable=1
+ )
#
# inject a packet from the core
#
- pcore = (Ether(dst=self.pg0.local_mac,
- src=self.pg0.remote_mac) /
- MPLS(label=55, ttl=64) /
- Ether(dst="00:00:de:ad:ba:be",
- src="00:00:de:ad:be:ef") /
- IP(src="10.10.10.10", dst="11.11.11.11") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ pcore = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / MPLS(label=55, ttl=64)
+ / Ether(dst="00:00:de:ad:ba:be", src="00:00:de:ad:be:ef")
+ / IP(src="10.10.10.10", dst="11.11.11.11")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
tx0 = pcore * NUM_PKTS
rx0 = self.send_and_expect(self.pg0, tx0, self.pg1)
tx1 = pcore[MPLS].payload
rx1 = self.send_and_expect(self.pg1, [tx1], self.pg0)
- self.verify_arp_req(rx1[0],
- self.pg0.local_mac,
- self.pg0.local_ip4,
- self.pg0.remote_ip4)
+ self.verify_arp_req(
+ rx1[0], self.pg0.local_mac, self.pg0.local_ip4, self.pg0.remote_ip4
+ )
#
# resolve the ARP entries and send again
self.verify_capture_tunneled_ethernet(rx1, tx1, [VppMplsLabel(42)])
def test_vpls(self):
- """ Virtual Private LAN Service """
+ """Virtual Private LAN Service"""
# we skipped this in the setup
self.pg0.resolve_arp()
#
mpls_tun1 = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(42)])],
- is_l2=1)
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(42)]
+ )
+ ],
+ is_l2=1,
+ )
mpls_tun1.add_vpp_config()
mpls_tun1.admin_up()
mpls_tun2 = VppMPLSTunnelInterface(
self,
- [VppRoutePath(self.pg0.remote_ip4,
- self.pg0.sw_if_index,
- labels=[VppMplsLabel(43)])],
- is_l2=1)
+ [
+ VppRoutePath(
+ self.pg0.remote_ip4, self.pg0.sw_if_index, labels=[VppMplsLabel(43)]
+ )
+ ],
+ is_l2=1,
+ )
mpls_tun2.add_vpp_config()
mpls_tun2.admin_up()
# the latter includes a Psuedo Wire Control Word
#
route_55_eos = VppMplsRoute(
- self, 55, 1,
- [VppRoutePath("0.0.0.0",
- mpls_tun1.sw_if_index,
- type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
- proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)
+ self,
+ 55,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ mpls_tun1.sw_if_index,
+ type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
+ ],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
route_56_eos = VppMplsRoute(
- self, 56, 1,
- [VppRoutePath("0.0.0.0",
- mpls_tun2.sw_if_index,
- type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
- flags=FibPathFlags.FIB_PATH_FLAG_POP_PW_CW,
- proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)],
- eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET)
+ self,
+ 56,
+ 1,
+ [
+ VppRoutePath(
+ "0.0.0.0",
+ mpls_tun2.sw_if_index,
+ type=FibPathType.FIB_PATH_TYPE_INTERFACE_RX,
+ flags=FibPathFlags.FIB_PATH_FLAG_POP_PW_CW,
+ proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
+ ],
+ eos_proto=FibPathProto.FIB_PATH_NH_PROTO_ETHERNET,
+ )
# move me
route_56_eos.add_vpp_config()
# add to tunnel to the customers bridge-domain
#
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=mpls_tun1.sw_if_index, bd_id=1)
+ rx_sw_if_index=mpls_tun1.sw_if_index, bd_id=1
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=mpls_tun2.sw_if_index, bd_id=1)
+ rx_sw_if_index=mpls_tun2.sw_if_index, bd_id=1
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.pg1.sw_if_index, bd_id=1)
+ rx_sw_if_index=self.pg1.sw_if_index, bd_id=1
+ )
#
# Packet from host on the customer interface to each host
# reachable over the core, and vice-versa
#
- p_cust1 = (Ether(dst="00:00:de:ad:ba:b1",
- src="00:00:de:ad:be:ef") /
- IP(src="10.10.10.10", dst="11.11.11.11") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
- p_cust2 = (Ether(dst="00:00:de:ad:ba:b2",
- src="00:00:de:ad:be:ef") /
- IP(src="10.10.10.10", dst="11.11.11.12") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
- p_core1 = (Ether(dst=self.pg0.local_mac,
- src=self.pg0.remote_mac) /
- MPLS(label=55, ttl=64) /
- Ether(src="00:00:de:ad:ba:b1",
- dst="00:00:de:ad:be:ef") /
- IP(dst="10.10.10.10", src="11.11.11.11") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
- p_core2 = (Ether(dst=self.pg0.local_mac,
- src=self.pg0.remote_mac) /
- MPLS(label=56, ttl=64) /
- Raw(b'\x01' * 4) / # PW CW
- Ether(src="00:00:de:ad:ba:b2",
- dst="00:00:de:ad:be:ef") /
- IP(dst="10.10.10.10", src="11.11.11.12") /
- UDP(sport=1234, dport=1234) /
- Raw(b'\xa5' * 100))
+ p_cust1 = (
+ Ether(dst="00:00:de:ad:ba:b1", src="00:00:de:ad:be:ef")
+ / IP(src="10.10.10.10", dst="11.11.11.11")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ p_cust2 = (
+ Ether(dst="00:00:de:ad:ba:b2", src="00:00:de:ad:be:ef")
+ / IP(src="10.10.10.10", dst="11.11.11.12")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ p_core1 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / MPLS(label=55, ttl=64)
+ / Ether(src="00:00:de:ad:ba:b1", dst="00:00:de:ad:be:ef")
+ / IP(dst="10.10.10.10", src="11.11.11.11")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
+ p_core2 = (
+ Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
+ / MPLS(label=56, ttl=64)
+ / Raw(b"\x01" * 4)
+ / Ether(src="00:00:de:ad:ba:b2", dst="00:00:de:ad:be:ef") # PW CW
+ / IP(dst="10.10.10.10", src="11.11.11.12")
+ / UDP(sport=1234, dport=1234)
+ / Raw(b"\xa5" * 100)
+ )
#
# The BD is learning, so send in one of each packet to learn
# now a stream in each direction from each host
#
rx = self.send_and_expect(self.pg1, p_cust1 * NUM_PKTS, self.pg0)
- self.verify_capture_tunneled_ethernet(rx, p_cust1 * NUM_PKTS,
- [VppMplsLabel(42)])
+ self.verify_capture_tunneled_ethernet(
+ rx, p_cust1 * NUM_PKTS, [VppMplsLabel(42)]
+ )
rx = self.send_and_expect(self.pg1, p_cust2 * NUM_PKTS, self.pg0)
- self.verify_capture_tunneled_ethernet(rx, p_cust2 * NUM_PKTS,
- [VppMplsLabel(43)])
+ self.verify_capture_tunneled_ethernet(
+ rx, p_cust2 * NUM_PKTS, [VppMplsLabel(43)]
+ )
rx = self.send_and_expect(self.pg0, p_core1 * NUM_PKTS, self.pg1)
rx = self.send_and_expect(self.pg0, p_core2 * NUM_PKTS, self.pg1)
# remove interfaces from customers bridge-domain
#
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=mpls_tun1.sw_if_index, bd_id=1, enable=0)
+ rx_sw_if_index=mpls_tun1.sw_if_index, bd_id=1, enable=0
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=mpls_tun2.sw_if_index, bd_id=1, enable=0)
+ rx_sw_if_index=mpls_tun2.sw_if_index, bd_id=1, enable=0
+ )
self.vapi.sw_interface_set_l2_bridge(
- rx_sw_if_index=self.pg1.sw_if_index, bd_id=1, enable=0)
+ rx_sw_if_index=self.pg1.sw_if_index, bd_id=1, enable=0
+ )
-if __name__ == '__main__':
+if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff