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