3 from framework import VppTestCase, VppTestRunner
4 from vpp_udp_encap import find_udp_encap, VppUdpEncap
5 from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, VppMplsLabel
7 from scapy.packet import Raw
8 from scapy.layers.l2 import Ether
9 from scapy.layers.inet import IP, UDP
10 from scapy.layers.inet6 import IPv6
11 from scapy.contrib.mpls import MPLS
16 class TestUdpEncap(VppTestCase):
17 """ UDP Encap Test Case """
21 super(TestUdpEncap, cls).setUpClass()
24 def tearDownClass(cls):
25 super(TestUdpEncap, cls).tearDownClass()
28 super(TestUdpEncap, self).setUp()
30 # create 2 pg interfaces
31 self.create_pg_interfaces(range(4))
34 # assign them different tables.
38 for i in self.pg_interfaces:
42 tbl = VppIpTable(self, table_id)
44 self.tables.append(tbl)
45 tbl = VppIpTable(self, table_id, is_ip6=1)
47 self.tables.append(tbl)
49 i.set_table_ip4(table_id)
50 i.set_table_ip6(table_id)
58 for i in self.pg_interfaces:
65 super(TestUdpEncap, self).tearDown()
67 def validate_outer4(self, rx, encap_obj):
68 self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
69 self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
70 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
71 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
73 def validate_outer6(self, rx, encap_obj):
74 self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
75 self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
76 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
77 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
79 def validate_inner4(self, rx, tx, ttl=None):
80 self.assertEqual(rx[IP].src, tx[IP].src)
81 self.assertEqual(rx[IP].dst, tx[IP].dst)
83 self.assertEqual(rx[IP].ttl, ttl)
85 self.assertEqual(rx[IP].ttl, tx[IP].ttl)
87 def validate_inner6(self, rx, tx):
88 self.assertEqual(rx.src, tx[IPv6].src)
89 self.assertEqual(rx.dst, tx[IPv6].dst)
90 self.assertEqual(rx.hlim, tx[IPv6].hlim)
92 def test_udp_encap(self):
97 # construct a UDP encap object through each of the peers
98 # v4 through the first two peers, v6 through the second.
100 udp_encap_0 = VppUdpEncap(self,
104 udp_encap_1 = VppUdpEncap(self,
109 udp_encap_2 = VppUdpEncap(self,
114 udp_encap_3 = VppUdpEncap(self,
119 udp_encap_0.add_vpp_config()
120 udp_encap_1.add_vpp_config()
121 udp_encap_2.add_vpp_config()
122 udp_encap_3.add_vpp_config()
124 self.logger.info(self.vapi.cli("sh udp encap"))
126 self.assertTrue(find_udp_encap(self, udp_encap_2))
127 self.assertTrue(find_udp_encap(self, udp_encap_3))
128 self.assertTrue(find_udp_encap(self, udp_encap_0))
129 self.assertTrue(find_udp_encap(self, udp_encap_1))
132 # Routes via each UDP encap object - all combinations of v4 and v6.
134 route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
135 [VppRoutePath("0.0.0.0",
138 next_hop_id=udp_encap_0.id)])
139 route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
140 [VppRoutePath("0.0.0.0",
143 next_hop_id=udp_encap_2.id)])
144 route_6o4 = VppIpRoute(self, "2001::1", 128,
145 [VppRoutePath("0.0.0.0",
148 next_hop_id=udp_encap_1.id)],
150 route_6o6 = VppIpRoute(self, "2001::3", 128,
151 [VppRoutePath("0.0.0.0",
154 next_hop_id=udp_encap_3.id)],
156 route_4o4.add_vpp_config()
157 route_4o6.add_vpp_config()
158 route_6o6.add_vpp_config()
159 route_6o4.add_vpp_config()
164 p_4o4 = (Ether(src=self.pg0.remote_mac,
165 dst=self.pg0.local_mac) /
166 IP(src="2.2.2.2", dst="1.1.0.1") /
167 UDP(sport=1234, dport=1234) /
169 rx = self.send_and_expect(self.pg0, p_4o4*NUM_PKTS, self.pg0)
171 self.validate_outer4(p, udp_encap_0)
172 p = IP(p["UDP"].payload.load)
173 self.validate_inner4(p, p_4o4)
174 self.assertEqual(udp_encap_0.get_stats()['packets'], NUM_PKTS)
179 p_4o6 = (Ether(src=self.pg0.remote_mac,
180 dst=self.pg0.local_mac) /
181 IP(src="2.2.2.2", dst="1.1.2.1") /
182 UDP(sport=1234, dport=1234) /
184 rx = self.send_and_expect(self.pg0, p_4o6*NUM_PKTS, self.pg2)
186 self.validate_outer6(p, udp_encap_2)
187 p = IP(p["UDP"].payload.load)
188 self.validate_inner4(p, p_4o6)
189 self.assertEqual(udp_encap_2.get_stats()['packets'], NUM_PKTS)
194 p_6o4 = (Ether(src=self.pg0.remote_mac,
195 dst=self.pg0.local_mac) /
196 IPv6(src="2001::100", dst="2001::1") /
197 UDP(sport=1234, dport=1234) /
199 rx = self.send_and_expect(self.pg0, p_6o4*NUM_PKTS, self.pg1)
201 self.validate_outer4(p, udp_encap_1)
202 p = IPv6(p["UDP"].payload.load)
203 self.validate_inner6(p, p_6o4)
204 self.assertEqual(udp_encap_1.get_stats()['packets'], NUM_PKTS)
209 p_6o6 = (Ether(src=self.pg0.remote_mac,
210 dst=self.pg0.local_mac) /
211 IPv6(src="2001::100", dst="2001::3") /
212 UDP(sport=1234, dport=1234) /
214 rx = self.send_and_expect(self.pg0, p_6o6*NUM_PKTS, self.pg3)
216 self.validate_outer6(p, udp_encap_3)
217 p = IPv6(p["UDP"].payload.load)
218 self.validate_inner6(p, p_6o6)
219 self.assertEqual(udp_encap_3.get_stats()['packets'], NUM_PKTS)
222 # A route with an output label
223 # the TTL of the inner packet is decremented on LSP ingress
225 route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
226 [VppRoutePath("0.0.0.0",
230 labels=[VppMplsLabel(66)])])
231 route_4oMPLSo4.add_vpp_config()
233 p_4omo4 = (Ether(src=self.pg0.remote_mac,
234 dst=self.pg0.local_mac) /
235 IP(src="2.2.2.2", dst="1.1.2.22") /
236 UDP(sport=1234, dport=1234) /
238 rx = self.send_and_expect(self.pg0, p_4omo4*NUM_PKTS, self.pg1)
240 self.validate_outer4(p, udp_encap_1)
241 p = MPLS(p["UDP"].payload.load)
242 self.validate_inner4(p, p_4omo4, ttl=63)
243 self.assertEqual(udp_encap_1.get_stats()['packets'], 2*NUM_PKTS)
246 class TestUDP(VppTestCase):
247 """ UDP Test Case """
251 super(TestUDP, cls).setUpClass()
254 def tearDownClass(cls):
255 super(TestUDP, cls).tearDownClass()
258 super(TestUDP, self).setUp()
259 self.vapi.session_enable_disable(is_enabled=1)
260 self.create_loopback_interfaces(2)
264 for i in self.lo_interfaces:
268 tbl = VppIpTable(self, table_id)
271 i.set_table_ip4(table_id)
275 # Configure namespaces
276 self.vapi.app_namespace_add_del(namespace_id="0",
277 sw_if_index=self.loop0.sw_if_index)
278 self.vapi.app_namespace_add_del(namespace_id="1",
279 sw_if_index=self.loop1.sw_if_index)
282 for i in self.lo_interfaces:
286 self.vapi.session_enable_disable(is_enabled=0)
287 super(TestUDP, self).tearDown()
289 def test_udp_transfer(self):
290 """ UDP echo client/server transfer """
292 # Add inter-table routes
293 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
294 [VppRoutePath("0.0.0.0",
297 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
298 [VppRoutePath("0.0.0.0",
300 nh_table_id=0)], table_id=1)
301 ip_t01.add_vpp_config()
302 ip_t10.add_vpp_config()
304 # Start builtin server and client
305 uri = "udp://" + self.loop0.local_ip4 + "/1234"
306 error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
309 self.logger.critical(error)
310 self.assertNotIn("failed", error)
312 error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
313 "fifo-size 4 no-output test-bytes " +
314 "syn-timeout 2 no-return uri " + uri)
316 self.logger.critical(error)
317 self.assertNotIn("failed", error)
319 # Delete inter-table routes
320 ip_t01.remove_vpp_config()
321 ip_t10.remove_vpp_config()
324 if __name__ == '__main__':
325 unittest.main(testRunner=VppTestRunner)