+ def test_imposition_fragmentation(self):
+ """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.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.add_vpp_config()
+
+ #
+ # a stream that matches the route for 10.0.0.1
+ # PG0 is in the default table
+ #
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.1")
+ for i in range(0, 257):
+ self.extend_packet(tx[i], 10000)
+
+ #
+ # 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)]
+ )
+
+ # packets with DF bit set generate ICMP
+ for t in tx:
+ 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"
+ )
+ # 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/dont_fragment_set"),
+ len(tx),
+ )
+ #
+ # a stream that matches the route for 1000::1/128
+ # PG0 is in the default table
+ #
+ tx = self.create_stream_ip6(self.pg0, "1000::1")
+ for i in range(0, 257):
+ self.extend_packet(tx[i], 10000)
+
+ rxs = self.send_and_expect_some(self.pg0, tx, self.pg0)
+ for rx in rxs:
+ self.assertEqual(rx[ICMPv6PacketTooBig].mtu, 9000 - 4)
+
+ #
+ # cleanup
+ #
+ route_10_0_0_1.remove_vpp_config()
+
+ def test_tunnel_pipe(self):
+ """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)],
+ )
+ ],
+ )
+ 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.add_vpp_config()
+
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.3")
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx = self.pg0.get_capture()
+ 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.add_vpp_config()
+
+ self.vapi.cli("clear trace")
+ tx = self.create_stream_ip4(self.pg0, "10.0.0.4")
+ self.pg0.add_stream(tx)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ 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)],
+ )
+
+ #
+ # 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)
+
+ 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)]
+ )
+
+ # 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)
+
+ 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)],
+ )
+
+ def test_tunnel_uniform(self):
+ """MPLS Tunnel Tests - Uniform"""