aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_mpls.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_mpls.py')
-rw-r--r--test/test_mpls.py114
1 files changed, 112 insertions, 2 deletions
diff --git a/test/test_mpls.py b/test/test_mpls.py
index d265e85ed5e..9590519e237 100644
--- a/test/test_mpls.py
+++ b/test/test_mpls.py
@@ -12,7 +12,7 @@ from vpp_mpls_tunnel_interface import VppMPLSTunnelInterface
from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, ICMP
-from scapy.layers.inet6 import IPv6
+from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded
from scapy.contrib.mpls import MPLS
@@ -288,6 +288,32 @@ class TestMPLS(VppTestCase):
except:
raise
+ def verify_capture_ip6_icmp(self, src_if, capture, sent):
+ try:
+ self.assertEqual(len(capture), len(sent))
+
+ for i in range(len(capture)):
+ tx = sent[i]
+ rx = capture[i]
+
+ # the rx'd packet has the MPLS label popped
+ eth = rx[Ether]
+ self.assertEqual(eth.type, 0x86DD)
+
+ tx_ip = tx[IPv6]
+ rx_ip = rx[IPv6]
+
+ self.assertEqual(rx_ip.dst, tx_ip.src)
+ # ICMP sourced from the interface's address
+ self.assertEqual(rx_ip.src, src_if.local_ip6)
+ # hop-limit reset to 255 for IMCP packet
+ self.assertEqual(rx_ip.hlim, 254)
+
+ icmp = rx[ICMPv6TimeExceeded]
+
+ except:
+ raise
+
def send_and_assert_no_replies(self, intf, pkts, remark):
intf.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
@@ -364,6 +390,91 @@ class TestMPLS(VppTestCase):
self.verify_capture_ip4(self.pg0, rx, tx)
#
+ # disposed packets have an invalid IPv4 checkusm
+ #
+ tx = self.create_stream_labelled_ip4(self.pg0, [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 IPv6 out
+ #
+ route_333_eos = VppMplsRoute(
+ self, 333, 1,
+ [VppRoutePath(self.pg0.remote_ip6,
+ self.pg0.sw_if_index,
+ labels=[],
+ proto=DpoProto.DPO_PROTO_IP6)])
+ route_333_eos.add_vpp_config()
+
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_labelled_ip6(self.pg0, [333], 64)
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture()
+ self.verify_capture_ip6(self.pg0, rx, tx)
+
+ #
+ # disposed packets have an TTL expired
+ #
+ tx = self.create_stream_labelled_ip6(self.pg0, [333], 64,
+ dst_ip=self.pg1.remote_ip6,
+ hlim=1)
+
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_labelled_ip6(self.pg0, [333], 64,
+ dst_ip=self.pg1.remote_ip6,
+ hlim=0)
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture()
+ 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=[3],
+ proto=DpoProto.DPO_PROTO_IP6)])
+ route_334_eos.add_vpp_config()
+
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_labelled_ip6(self.pg0, [334], 64)
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture()
+ self.verify_capture_ip6(self.pg0, rx, tx)
+
+ #
+ # disposed packets have an TTL expired
+ #
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_labelled_ip6(self.pg0, [334], 64,
+ dst_ip=self.pg1.remote_ip6,
+ hlim=0)
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture()
+ 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.
#
@@ -1043,7 +1154,6 @@ class TestMPLS(VppTestCase):
tx = self.create_stream_labelled_ip4(self.pg0, [34],
dst_ip="232.1.1.1")
self.send_and_assert_no_replies(self.pg0, tx, "RPF-ID drop 56")
- self.logger.error(self.vapi.cli("sh error"))
def test_mcast_ip6_tail(self):
""" MPLS IPv6 Multicast Tail """