test/test_memif.py

   1 import socket
   2 import unittest
   3
   4 from scapy.layers.l2 import Ether
   5 from scapy.layers.inet import IP, ICMP
   6
   7 from framework import VppTestCase, VppTestRunner, running_extended_tests
   8 from remote_test import RemoteClass, RemoteVppTestCase
   9 from vpp_memif import MEMIF_MODE, MEMIF_ROLE, remove_all_memif_vpp_config, \
  10     VppSocketFilename, VppMemif
  11
  12
  13 class TestMemif(VppTestCase):
  14     """ Memif Test Case """
  15
  16     @classmethod
  17     def setUpClass(cls):
  18         # fork new process before client connects to VPP
  19         cls.remote_test = RemoteClass(RemoteVppTestCase)
  20         cls.remote_test.start_remote()
  21         cls.remote_test.set_request_timeout(10)
  22         super(TestMemif, cls).setUpClass()
  23         cls.remote_test.setUpClass(cls.tempdir)
  24         cls.create_pg_interfaces(range(1))
  25         for pg in cls.pg_interfaces:
  26             pg.config_ip4()
  27             pg.admin_up()
  28             pg.resolve_arp()
  29
  30     @classmethod
  31     def tearDownClass(cls):
  32         cls.remote_test.tearDownClass()
  33         cls.remote_test.quit_remote()
  34         for pg in cls.pg_interfaces:
  35             pg.unconfig_ip4()
  36             pg.set_table_ip4(0)
  37             pg.admin_down()
  38         super(TestMemif, cls).tearDownClass()
  39
  40     def tearDown(self):
  41         remove_all_memif_vpp_config(self)
  42         remove_all_memif_vpp_config(self.remote_test)
  43         super(TestMemif, self).tearDown()
  44
  45     def _check_socket_filename(self, dump, socket_id, filename):
  46         for d in dump:
  47             if (d.socket_id == socket_id) and (
  48                     d.socket_filename.rstrip("\0") == filename):
  49                 return True
  50         return False
  51
  52     def test_memif_socket_filename_add_del(self):
  53         """ Memif socket filenale add/del """
  54
  55         # dump default socket filename
  56         dump = self.vapi.memif_socket_filename_dump()
  57         self.assertTrue(
  58             self._check_socket_filename(
  59                 dump, 0, self.tempdir + "/memif.sock"))
  60
  61         memif_sockets = []
  62         # existing path
  63         memif_sockets.append(
  64             VppSocketFilename(
  65                 self, 1, self.tempdir + "/memif1.sock"))
  66         # default path (test tempdir)
  67         memif_sockets.append(
  68             VppSocketFilename(
  69                 self,
  70                 2,
  71                 "memif2.sock",
  72                 add_default_folder=True))
  73         # create new folder in default folder
  74         memif_sockets.append(
  75             VppSocketFilename(
  76                 self,
  77                 3,
  78                 "sock/memif3.sock",
  79                 add_default_folder=True))
  80
  81         for sock in memif_sockets:
  82             sock.add_vpp_config()
  83             dump = sock.query_vpp_config()
  84             self.assertTrue(
  85                 self._check_socket_filename(
  86                     dump,
  87                     sock.socket_id,
  88                     sock.socket_filename))
  89
  90         for sock in memif_sockets:
  91             sock.remove_vpp_config()
  92
  93         dump = self.vapi.memif_socket_filename_dump()
  94         self.assertTrue(
  95             self._check_socket_filename(
  96                 dump, 0, self.tempdir + "/memif.sock"))
  97
  98     def _create_delete_test_one_interface(self, memif):
  99         memif.add_vpp_config()
 100
 101         dump = memif.query_vpp_config()
 102
 103         self.assertTrue(dump)
 104         self.assertEqual(dump.sw_if_index, memif.sw_if_index)
 105         self.assertEqual(dump.role, memif.role)
 106         self.assertEqual(dump.mode, memif.mode)
 107         if (memif.socket_id is not None):
 108             self.assertEqual(dump.socket_id, memif.socket_id)
 109
 110         memif.remove_vpp_config()
 111
 112         dump = memif.query_vpp_config()
 113
 114         self.assertFalse(dump)
 115
 116     def _connect_test_one_interface(self, memif):
 117         self.assertTrue(memif.wait_for_link_up(5))
 118         dump = memif.query_vpp_config()
 119
 120         if memif.role == MEMIF_ROLE.SLAVE:
 121             self.assertEqual(dump.ring_size, memif.ring_size)
 122             self.assertEqual(dump.buffer_size, memif.buffer_size)
 123         else:
 124             self.assertEqual(dump.ring_size, 1)
 125             self.assertEqual(dump.buffer_size, 0)
 126
 127     def _connect_test_interface_pair(self, memif0, memif1):
 128         memif0.add_vpp_config()
 129         memif1.add_vpp_config()
 130
 131         memif0.admin_up()
 132         memif1.admin_up()
 133
 134         self._connect_test_one_interface(memif0)
 135         self._connect_test_one_interface(memif1)
 136
 137         memif0.remove_vpp_config()
 138         memif1.remove_vpp_config()
 139
 140     def test_memif_create_delete(self):
 141         """ Memif create/delete interface """
 142
 143         memif = VppMemif(self, MEMIF_ROLE.SLAVE, MEMIF_MODE.ETHERNET)
 144         self._create_delete_test_one_interface(memif)
 145         memif.role = MEMIF_ROLE.MASTER
 146         self._create_delete_test_one_interface(memif)
 147
 148     def test_memif_create_custom_socket(self):
 149         """ Memif create with non-default socket filname """
 150
 151         memif_sockets = []
 152         # existing path
 153         memif_sockets.append(
 154             VppSocketFilename(
 155                 self, 1, self.tempdir + "/memif1.sock"))
 156         # default path (test tempdir)
 157         memif_sockets.append(
 158             VppSocketFilename(
 159                 self,
 160                 2,
 161                 "memif2.sock",
 162                 add_default_folder=True))
 163         # create new folder in default folder
 164         memif_sockets.append(
 165             VppSocketFilename(
 166                 self,
 167                 3,
 168                 "sock/memif3.sock",
 169                 add_default_folder=True))
 170
 171         memif = VppMemif(self, MEMIF_ROLE.SLAVE, MEMIF_MODE.ETHERNET)
 172
 173         for sock in memif_sockets:
 174             sock.add_vpp_config()
 175             memif.socket_id = sock.socket_id
 176             memif.role = MEMIF_ROLE.SLAVE
 177             self._create_delete_test_one_interface(memif)
 178             memif.role = MEMIF_ROLE.MASTER
 179             self._create_delete_test_one_interface(memif)
 180
 181     def test_memif_connect(self):
 182         """ Memif connect """
 183         memif = VppMemif(self, MEMIF_ROLE.SLAVE,  MEMIF_MODE.ETHERNET,
 184                          ring_size=1024, buffer_size=2048)
 185
 186         remote_socket = VppSocketFilename(self.remote_test, 1,
 187                                           self.tempdir + "/memif.sock")
 188         remote_socket.add_vpp_config()
 189
 190         remote_memif = VppMemif(self.remote_test, MEMIF_ROLE.MASTER,
 191                                 MEMIF_MODE.ETHERNET, socket_id=1,
 192                                 ring_size=1024, buffer_size=2048)
 193
 194         self._connect_test_interface_pair(memif, remote_memif)
 195
 196         memif.role = MEMIF_ROLE.MASTER
 197         remote_memif.role = MEMIF_ROLE.SLAVE
 198
 199         self._connect_test_interface_pair(memif, remote_memif)
 200
 201     def _create_icmp(self, pg, memif, num):
 202         pkts = []
 203         for i in range(num):
 204             pkt = (Ether(dst=pg.local_mac, src=pg.remote_mac) /
 205                    IP(src=pg.remote_ip4, dst=memif.ip4_addr) /
 206                    ICMP(id=memif.if_id, type='echo-request', seq=i))
 207             pkts.append(pkt)
 208         return pkts
 209
 210     def _verify_icmp(self, pg, memif, rx, seq):
 211         ip = rx[IP]
 212         self.assertEqual(ip.src, memif.ip4_addr)
 213         self.assertEqual(ip.dst, pg.remote_ip4)
 214         self.assertEqual(ip.proto, 1)
 215         icmp = rx[ICMP]
 216         self.assertEqual(icmp.type, 0)  # echo-reply
 217         self.assertEqual(icmp.id, memif.if_id)
 218         self.assertEqual(icmp.seq, seq)
 219
 220     def test_memif_ping(self):
 221         """ Memif ping """
 222         memif = VppMemif(self, MEMIF_ROLE.SLAVE,  MEMIF_MODE.ETHERNET)
 223
 224         remote_socket = VppSocketFilename(self.remote_test, 1,
 225                                           self.tempdir + "/memif.sock")
 226         remote_socket.add_vpp_config()
 227
 228         remote_memif = VppMemif(self.remote_test, MEMIF_ROLE.MASTER,
 229                                 MEMIF_MODE.ETHERNET, socket_id=1)
 230
 231         memif.add_vpp_config()
 232         memif.config_ip4()
 233         memif.admin_up()
 234
 235         remote_memif.add_vpp_config()
 236         remote_memif.config_ip4()
 237         remote_memif.admin_up()
 238
 239         self.assertTrue(memif.wait_for_link_up(5))
 240         self.assertTrue(remote_memif.wait_for_link_up(5))
 241
 242         # add routing to remote vpp
 243         dst_addr = socket.inet_pton(socket.AF_INET, self.pg0._local_ip4_subnet)
 244         dst_addr_len = 24
 245         next_hop_addr = socket.inet_pton(socket.AF_INET, memif.ip4_addr)
 246         self.remote_test.vapi.ip_add_del_route(
 247             dst_addr, dst_addr_len, next_hop_addr)
 248
 249         # create ICMP echo-request from local pg to remote memif
 250         packet_num = 10
 251         pkts = self._create_icmp(self.pg0, remote_memif, packet_num)
 252
 253         self.pg0.add_stream(pkts)
 254         self.pg_enable_capture(self.pg_interfaces)
 255         self.pg_start()
 256         capture = self.pg0.get_capture(packet_num, timeout=2)
 257         seq = 0
 258         for c in capture:
 259             self._verify_icmp(self.pg0, remote_memif, c, seq)
 260             seq += 1
 261
 262
 263 if __name__ == '__main__':
 264     unittest.main(testRunner=VppTestRunner)