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
  27     def test_getcontext(self):
  28         vpp_papi.VPPApiClient.apidir = '.'
  29         c = vpp_papi.VPPApiClient(testmode=True, use_socket=True)
  30
  31         # reset initialization at module load time.
  32         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  33         for _ in range(10):
  34             c.get_context()
  35         self.assertEqual(11, c.get_context())
  36
  37
  38 class TestVppPapiVPPApiClientMp(unittest.TestCase):
  39     # Test under multiple processes to simulate running forked under
  40     # run_tests.py (eg. make test TEST_JOBS=10)
  41
  42     def test_get_context_mp(self):
  43         vpp_papi.VPPApiClient.apidir = '.'
  44         c = vpp_papi.VPPApiClient(testmode=True, use_socket=True)
  45
  46         # reset initialization at module load time.
  47         c.get_context.context = mp.Value(ctypes.c_uint, 0)
  48         procs = [mp.Process(target=c.get_context, args=()) for i in range(10)]
  49
  50         for p in procs:
  51             p.start()
  52         for p in procs:
  53             p.join()
  54
  55         # AssertionError: 11 != 1
  56         self.assertEqual(11, c.get_context())
  57
  58
  59 class TestVppTypes(unittest.TestCase):
  60
  61     def test_enum_from_json(self):
  62         json_api = """\
  63 {
  64     "enums": [
  65
  66         [
  67             "address_family",
  68             [
  69                 "ADDRESS_IP4",
  70                 0
  71             ],
  72             [
  73                 "ADDRESS_IP6",
  74                 1
  75             ],
  76             {
  77                 "enumtype": "u8"
  78             }
  79         ],
  80         [
  81             "if_type",
  82             [
  83                 "IF_API_TYPE_HARDWARE",
  84                 0
  85             ],
  86             [
  87                 "IF_API_TYPE_SUB",
  88                 1
  89             ],
  90             [
  91                 "IF_API_TYPE_P2P",
  92                 2
  93             ],
  94             [
  95                 "IF_API_TYPE_PIPE",
  96                 3
  97             ],
  98             {
  99                 "enumtype": "u32"
 100             }
 101         ]
 102     ]
 103 }
 104 """
 105         processor = vpp_papi.VPPApiJSONFiles()
 106
 107         # add the types to vpp_serializer
 108         processor.process_json_str(json_api)
 109
 110         vpp_transport_shmem.VppTransport = mock.MagicMock()
 111         ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
 112         type_name = "vl_api_if_type_t"
 113         t = ac.get_type(type_name)
 114         self.assertTrue(str(t).startswith("VPPEnumType"))
 115         self.assertEqual(t.name, type_name)
 116