src/vpp-api/python/vpp_papi/tests/test_vpp_papi.py

   1 #  Copyright (c) 2019. Vinci Consulting Corp. All Rights Reserved.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 import ctypes
  16 import multiprocessing as mp
  17 import sys
  18 import unittest
  19 from unittest import mock
  20
  21 from vpp_papi import vpp_papi
  22 from vpp_papi import vpp_transport_shmem
  23
  24
  25 class TestVppPapiVPPApiClient(unittest.TestCase):
  26     def test_getcontext(self):
  27         vpp_papi.VPPApiClient.apidir = '.'
  28         c = vpp_papi.VPPApiClient(testmode=True, use_socket=True)
  29
  30         # reset initialization at module load time.
  31         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  32         for _ in range(10):
  33             c.get_context()
  34         self.assertEqual(11, c.get_context())
  35
  36
  37 class TestVppPapiVPPApiClientMp(unittest.TestCase):
  38     # Test under multiple processes to simulate running forked under
  39     # run_tests.py (eg. make test TEST_JOBS=10)
  40
  41     def test_get_context_mp(self):
  42         vpp_papi.VPPApiClient.apidir = '.'
  43         c = vpp_papi.VPPApiClient(testmode=True, use_socket=True)
  44
  45         # reset initialization at module load time.
  46         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  47         procs = [mp.Process(target=c.get_context, args=()) for i in range(10)]
  48
  49         for p in procs:
  50             p.start()
  51         for p in procs:
  52             p.join()
  53
  54         # AssertionError: 11 != 1
  55         self.assertEqual(11, c.get_context())
  56
  57
  58 class TestVppTypes(unittest.TestCase):
  59     def test_enum_from_json(self):
  60         json_api = """\
  61 {
  62     "enums": [
  63
  64         [
  65             "address_family",
  66             [
  67                 "ADDRESS_IP4",
  68                 0
  69             ],
  70             [
  71                 "ADDRESS_IP6",
  72                 1
  73             ],
  74             {
  75                 "enumtype": "u8"
  76             }
  77         ],
  78         [
  79             "if_type",
  80             [
  81                 "IF_API_TYPE_HARDWARE",
  82                 0
  83             ],
  84             [
  85                 "IF_API_TYPE_SUB",
  86                 1
  87             ],
  88             [
  89                 "IF_API_TYPE_P2P",
  90                 2
  91             ],
  92             [
  93                 "IF_API_TYPE_PIPE",
  94                 3
  95             ],
  96             {
  97                 "enumtype": "u32"
  98             }
  99         ]
 100     ]
 101 }
 102 """
 103         processor = vpp_papi.VPPApiJSONFiles()
 104
 105         # add the types to vpp_serializer
 106         processor.process_json_str(json_api)
 107
 108         vpp_transport_shmem.VppTransport = mock.MagicMock()
 109         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 110         type_name = "vl_api_if_type_t"
 111         t = ac.get_type(type_name)
 112         self.assertTrue(str(t).startswith("VPPEnumType"))
 113         self.assertEqual(t.name, type_name)
 114
 115     def test_enumflagmixed_from_json(self):
 116         json_api = """\
 117 {
 118     "enums": [
 119
 120         [
 121             "address_family",
 122             [
 123                 "ADDRESS_IP4",
 124                 0
 125             ],
 126             [
 127                 "ADDRESS_IP6",
 128                 1
 129             ],
 130             {
 131                 "enumtype": "u8"
 132             }
 133         ]
 134         ],
 135     "enumflags": [
 136
 137         [
 138             "if_type",
 139             [
 140                 "IF_API_TYPE_HARDWARE",
 141                 0
 142             ],
 143             [
 144                 "IF_API_TYPE_SUB",
 145                 1
 146             ],
 147             [
 148                 "IF_API_TYPE_P2P",
 149                 2
 150             ],
 151             [
 152                 "IF_API_TYPE_PIPE",
 153                 3
 154             ],
 155             {
 156                 "enumtype": "u32"
 157             }
 158         ]
 159     ]
 160 }
 161 """
 162
 163         processor = vpp_papi.VPPApiJSONFiles()
 164
 165         # add the types to vpp_serializer
 166         processor.process_json_str(json_api)
 167
 168         vpp_transport_shmem.VppTransport = mock.MagicMock()
 169         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 170         print(ac)
 171         type_name = "vl_api_if_type_t"
 172         t = ac.get_type(type_name)
 173         print(t)
 174         self.assertTrue(str(t).startswith("VPPEnumType"))
 175         self.assertEqual(t.name, type_name)
 176
 177     def test_enumflag_from_json(self):
 178         json_api = """\
 179 {
 180     "enumflags": [
 181
 182         [
 183             "address_family",
 184             [
 185                 "ADDRESS_IP4",
 186                 0
 187             ],
 188             [
 189                 "ADDRESS_IP6",
 190                 1
 191             ],
 192             {
 193                 "enumtype": "u8"
 194             }
 195         ],
 196         [
 197             "if_type",
 198             [
 199                 "IF_API_TYPE_HARDWARE",
 200                 0
 201             ],
 202             [
 203                 "IF_API_TYPE_SUB",
 204                 1
 205             ],
 206             [
 207                 "IF_API_TYPE_P2P",
 208                 2
 209             ],
 210             [
 211                 "IF_API_TYPE_PIPE",
 212                 3
 213             ],
 214             {
 215                 "enumtype": "u32"
 216             }
 217         ]
 218     ]
 219 }
 220 """
 221         processor = vpp_papi.VPPApiJSONFiles()
 222
 223         # add the types to vpp_serializer
 224         processor.process_json_str(json_api)
 225
 226         vpp_transport_shmem.VppTransport = mock.MagicMock()
 227         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 228         type_name = "vl_api_if_type_t"
 229         t = ac.get_type(type_name)
 230         self.assertTrue(str(t).startswith("VPPEnumType"))
 231         self.assertEqual(t.name, type_name)
 232
 233
 234 class TestVppPapiLogging(unittest.TestCase):
 235     def test_logger(self):
 236         class Transport:
 237             connected = True
 238
 239         class Vpp:
 240             transport = Transport()
 241
 242             def disconnect(self):
 243                 pass
 244
 245         client = Vpp
 246         with self.assertLogs('vpp_papi', level='DEBUG') as cm:
 247             vpp_papi.vpp_atexit(client)
 248         self.assertEqual(cm.output, ['DEBUG:vpp_papi:Cleaning up VPP on exit'])
 249
 250         with self.assertRaises(AssertionError):
 251             with self.assertLogs('vpp_papi.serializer', level='DEBUG') as cm:
 252                 vpp_papi.vpp_atexit(client)
 253         self.assertEqual(cm.output, [])