tests: replace pycodestyle with black
Drop pycodestyle for code style checking in favor of black. Black is
much faster, stable PEP8 compliant code style checker offering also
automatic formatting. It aims to be very stable and produce smallest
diffs. It's used by many small and big projects.
Running checkstyle with black takes a few seconds with a terse output.
Thus, test-checkstyle-diff is no longer necessary.
Expand scope of checkstyle to all python files in the repo, replacing
test-checkstyle with checkstyle-python.
Also, fixstyle-python is now available for automatic style formatting.
Note: python virtualenv has been consolidated in test/Makefile,
test/requirements*.txt which will eventually be moved to a central
location. This is required to simply the automated generation of
docker executor images in the CI.
Type: improvement
Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8
Signed-off-by: Klement Sekera <klement.sekera@gmail.com>
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
diff --git a/test/test_mpls.py b/test/test_mpls.py
index a568f84..6e01a03 100644
--- a/test/test_mpls.py
+++ b/test/test_mpls.py
@@ -6,11 +6,24 @@
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
@@ -18,8 +31,12 @@
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
@@ -63,7 +80,7 @@
@tag_fixme_vpp_workers
class TestMPLS(VppTestCase):
- """ MPLS Test Case """
+ """MPLS Test Case"""
@classmethod
def setUpClass(cls):
@@ -120,15 +137,16 @@
# 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):
@@ -137,25 +155,32 @@
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
@@ -163,18 +188,20 @@
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)
@@ -187,18 +214,19 @@
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()
@@ -211,19 +239,23 @@
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)
@@ -256,8 +288,9 @@
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)
@@ -282,8 +315,9 @@
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)
@@ -330,8 +364,7 @@
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)
@@ -343,9 +376,9 @@
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))
@@ -363,7 +396,7 @@
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)
@@ -402,8 +435,9 @@
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)
@@ -426,8 +460,9 @@
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)
@@ -452,82 +487,115 @@
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)])
@@ -537,30 +605,36 @@
#
# 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)
@@ -568,11 +642,12 @@
# 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)])
@@ -582,10 +657,9 @@
#
# 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)
@@ -593,15 +667,19 @@
# 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)
@@ -609,25 +687,32 @@
# 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)
@@ -635,104 +720,136 @@
# 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()
@@ -743,15 +860,21 @@
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
@@ -759,19 +882,18 @@
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")
@@ -785,15 +907,21 @@
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()
#
@@ -807,65 +935,86 @@
#
# 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()
#
@@ -874,20 +1023,25 @@
#
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()
#
@@ -896,24 +1050,36 @@
#
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.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.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
@@ -924,20 +1090,32 @@
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()
#
@@ -952,24 +1130,29 @@
# 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
@@ -988,26 +1171,30 @@
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")
@@ -1018,17 +1205,19 @@
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")
@@ -1039,10 +1228,12 @@
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
@@ -1050,34 +1241,34 @@
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
@@ -1085,19 +1276,26 @@
#
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")
@@ -1108,18 +1306,23 @@
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")
@@ -1130,45 +1333,63 @@
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
@@ -1188,7 +1409,7 @@
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
@@ -1208,42 +1429,40 @@
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()
#
@@ -1252,26 +1471,25 @@
# 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)
@@ -1280,16 +1498,19 @@
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()
#
@@ -1301,33 +1522,43 @@
# 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()
#
@@ -1335,17 +1566,28 @@
# 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()
#
@@ -1353,10 +1595,9 @@
# 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)
@@ -1366,14 +1607,12 @@
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
@@ -1383,22 +1622,25 @@
#
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")
@@ -1421,12 +1663,18 @@
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"))
@@ -1443,7 +1691,7 @@
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
@@ -1455,11 +1703,16 @@
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()
#
@@ -1471,13 +1724,13 @@
# 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()
@@ -1485,8 +1738,9 @@
# 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")
#
@@ -1495,29 +1749,31 @@
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
@@ -1529,12 +1785,18 @@
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()
#
@@ -1546,21 +1808,22 @@
# 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")
#
@@ -1568,18 +1831,18 @@
#
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)
@@ -1587,21 +1850,27 @@
# 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
@@ -1612,55 +1881,57 @@
# 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
@@ -1672,7 +1943,7 @@
class TestMPLSDisabled(VppTestCase):
- """ MPLS disabled """
+ """MPLS disabled"""
@classmethod
def setUpClass(cls):
@@ -1709,26 +1980,29 @@
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()
#
@@ -1766,7 +2040,7 @@
class TestMPLSPIC(VppTestCase):
- """ MPLS Prefix-Independent Convergence (PIC) edge convergence """
+ """MPLS Prefix-Independent Convergence (PIC) edge convergence"""
@classmethod
def setUpClass(cls):
@@ -1829,7 +2103,7 @@
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
@@ -1840,16 +2114,20 @@
#
# 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()
#
@@ -1860,26 +2138,36 @@
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)
@@ -1896,9 +2184,12 @@
# 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
@@ -1921,9 +2212,12 @@
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
@@ -1938,9 +2232,12 @@
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
@@ -1955,12 +2252,15 @@
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
@@ -1977,31 +2277,42 @@
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)
@@ -2018,9 +2329,12 @@
# 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
@@ -2042,9 +2356,12 @@
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
@@ -2059,9 +2376,12 @@
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
@@ -2077,12 +2397,15 @@
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
@@ -2099,31 +2422,42 @@
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)
@@ -2134,9 +2468,12 @@
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
@@ -2160,9 +2497,12 @@
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
@@ -2173,9 +2513,12 @@
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
@@ -2193,13 +2536,16 @@
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):
@@ -2274,7 +2620,7 @@
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
@@ -2283,10 +2629,15 @@
#
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()
@@ -2294,35 +2645,42 @@
# 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)
@@ -2339,10 +2697,9 @@
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
@@ -2354,7 +2711,7 @@
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()
@@ -2364,19 +2721,25 @@
#
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()
@@ -2385,21 +2748,35 @@
# 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()
@@ -2411,43 +2788,48 @@
# 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
@@ -2470,12 +2852,14 @@
# 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)
@@ -2484,12 +2868,15 @@
# 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)