1 # Copyright (c) 2019. Vinci Consulting Corp. All Rights Reserved.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 import multiprocessing as mp
19 from unittest import mock
21 from vpp_papi import vpp_papi
22 from vpp_papi import vpp_transport_shmem
25 class TestVppPapiVPPApiClient(unittest.TestCase):
26 def test_getcontext(self):
27 c = vpp_papi.VPPApiClient(apidir=".", testmode=True, use_socket=True)
29 # reset initialization at module load time.
30 c.get_context.context = mp.Value(ctypes.c_uint, 0)
33 self.assertEqual(11, c.get_context())
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)
40 def test_get_context_mp(self):
41 c = vpp_papi.VPPApiClient(apidir=".", testmode=True, use_socket=True)
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)]
52 # AssertionError: 11 != 1
53 self.assertEqual(11, c.get_context())
56 class TestVppTypes(unittest.TestCase):
57 def test_enum_from_json(self):
79 "IF_API_TYPE_HARDWARE",
101 processor = vpp_papi.VPPApiJSONFiles()
103 # add the types to vpp_serializer
104 processor.process_json_str(json_api)
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)
113 def test_enumflagmixed_from_json(self):
138 "IF_API_TYPE_HARDWARE",
161 processor = vpp_papi.VPPApiJSONFiles()
163 # add the types to vpp_serializer
164 processor.process_json_str(json_api)
166 vpp_transport_shmem.VppTransport = mock.MagicMock()
167 ac = vpp_papi.VPPApiClient(apifiles=[], testmode=True)
169 type_name = "vl_api_if_type_t"
170 t = ac.get_type(type_name)
172 self.assertTrue(str(t).startswith("VPPEnumType"))
173 self.assertEqual(t.name, type_name)
175 def test_enumflag_from_json(self):
197 "IF_API_TYPE_HARDWARE",
219 processor = vpp_papi.VPPApiJSONFiles()
221 # add the types to vpp_serializer
222 processor.process_json_str(json_api)
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)
232 class TestVppPapiLogging(unittest.TestCase):
233 def test_logger(self):
238 transport = Transport()
240 def disconnect(self):
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"])
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, [])