+ def test_6pe(self):
+ """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.add_vpp_config()
+
+ # bind a local label to the route
+ binding = VppMplsIpBind(self, 44, "10.0.0.1", 32)
+ binding.add_vpp_config()
+
+ #
+ # 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)])],
+ )
+ 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)]
+ )
+
+ #
+ # 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)])],
+ )
+ 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)]
+ )
+
+ def test_attached(self):
+ """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.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)
+ )
+ 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
+ # which is wrong.
+ self.assertFalse(rx.haslayer(ARP))
+ # it should be unicasted to the host
+ self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
+ self.assertEqual(rx[IP].dst, self.pg0.remote_ip4)
+