test/test_udp.py

   1 #!/usr/bin/env python
   2 import unittest
   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
   6
   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
  12
  13
  14 class TestUdpEncap(VppTestCase):
  15     """ UDP Encap Test Case """
  16
  17     def setUp(self):
  18         super(TestUdpEncap, self).setUp()
  19
  20         # create 2 pg interfaces
  21         self.create_pg_interfaces(range(4))
  22
  23         # setup interfaces
  24         # assign them different tables.
  25         table_id = 0
  26         self.tables = []
  27
  28         for i in self.pg_interfaces:
  29             i.admin_up()
  30
  31             if table_id != 0:
  32                 tbl = VppIpTable(self, table_id)
  33                 tbl.add_vpp_config()
  34                 self.tables.append(tbl)
  35                 tbl = VppIpTable(self, table_id, is_ip6=1)
  36                 tbl.add_vpp_config()
  37                 self.tables.append(tbl)
  38
  39             i.set_table_ip4(table_id)
  40             i.set_table_ip6(table_id)
  41             i.config_ip4()
  42             i.resolve_arp()
  43             i.config_ip6()
  44             i.resolve_ndp()
  45             table_id += 1
  46
  47     def tearDown(self):
  48         for i in self.pg_interfaces:
  49             i.unconfig_ip4()
  50             i.unconfig_ip6()
  51             i.ip6_disable()
  52             i.set_table_ip4(0)
  53             i.set_table_ip6(0)
  54             i.admin_down()
  55         super(TestUdpEncap, self).tearDown()
  56
  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)
  62
  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)
  68
  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)
  72         if ttl:
  73             self.assertEqual(rx[IP].ttl, ttl)
  74         else:
  75             self.assertEqual(rx[IP].ttl, tx[IP].ttl)
  76
  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)
  81
  82     def test_udp_encap(self):
  83         """ UDP Encap test
  84         """
  85
  86         #
  87         # construct a UDP encap object through each of the peers
  88         # v4 through the first two peers, v6 through the second.
  89         #
  90         udp_encap_0 = VppUdpEncap(self,
  91                                   self.pg0.local_ip4,
  92                                   self.pg0.remote_ip4,
  93                                   330, 440)
  94         udp_encap_1 = VppUdpEncap(self,
  95                                   self.pg1.local_ip4,
  96                                   self.pg1.remote_ip4,
  97                                   331, 441,
  98                                   table_id=1)
  99         udp_encap_2 = VppUdpEncap(self,
 100                                   self.pg2.local_ip6,
 101                                   self.pg2.remote_ip6,
 102                                   332, 442,
 103                                   table_id=2)
 104         udp_encap_3 = VppUdpEncap(self,
 105                                   self.pg3.local_ip6,
 106                                   self.pg3.remote_ip6,
 107                                   333, 443,
 108                                   table_id=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()
 113
 114         self.logger.info(self.vapi.cli("sh udp encap"))
 115
 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))
 120
 121         #
 122         # Routes via each UDP encap object - all combinations of v4 and v6.
 123         #
 124         route_4o4 = VppIpRoute(self, "1.1.0.1", 32,
 125                                [VppRoutePath("0.0.0.0",
 126                                              0xFFFFFFFF,
 127                                              is_udp_encap=1,
 128                                              next_hop_id=udp_encap_0.id)])
 129         route_4o6 = VppIpRoute(self, "1.1.2.1", 32,
 130                                [VppRoutePath("0.0.0.0",
 131                                              0xFFFFFFFF,
 132                                              is_udp_encap=1,
 133                                              next_hop_id=udp_encap_2.id)])
 134         route_6o4 = VppIpRoute(self, "2001::1", 128,
 135                                [VppRoutePath("0.0.0.0",
 136                                              0xFFFFFFFF,
 137                                              is_udp_encap=1,
 138                                              next_hop_id=udp_encap_1.id)],
 139                                is_ip6=1)
 140         route_6o6 = VppIpRoute(self, "2001::3", 128,
 141                                [VppRoutePath("0.0.0.0",
 142                                              0xFFFFFFFF,
 143                                              is_udp_encap=1,
 144                                              next_hop_id=udp_encap_3.id)],
 145                                is_ip6=1)
 146         route_4o4.add_vpp_config()
 147         route_4o6.add_vpp_config()
 148         route_6o6.add_vpp_config()
 149         route_6o4.add_vpp_config()
 150
 151         #
 152         # 4o4 encap
 153         #
 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) /
 158                  Raw('\xa5' * 100))
 159         rx = self.send_and_expect(self.pg0, p_4o4*65, self.pg0)
 160         for p in rx:
 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)
 165
 166         #
 167         # 4o6 encap
 168         #
 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) /
 173                  Raw('\xa5' * 100))
 174         rx = self.send_and_expect(self.pg0, p_4o6*65, self.pg2)
 175         for p in rx:
 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)
 180
 181         #
 182         # 6o4 encap
 183         #
 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) /
 188                  Raw('\xa5' * 100))
 189         rx = self.send_and_expect(self.pg0, p_6o4*65, self.pg1)
 190         for p in rx:
 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)
 195
 196         #
 197         # 6o6 encap
 198         #
 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) /
 203                  Raw('\xa5' * 100))
 204         rx = self.send_and_expect(self.pg0, p_6o6*65, self.pg3)
 205         for p in rx:
 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)
 210
 211         #
 212         # A route with an output label
 213         # the TTL of the inner packet is decremented on LSP ingress
 214         #
 215         route_4oMPLSo4 = VppIpRoute(self, "1.1.2.22", 32,
 216                                     [VppRoutePath("0.0.0.0",
 217                                                   0xFFFFFFFF,
 218                                                   is_udp_encap=1,
 219                                                   next_hop_id=1,
 220                                                   labels=[VppMplsLabel(66)])])
 221         route_4oMPLSo4.add_vpp_config()
 222
 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) /
 227                    Raw('\xa5' * 100))
 228         rx = self.send_and_expect(self.pg0, p_4omo4*65, self.pg1)
 229         for p in rx:
 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)
 234
 235
 236 class TestUDP(VppTestCase):
 237     """ UDP Test Case """
 238
 239     @classmethod
 240     def setUpClass(cls):
 241         super(TestUDP, cls).setUpClass()
 242
 243     def setUp(self):
 244         super(TestUDP, self).setUp()
 245         self.vapi.session_enable_disable(is_enabled=1)
 246         self.create_loopback_interfaces(2)
 247
 248         table_id = 0
 249
 250         for i in self.lo_interfaces:
 251             i.admin_up()
 252
 253             if table_id != 0:
 254                 tbl = VppIpTable(self, table_id)
 255                 tbl.add_vpp_config()
 256
 257             i.set_table_ip4(table_id)
 258             i.config_ip4()
 259             table_id += 1
 260
 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)
 266
 267     def tearDown(self):
 268         for i in self.lo_interfaces:
 269             i.unconfig_ip4()
 270             i.set_table_ip4(0)
 271             i.admin_down()
 272         self.vapi.session_enable_disable(is_enabled=0)
 273         super(TestUDP, self).tearDown()
 274
 275     def test_udp_transfer(self):
 276         """ UDP echo client/server transfer """
 277
 278         # Add inter-table routes
 279         ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32,
 280                             [VppRoutePath("0.0.0.0",
 281                                           0xffffffff,
 282                                           nh_table_id=1)])
 283         ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32,
 284                             [VppRoutePath("0.0.0.0",
 285                                           0xffffffff,
 286                                           nh_table_id=0)], table_id=1)
 287         ip_t01.add_vpp_config()
 288         ip_t10.add_vpp_config()
 289
 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" +
 293                               "uri " + uri)
 294         if error:
 295             self.logger.critical(error)
 296             self.assertNotIn("failed", error)
 297
 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)
 301         if error:
 302             self.logger.critical(error)
 303             self.assertNotIn("failed", error)
 304
 305         # Delete inter-table routes
 306         ip_t01.remove_vpp_config()
 307         ip_t10.remove_vpp_config()
 308
 309
 310 if __name__ == '__main__':
 311     unittest.main(testRunner=VppTestRunner)