3 from socket import inet_pton, inet_ntop, AF_INET, AF_INET6
6 from framework import VppTestCase, VppTestRunner
7 from vpp_ip import DpoProto
8 from vpp_ip_route import VppIpRoute, VppRoutePath, VppMplsLabel, VppIpTable
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether
12 from scapy.layers.inet import IP, UDP
13 from scapy.layers.inet6 import IPv6
15 from vpp_object import VppObject
20 def find_l3xc(test, sw_if_index, dump_sw_if_index=None):
21 if not dump_sw_if_index:
22 dump_sw_if_index = sw_if_index
23 xcs = test.vapi.l3xc_dump(dump_sw_if_index)
25 if sw_if_index == xc.l3xc.sw_if_index:
30 class VppL3xc(VppObject):
32 def __init__(self, test, intf, paths, is_ip6=False):
38 def encode_paths(self):
43 if type(l) == VppMplsLabel:
44 lstack.append(l.encode())
46 lstack.append({'label': l, 'ttl': 255})
47 n_labels = len(lstack)
48 while (len(lstack) < 16):
50 br_paths.append({'next_hop': p.nh_addr,
53 'sw_if_index': p.nh_itf,
55 'table_id': p.nh_table_id,
56 'next_hop_id': p.next_hop_id,
57 'is_udp_encap': p.is_udp_encap,
59 'label_stack': lstack})
62 def add_vpp_config(self):
63 self._test.vapi.l3xc_update(
65 'is_ip6': self.is_ip6,
66 'sw_if_index': self.intf.sw_if_index,
67 'n_paths': len(self.paths),
68 'paths': self.encode_paths()
70 self._test.registry.register(self, self._test.logger)
72 def remove_vpp_config(self):
73 self._test.vapi.l3xc_del(
75 sw_if_index=self.intf.sw_if_index)
77 def query_vpp_config(self):
78 return find_l3xc(self._test, self.intf.sw_if_index)
81 return ("l3xc-%d" % self.intf.sw_if_index)
84 class TestL3xc(VppTestCase):
85 """ L3XC Test Case """
89 super(TestL3xc, cls).setUpClass()
92 def tearDownClass(cls):
93 super(TestL3xc, cls).tearDownClass()
96 super(TestL3xc, self).setUp()
98 self.create_pg_interfaces(range(6))
100 for i in self.pg_interfaces:
108 for i in self.pg_interfaces:
113 super(TestL3xc, self).tearDown()
115 def send_and_expect_load_balancing(self, input, pkts, outputs):
116 self.pg_send(input, pkts)
119 rx = oo._get_capture(1)
120 self.assertNotEqual(0, len(rx))
125 def test_l3xc4(self):
126 """ IPv4 X-Connect """
129 # x-connect pg0 to pg1 and pg2 to pg3->5
131 l3xc_1 = VppL3xc(self, self.pg0,
132 [VppRoutePath(self.pg1.remote_ip4,
133 self.pg1.sw_if_index)])
134 l3xc_1.add_vpp_config()
135 l3xc_2 = VppL3xc(self, self.pg2,
136 [VppRoutePath(self.pg3.remote_ip4,
137 self.pg3.sw_if_index),
138 VppRoutePath(self.pg4.remote_ip4,
139 self.pg4.sw_if_index),
140 VppRoutePath(self.pg5.remote_ip4,
141 self.pg5.sw_if_index)])
142 l3xc_2.add_vpp_config()
144 self.assertTrue(find_l3xc(self, self.pg2.sw_if_index, 0xffffffff))
146 self.logger.info(self.vapi.cli("sh l3xc"))
149 # fire in packets. If it's forwarded then the L3XC was successful,
150 # since default routing will drop it
152 p_1 = (Ether(src=self.pg0.remote_mac,
153 dst=self.pg0.local_mac) /
154 IP(src="1.1.1.1", dst="1.1.1.2") /
155 UDP(sport=1234, dport=1234) /
157 # self.send_and_expect(self.pg0, p_1*NUM_PKTS, self.pg1)
160 for ii in range(NUM_PKTS):
161 p_2.append(Ether(src=self.pg0.remote_mac,
162 dst=self.pg0.local_mac) /
163 IP(src="1.1.1.1", dst="1.1.1.2") /
164 UDP(sport=1000 + ii, dport=1234) /
166 self.send_and_expect_load_balancing(self.pg2, p_2,
167 [self.pg3, self.pg4, self.pg5])
169 l3xc_2.remove_vpp_config()
170 self.send_and_assert_no_replies(self.pg2, p_2)
173 if __name__ == '__main__':
174 unittest.main(testRunner=VppTestRunner)