3 from framework import VppTestCase, VppTestRunner
4 from vpp_udp_encap import *
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, 0,
94 udp_encap_1 = VppUdpEncap(self, 1,
99 udp_encap_2 = VppUdpEncap(self, 2,
104 udp_encap_3 = VppUdpEncap(self, 3,
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",
129 route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
130 [VppRoutePath("0.0.0.0",
134 route_6o4 = VppIpRoute(self, "2001::1", 128,
135 [VppRoutePath("0.0.0.0",
140 route_6o6 = VppIpRoute(self, "2001::3", 128,
141 [VppRoutePath("0.0.0.0",
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)
168 p_4o6 = (Ether(src=self.pg0.remote_mac,
169 dst=self.pg0.local_mac) /
170 IP(src="2.2.2.2", dst="1.1.2.1") /
171 UDP(sport=1234, dport=1234) /
173 rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
175 self.validate_outer6(p, udp_encap_2)
176 p = IP(p["UDP"].payload.load)
177 self.validate_inner4(p, p_4o6)
182 p_6o4 = (Ether(src=self.pg0.remote_mac,
183 dst=self.pg0.local_mac) /
184 IPv6(src="2001::100", dst="2001::1") /
185 UDP(sport=1234, dport=1234) /
187 rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
189 self.validate_outer4(p, udp_encap_1)
190 p = IPv6(p["UDP"].payload.load)
191 self.validate_inner6(p, p_6o4)
196 p_6o6 = (Ether(src=self.pg0.remote_mac,
197 dst=self.pg0.local_mac) /
198 IPv6(src="2001::100", dst="2001::3") /
199 UDP(sport=1234, dport=1234) /
201 rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
203 self.validate_outer6(p, udp_encap_3)
204 p = IPv6(p["UDP"].payload.load)
205 self.validate_inner6(p, p_6o6)
208 # A route with an output label
209 # the TTL of the inner packet is decremented on LSP ingress
211 route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
212 [VppRoutePath("0.0.0.0",
216 labels=[VppMplsLabel(66)])])
217 route_4oMPLSo4.add_vpp_config()
219 p_4omo4 = (Ether(src=self.pg0.remote_mac,
220 dst=self.pg0.local_mac) /
221 IP(src="2.2.2.2", dst="1.1.2.22") /
222 UDP(sport=1234, dport=1234) /
224 rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
226 self.validate_outer4(p, udp_encap_1)
227 p = MPLS(p["UDP"].payload.load)
228 self.validate_inner4(p, p_4omo4, ttl=63)
231 class TestUDP(VppTestCase):
232 """ UDP Test Case """
236 super(TestUDP, cls).setUpClass()
239 super(TestUDP, self).setUp()
240 self.vapi.session_enable_disable(is_enabled=1)
241 self.create_loopback_interfaces(2)
245 for i in self.lo_interfaces:
249 tbl = VppIpTable(self, table_id)
252 i.set_table_ip4(table_id)
256 # Configure namespaces
257 self.vapi.app_namespace_add(namespace_id="0",
258 sw_if_index=self.loop0.sw_if_index)
259 self.vapi.app_namespace_add(namespace_id="1",
260 sw_if_index=self.loop1.sw_if_index)
263 for i in self.lo_interfaces:
267 self.vapi.session_enable_disable(is_enabled=0)
268 super(TestUDP, self).tearDown()
270 def test_udp_transfer(self):
271 """ UDP echo client/server transfer """
273 # Add inter-table routes
274 ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
275 [VppRoutePath("0.0.0.0",
278 ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
279 [VppRoutePath("0.0.0.0",
281 nh_table_id=0)], table_id=1)
282 ip_t01.add_vpp_config()
283 ip_t10.add_vpp_config()
285 # Start builtin server and client
286 uri = "udp://" + self.loop0.local_ip4 + "/1234"
287 error = self.vapi.cli("test echo server appns 0 fifo-size 4 no-echo" +
290 self.logger.critical(error)
291 self.assertEqual(error.find("failed"), -1)
293 error = self.vapi.cli("test echo client mbytes 10 appns 1 " +
294 "fifo-size 4 no-output test-bytes " +
295 "syn-timeout 2 no-return uri " + uri)
297 self.logger.critical(error)
298 self.assertEqual(error.find("failed"), -1)
300 # Delete inter-table routes
301 ip_t01.remove_vpp_config()
302 ip_t10.remove_vpp_config()
305 if __name__ == '__main__':
306 unittest.main(testRunner=VppTestRunner)