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 """
18 super(TestUdpEncap, self).setUp()
20 # create 2 pg interfaces
21 self.create_pg_interfaces(range(4))
24 # assign them different tables.
28 for i in self.pg_interfaces:
32 tbl = VppIpTable(self, table_id)
34 self.tables.append(tbl)
35 tbl = VppIpTable(self, table_id, is_ip6=1)
37 self.tables.append(tbl)
39 i.set_table_ip4(table_id)
40 i.set_table_ip6(table_id)
48 for i in self.pg_interfaces:
55 super(TestUdpEncap, self).tearDown()
57 def validate_outer4(self, rx, encap_obj):
58 self.assertEqual(rx[IP].src, encap_obj.src_ip_s)
59 self.assertEqual(rx[IP].dst, encap_obj.dst_ip_s)
60 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
61 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
63 def validate_outer6(self, rx, encap_obj):
64 self.assertEqual(rx[IPv6].src, encap_obj.src_ip_s)
65 self.assertEqual(rx[IPv6].dst, encap_obj.dst_ip_s)
66 self.assertEqual(rx[UDP].sport, encap_obj.src_port)
67 self.assertEqual(rx[UDP].dport, encap_obj.dst_port)
69 def validate_inner4(self, rx, tx, ttl=None):
70 self.assertEqual(rx[IP].src, tx[IP].src)
71 self.assertEqual(rx[IP].dst, tx[IP].dst)
73 self.assertEqual(rx[IP].ttl, ttl)
75 self.assertEqual(rx[IP].ttl, tx[IP].ttl)
77 def validate_inner6(self, rx, tx):
78 self.assertEqual(rx.src, tx[IPv6].src)
79 self.assertEqual(rx.dst, tx[IPv6].dst)
80 self.assertEqual(rx.hlim, tx[IPv6].hlim)
82 def test_udp_encap(self):
87 # construct a UDP encap object through each of the peers
88 # v4 through the first two peears, v6 through the second.
90 udp_encap_0 = VppUdpEncap(self,
94 udp_encap_1 = VppUdpEncap(self,
99 udp_encap_2 = VppUdpEncap(self,
104 udp_encap_3 = VppUdpEncap(self,
109 udp_encap_0.add_vpp_config()
110 udp_encap_1.add_vpp_config()
111 udp_encap_2.add_vpp_config()
112 udp_encap_3.add_vpp_config()
114 self.logger.info(self.vapi.cli("sh udp encap"))
116 self.assertTrue(find_udp_encap(self, udp_encap_2))
117 self.assertTrue(find_udp_encap(self, udp_encap_3))
118 self.assertTrue(find_udp_encap(self, udp_encap_0))
119 self.assertTrue(find_udp_encap(self, udp_encap_1))
122 # Routes via each UDP encap object - all combinations of v4 and v6.
124 route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
125 [VppRoutePath("0.0.0.0",
128 next_hop_id=udp_encap_0.id)])
129 route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
130 [VppRoutePath("0.0.0.0",
133 next_hop_id=udp_encap_2.id)])
134 route_6o4 = VppIpRoute(self, "2001::1", 128,
135 [VppRoutePath("0.0.0.0",
138 next_hop_id=udp_encap_1.id)],
140 route_6o6 = VppIpRoute(self, "2001::3", 128,
141 [VppRoutePath("0.0.0.0",
144 next_hop_id=udp_encap_3.id)],
146 route_4o4.add_vpp_config()
147 route_4o6.add_vpp_config()
148 route_6o6.add_vpp_config()
149 route_6o4.add_vpp_config()
154 p_4o4 = (Ether(src=self.pg0.remote_mac,
155 dst=self.pg0.local_mac) /
156 IP(src="2.2.2.2", dst="1.1.0.1") /
157 UDP(sport=1234, dport=1234) /
159 rx = self.send_and_expect(self.pg0, p_4o4*65, self.pg0)
161 self.validate_outer4(p, udp_encap_0)
162 p = IP(p["UDP"].payload.load)
163 self.validate_inner4(p, p_4o4)
164 self.assertEqual(udp_encap_0.get_stats()['packets'], 65)
169 p_4o6 = (Ether(src=self.pg0.remote_mac,
170 dst=self.pg0.local_mac) /
171 IP(src="2.2.2.2", dst="1.1.2.1") /
172 UDP(sport=1234, dport=1234) /
174 rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
176 self.validate_outer6(p, udp_encap_2)
177 p = IP(p["UDP"].payload.load)
178 self.validate_inner4(p, p_4o6)
179 self.assertEqual(udp_encap_2.get_stats()['packets'], 65)
184 p_6o4 = (Ether(src=self.pg0.remote_mac,
185 dst=self.pg0.local_mac) /
186 IPv6(src="2001::100", dst="2001::1") /
187 UDP(sport=1234, dport=1234) /
189 rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
191 self.validate_outer4(p, udp_encap_1)
192 p = IPv6(p["UDP"].payload.load)
193 self.validate_inner6(p, p_6o4)
194 self.assertEqual(udp_encap_1.get_stats()['packets'], 65)
199 p_6o6 = (Ether(src=self.pg0.remote_mac,
200 dst=self.pg0.local_mac) /
201 IPv6(src="2001::100", dst="2001::3") /
202 UDP(sport=1234, dport=1234) /
204 rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
206 self.validate_outer6(p, udp_encap_3)
207 p = IPv6(p["UDP"].payload.load)
208 self.validate_inner6(p, p_6o6)
209 self.assertEqual(udp_encap_3.get_stats()['packets'], 65)
212 # A route with an output label
213 # the TTL of the inner packet is decremented on LSP ingress
215 route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
216 [VppRoutePath("0.0.0.0",
220 labels=[VppMplsLabel(66)])])
221 route_4oMPLSo4.add_vpp_config()
223 p_4omo4 = (Ether(src=self.pg0.remote_mac,
224 dst=self.pg0.local_mac) /
225 IP(src="2.2.2.2", dst="1.1.2.22") /
226 UDP(sport=1234, dport=1234) /
228 rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
230 self.validate_outer4(p, udp_encap_1)
231 p = MPLS(p["UDP"].payload.load)
232 self.validate_inner4(p, p_4omo4, ttl=63)
233 self.assertEqual(udp_encap_1.get_stats()['packets'], 130)
236 class TestUDP(VppTestCase):
237 """ UDP Test Case """
241 super(TestUDP, cls).setUpClass()
244 super(TestUDP, self).setUp()
245 self.vapi.session_enable_disable(is_enabled=1)
246 self.create_loopback_interfaces(2)
250 for i in self.lo_interfaces:
254 tbl = VppIpTable(self, table_id)
257 i.set_table_ip4(table_id)
261 # Configure namespaces
262 self.vapi.app_namespace_add_del(namespace_id="0",
263 sw_if_index=self.loop0.sw_if_index)
264 self.vapi.app_namespace_add_del(namespace_id="1",
265 sw_if_index=self.loop1.sw_if_index)
268 for i in self.lo_interfaces:
272 self.vapi.session_enable_disable(is_enabled=0)
273 super(TestUDP, self).tearDown()
275 def test_udp_transfer(self):
276 """ UDP echo client/server transfer """
278 # Add inter-table routes
279 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
280 [VppRoutePath("0.0.0.0",
283 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
284 [VppRoutePath("0.0.0.0",
286 nh_table_id=0)], table_id=1)
287 ip_t01.add_vpp_config()
288 ip_t10.add_vpp_config()
290 # Start builtin server and client
291 uri = "udp://" + self.loop0.local_ip4 + "/1234"
292 error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
295 self.logger.critical(error)
296 self.assertNotIn("failed", error)
298 error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
299 "fifo-size 4 no-output test-bytes " +
300 "syn-timeout 2 no-return uri " + uri)
302 self.logger.critical(error)
303 self.assertNotIn("failed", error)
305 # Delete inter-table routes
306 ip_t01.remove_vpp_config()
307 ip_t10.remove_vpp_config()
310 if __name__ == '__main__':
311 unittest.main(testRunner=VppTestRunner)