test/test_memif.py

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