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
14 class TestUdpEncap(VppTestCase):
15 """ UDP Encap Test Case """
19 super(TestUdpEncap, cls).setUpClass()
22 def tearDownClass(cls):
23 super(TestUdpEncap, cls).tearDownClass()
26 super(TestUdpEncap, self).setUp()
28 # create 2 pg interfaces
29 self.create_pg_interfaces(range(4))
32 # assign them different tables.
36 for i in self.pg_interfaces:
40 tbl = VppIpTable(self, table_id)
42 self.tables.append(tbl)
43 tbl = VppIpTable(self, table_id, is_ip6=1)
45 self.tables.append(tbl)
47 i.set_table_ip4(table_id)
48 i.set_table_ip6(table_id)
56 for i in self.pg_interfaces:
63 super(TestUdpEncap, self).tearDown()
65 def validate_outer4(self, rx, encap_obj):
66 self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
67 self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
68 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
69 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
71 def validate_outer6(self, rx, encap_obj):
72 self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
73 self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
74 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
75 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
77 def validate_inner4(self, rx, tx, ttl=None):
78 self.assertEqual(rx[IP].src, tx[IP].src)
79 self.assertEqual(rx[IP].dst, tx[IP].dst)
81 self.assertEqual(rx[IP].ttl, ttl)
83 self.assertEqual(rx[IP].ttl, tx[IP].ttl)
85 def validate_inner6(self, rx, tx):
86 self.assertEqual(rx.src, tx[IPv6].src)
87 self.assertEqual(rx.dst, tx[IPv6].dst)
88 self.assertEqual(rx.hlim, tx[IPv6].hlim)
90 def test_udp_encap(self):
95 # construct a UDP encap object through each of the peers
96 # v4 through the first two peers, v6 through the second.
98 udp_encap_0 = VppUdpEncap(self,
102 udp_encap_1 = VppUdpEncap(self,
107 udp_encap_2 = VppUdpEncap(self,
112 udp_encap_3 = VppUdpEncap(self,
117 udp_encap_0.add_vpp_config()
118 udp_encap_1.add_vpp_config()
119 udp_encap_2.add_vpp_config()
120 udp_encap_3.add_vpp_config()
122 self.logger.info(self.vapi.cli("sh udp encap"))
124 self.assertTrue(find_udp_encap(self, udp_encap_2))
125 self.assertTrue(find_udp_encap(self, udp_encap_3))
126 self.assertTrue(find_udp_encap(self, udp_encap_0))
127 self.assertTrue(find_udp_encap(self, udp_encap_1))
130 # Routes via each UDP encap object - all combinations of v4 and v6.
132 route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
133 [VppRoutePath("0.0.0.0",
136 next_hop_id=udp_encap_0.id)])
137 route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
138 [VppRoutePath("0.0.0.0",
141 next_hop_id=udp_encap_2.id)])
142 route_6o4 = VppIpRoute(self, "2001::1", 128,
143 [VppRoutePath("0.0.0.0",
146 next_hop_id=udp_encap_1.id)],
148 route_6o6 = VppIpRoute(self, "2001::3", 128,
149 [VppRoutePath("0.0.0.0",
152 next_hop_id=udp_encap_3.id)],
154 route_4o4.add_vpp_config()
155 route_4o6.add_vpp_config()
156 route_6o6.add_vpp_config()
157 route_6o4.add_vpp_config()
162 p_4o4 = (Ether(src=self.pg0.remote_mac,
163 dst=self.pg0.local_mac) /
164 IP(src="2.2.2.2", dst="1.1.0.1") /
165 UDP(sport=1234, dport=1234) /
167 rx = self.send_and_expect(self.pg0, p_4o4*65, self.pg0)
169 self.validate_outer4(p, udp_encap_0)
170 p = IP(p["UDP"].payload.load)
171 self.validate_inner4(p, p_4o4)
172 self.assertEqual(udp_encap_0.get_stats()['packets'], 65)
177 p_4o6 = (Ether(src=self.pg0.remote_mac,
178 dst=self.pg0.local_mac) /
179 IP(src="2.2.2.2", dst="1.1.2.1") /
180 UDP(sport=1234, dport=1234) /
182 rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
184 self.validate_outer6(p, udp_encap_2)
185 p = IP(p["UDP"].payload.load)
186 self.validate_inner4(p, p_4o6)
187 self.assertEqual(udp_encap_2.get_stats()['packets'], 65)
192 p_6o4 = (Ether(src=self.pg0.remote_mac,
193 dst=self.pg0.local_mac) /
194 IPv6(src="2001::100", dst="2001::1") /
195 UDP(sport=1234, dport=1234) /
197 rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
199 self.validate_outer4(p, udp_encap_1)
200 p = IPv6(p["UDP"].payload.load)
201 self.validate_inner6(p, p_6o4)
202 self.assertEqual(udp_encap_1.get_stats()['packets'], 65)
207 p_6o6 = (Ether(src=self.pg0.remote_mac,
208 dst=self.pg0.local_mac) /
209 IPv6(src="2001::100", dst="2001::3") /
210 UDP(sport=1234, dport=1234) /
212 rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
214 self.validate_outer6(p, udp_encap_3)
215 p = IPv6(p["UDP"].payload.load)
216 self.validate_inner6(p, p_6o6)
217 self.assertEqual(udp_encap_3.get_stats()['packets'], 65)
220 # A route with an output label
221 # the TTL of the inner packet is decremented on LSP ingress
223 route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
224 [VppRoutePath("0.0.0.0",
228 labels=[VppMplsLabel(66)])])
229 route_4oMPLSo4.add_vpp_config()
231 p_4omo4 = (Ether(src=self.pg0.remote_mac,
232 dst=self.pg0.local_mac) /
233 IP(src="2.2.2.2", dst="1.1.2.22") /
234 UDP(sport=1234, dport=1234) /
236 rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
238 self.validate_outer4(p, udp_encap_1)
239 p = MPLS(p["UDP"].payload.load)
240 self.validate_inner4(p, p_4omo4, ttl=63)
241 self.assertEqual(udp_encap_1.get_stats()['packets'], 130)
244 class TestUDP(VppTestCase):
245 """ UDP Test Case """
249 super(TestUDP, cls).setUpClass()
252 def tearDownClass(cls):
253 super(TestUDP, cls).tearDownClass()
256 super(TestUDP, self).setUp()
257 self.vapi.session_enable_disable(is_enabled=1)
258 self.create_loopback_interfaces(2)
262 for i in self.lo_interfaces:
266 tbl = VppIpTable(self, table_id)
269 i.set_table_ip4(table_id)
273 # Configure namespaces
274 self.vapi.app_namespace_add_del(namespace_id="0",
275 sw_if_index=self.loop0.sw_if_index)
276 self.vapi.app_namespace_add_del(namespace_id="1",
277 sw_if_index=self.loop1.sw_if_index)
280 for i in self.lo_interfaces:
284 self.vapi.session_enable_disable(is_enabled=0)
285 super(TestUDP, self).tearDown()
287 def test_udp_transfer(self):
288 """ UDP echo client/server transfer """
290 # Add inter-table routes
291 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
292 [VppRoutePath("0.0.0.0",
295 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
296 [VppRoutePath("0.0.0.0",
298 nh_table_id=0)], table_id=1)
299 ip_t01.add_vpp_config()
300 ip_t10.add_vpp_config()
302 # Start builtin server and client
303 uri = "udp://" + self.loop0.local_ip4 + "/1234"
304 error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
307 self.logger.critical(error)
308 self.assertNotIn("failed", error)
310 error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
311 "fifo-size 4 no-output test-bytes " +
312 "syn-timeout 2 no-return uri " + uri)
314 self.logger.critical(error)
315 self.assertNotIn("failed", error)
317 # Delete inter-table routes
318 ip_t01.remove_vpp_config()
319 ip_t10.remove_vpp_config()
322 if __name__ == '__main__':
323 unittest.main(testRunner=VppTestRunner)