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         c = vpp_papi.VPPApiClient(apidir=".", testmode=True, use_socket=True)
  28
  29         # reset initialization at module load time.
  30         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  31         for _ in range(10):
  32             c.get_context()
  33         self.assertEqual(11, c.get_context())
  34
  35
  36 class TestVppPapiVPPApiClientMp(unittest.TestCase):
  37     # Test under multiple processes to simulate running forked under
  38     # run_tests.py (eg. make test TEST_JOBS=10)
  39
  40     def test_get_context_mp(self):
  41         c = vpp_papi.VPPApiClient(apidir=".", testmode=True, use_socket=True)
  42
  43         # reset initialization at module load time.
  44         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  45         procs = [mp.Process(target=c.get_context, args=()) for i in range(10)]
  46
  47         for p in procs:
  48             p.start()
  49         for p in procs:
  50             p.join()
  51
  52         # AssertionError: 11 != 1
  53         self.assertEqual(11, c.get_context())
  54
  55
  56 class TestVppTypes(unittest.TestCase):
  57     def test_enum_from_json(self):
  58         json_api = """\
  59 {
  60     "enums": [
  61
  62         [
  63             "address_family",
  64             [
  65                 "ADDRESS_IP4",
  66                 0
  67             ],
  68             [
  69                 "ADDRESS_IP6",
  70                 1
  71             ],
  72             {
  73                 "enumtype": "u8"
  74             }
  75         ],
  76         [
  77             "if_type",
  78             [
  79                 "IF_API_TYPE_HARDWARE",
  80                 0
  81             ],
  82             [
  83                 "IF_API_TYPE_SUB",
  84                 1
  85             ],
  86             [
  87                 "IF_API_TYPE_P2P",
  88                 2
  89             ],
  90             [
  91                 "IF_API_TYPE_PIPE",
  92                 3
  93             ],
  94             {
  95                 "enumtype": "u32"
  96             }
  97         ]
  98     ]
  99 }
 100 """
 101         processor = vpp_papi.VPPApiJSONFiles()
 102
 103         # add the types to vpp_serializer
 104         processor.process_json_str(json_api)
 105
 106         vpp_transport_shmem.VppTransport = mock.MagicMock()
 107         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 108         type_name = "vl_api_if_type_t"
 109         t = ac.get_type(type_name)
 110         self.assertTrue(str(t).startswith("VPPEnumType"))
 111         self.assertEqual(t.name, type_name)
 112
 113     def test_enumflagmixed_from_json(self):
 114         json_api = """\
 115 {
 116     "enums": [
 117
 118         [
 119             "address_family",
 120             [
 121                 "ADDRESS_IP4",
 122                 0
 123             ],
 124             [
 125                 "ADDRESS_IP6",
 126                 1
 127             ],
 128             {
 129                 "enumtype": "u8"
 130             }
 131         ]
 132         ],
 133     "enumflags": [
 134
 135         [
 136             "if_type",
 137             [
 138                 "IF_API_TYPE_HARDWARE",
 139                 0
 140             ],
 141             [
 142                 "IF_API_TYPE_SUB",
 143                 1
 144             ],
 145             [
 146                 "IF_API_TYPE_P2P",
 147                 2
 148             ],
 149             [
 150                 "IF_API_TYPE_PIPE",
 151                 3
 152             ],
 153             {
 154                 "enumtype": "u32"
 155             }
 156         ]
 157     ]
 158 }
 159 """
 160
 161         processor = vpp_papi.VPPApiJSONFiles()
 162
 163         # add the types to vpp_serializer
 164         processor.process_json_str(json_api)
 165
 166         vpp_transport_shmem.VppTransport = mock.MagicMock()
 167         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 168         print(ac)
 169         type_name = "vl_api_if_type_t"
 170         t = ac.get_type(type_name)
 171         print(t)
 172         self.assertTrue(str(t).startswith("VPPEnumType"))
 173         self.assertEqual(t.name, type_name)
 174
 175     def test_enumflag_from_json(self):
 176         json_api = """\
 177 {
 178     "enumflags": [
 179
 180         [
 181             "address_family",
 182             [
 183                 "ADDRESS_IP4",
 184                 0
 185             ],
 186             [
 187                 "ADDRESS_IP6",
 188                 1
 189             ],
 190             {
 191                 "enumtype": "u8"
 192             }
 193         ],
 194         [
 195             "if_type",
 196             [
 197                 "IF_API_TYPE_HARDWARE",
 198                 0
 199             ],
 200             [
 201                 "IF_API_TYPE_SUB",
 202                 1
 203             ],
 204             [
 205                 "IF_API_TYPE_P2P",
 206                 2
 207             ],
 208             [
 209                 "IF_API_TYPE_PIPE",
 210                 3
 211             ],
 212             {
 213                 "enumtype": "u32"
 214             }
 215         ]
 216     ]
 217 }
 218 """
 219         processor = vpp_papi.VPPApiJSONFiles()
 220
 221         # add the types to vpp_serializer
 222         processor.process_json_str(json_api)
 223
 224         vpp_transport_shmem.VppTransport = mock.MagicMock()
 225         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 226         type_name = "vl_api_if_type_t"
 227         t = ac.get_type(type_name)
 228         self.assertTrue(str(t).startswith("VPPEnumType"))
 229         self.assertEqual(t.name, type_name)
 230
 231
 232 class TestVppPapiLogging(unittest.TestCase):
 233     def test_logger(self):
 234         class Transport:
 235             connected = True
 236
 237         class Vpp:
 238             transport = Transport()
 239
 240             def disconnect(self):
 241                 pass
 242
 243         client = Vpp
 244         with self.assertLogs("vpp_papi", level="DEBUG") as cm:
 245             vpp_papi.vpp_atexit(client)
 246         self.assertEqual(cm.output, ["DEBUG:vpp_papi:Cleaning up VPP on exit"])
 247
 248         with self.assertRaises(AssertionError):
 249             with self.assertLogs("vpp_papi.serializer", level="DEBUG") as cm:
 250                 vpp_papi.vpp_atexit(client)
 251         self.assertEqual(cm.output, [])