+ 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"""
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestMPLSL2, cls).setUpClass()
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestMPLSL2, cls).tearDownClass()
+
+ def setUp(self):
+ super(TestMPLSL2, self).setUp()
+
+ # create 2 pg interfaces
+ self.create_pg_interfaces(range(2))
+
+ # create the default MPLS table
+ self.tables = []
+ tbl = VppMplsTable(self, 0)
+ tbl.add_vpp_config()
+ self.tables.append(tbl)
+
+ # use pg0 as the core facing interface, don't resolve ARP
+ self.pg0.admin_up()
+ self.pg0.config_ip4()
+ self.pg0.enable_mpls()
+
+ # use the other 2 for customer facing L2 links
+ for i in self.pg_interfaces[1:]:
+ i.admin_up()
+
+ def tearDown(self):
+ for i in self.pg_interfaces[1:]:
+ i.admin_down()
+
+ self.pg0.disable_mpls()
+ self.pg0.unconfig_ip4()
+ self.pg0.admin_down()
+ super(TestMPLSL2, self).tearDown()
+
+ def verify_capture_tunneled_ethernet(self, capture, sent, mpls_labels):
+ capture = verify_filter(capture, sent)
+
+ self.assertEqual(len(capture), len(sent))
+
+ for i in range(len(capture)):
+ tx = sent[i]
+ rx = capture[i]
+
+ # the MPLS TTL is 255 since it enters a new tunnel
+ verify_mpls_stack(self, rx, mpls_labels)
+
+ tx_eth = tx[Ether]
+ rx_eth = Ether(scapy.compat.raw(rx[MPLS].payload))
+
+ self.assertEqual(rx_eth.src, tx_eth.src)
+ self.assertEqual(rx_eth.dst, tx_eth.dst)
+
+ def verify_arp_req(self, rx, smac, sip, dip):
+ ether = rx[Ether]
+ self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff")
+ self.assertEqual(ether.src, smac)
+
+ arp = rx[ARP]
+ self.assertEqual(arp.hwtype, 1)
+ self.assertEqual(arp.ptype, 0x800)
+ self.assertEqual(arp.hwlen, 6)
+ self.assertEqual(arp.plen, 4)
+ self.assertEqual(arp.op, ARP.who_has)
+ self.assertEqual(arp.hwsrc, smac)
+ self.assertEqual(arp.hwdst, "00:00:00:00:00:00")
+ self.assertEqual(arp.psrc, sip)
+ self.assertEqual(arp.pdst, dip)
+
+ def test_vpws(self):
+ """Virtual Private Wire Service"""
+
+ #
+ # Create an MPLS tunnel that pushes 1 label
+ # For Ethernet over MPLS the uniform mode is irrelevant since ttl/cos
+ # information is not in the packet, but we test it works anyway
+ #
+ mpls_tun_1 = VppMPLSTunnelInterface(
+ self,
+ [
+ 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,
+ )
+ 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
+ )
+
+ #
+ # 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)
+ )
+
+ tx0 = pcore * NUM_PKTS
+ rx0 = self.send_and_expect(self.pg0, tx0, self.pg1)
+ payload = pcore[MPLS].payload
+
+ self.assertEqual(rx0[0][Ether].dst, payload[Ether].dst)
+ self.assertEqual(rx0[0][Ether].src, payload[Ether].src)
+
+ #
+ # Inject a packet from the customer/L2 side
+ # there's no resolved ARP entry so the first packet we see should be
+ # an ARP request
+ #
+ 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
+ )