import binascii from framework import VppTestCase from vpp_papi import VPP import json """ TestPAPI is a subclass of VPPTestCase classes. Basic test for sanity check of the Python API binding. """ class TestPAPI(VppTestCase): """ PAPI Test Case """ @classmethod def setUpClass(cls): super(TestPAPI, cls).setUpClass() cls.v = cls.vapi.papi def test_show_version(self): rv = self.v.show_version() self.assertEqual(rv.retval, 0) def test_show_version_invalid_param(self): self.assertRaises(ValueError, self.v.show_version, foobar='foo') def test_u8_array(self): rv = self.v.get_node_index(node_name='ip4-lookup') self.assertEqual(rv.retval, 0) node_name = 'X' * 100 self.assertRaises(ValueError, self.v.get_node_index, node_name=node_name) class TestPAPIMessageParsing(VppTestCase): """ PAPI Message parsing Test Case """ show_version_msg = '''["show_version", ["u16", "_vl_msg_id"], ["u32", "client_index"], ["u32", "context"], {"crc" : "0xf18f9480"} ]''' ip_address_details_msg = '''["ip_address_details", ["u16", "_vl_msg_id"], ["u32", "client_index"], ["u32", "context"], ["u8", "ip", 16], ["u8", "prefix_length"], {"crc" : "0x87d522a1"} ]''' cli_inband_msg = '''["cli_inband", ["u16", "_vl_msg_id"], ["u32", "client_index"], ["u32", "context"], ["u32", "length"], ["u8", "cmd", 0, "length"], {"crc" : "0x22345937"} ]''' def test_adding_new_message_object(self): p = json.loads(TestPAPIMessageParsing.show_version_msg) msglist = VPP(testmode=json) msgdef = msglist.add_message(p[0], p[1:]) # Verify that message can be retrieved self.assertTrue(msglist['show_version']) self.assertFalse(msglist['foobar']) # Test duplicate self.assertRaises(ValueError, msglist.add_message, p[0], p[1:]) # Look at return tuple self.assertTrue(msglist.ret_tup('show_version')) def test_adding_new_message_object_with_array(self): p = json.loads(TestPAPIMessageParsing.ip_address_details_msg) msglist = VPP(testmode=True) msglist.add_message(p[0], p[1:]) self.assertTrue(msglist['ip_address_details']) def test_message_to_bytes(self): msglist = VPP(testmode=True) p = json.loads(TestPAPIMessageParsing.show_version_msg) msgdef = msglist.add_message(p[0], p[1:]) # Give me a byte string for given message and given arguments b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123}) self.assertEqual(10, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(rv._0, 50) self.assertEqual(rv.context, 123) p = json.loads(TestPAPIMessageParsing.ip_address_details_msg) msgdef = msglist.add_message(p[0], p[1:]) # Give me a byte string for given message and given arguments b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123, 'ip': b'\xf0\xf1\xf2', 'prefix_length': 12}) self.assertEqual(27, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(rv.context, 123) self.assertEqual(rv.ip, b'\xf0\xf1\xf2\x00\x00\x00' + '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00') self.assertEqual(rv.prefix_length, 12) p = json.loads(TestPAPIMessageParsing.cli_inband_msg) msgdef = msglist.add_message(p[0], p[1:]) # Give me a byte string for given message and given arguments b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123, 'length': 20, 'cmd': 'show version verbose'}) self.assertEqual(34, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(rv._0, 50) self.assertEqual(rv.context, 123) self.assertEqual(rv.cmd.decode('ascii'), 'show version verbose') variable_array_16_msg = '''["variable_array_16", ["u32", "length"], ["u16", "list", 0, "length"] ]''' p = json.loads(variable_array_16_msg) msgdef = msglist.add_message(p[0], p[1:]) # Give me a byte string for given message and given arguments b = msglist.encode(msgdef, {'list': [1, 2], 'length': 2}) self.assertEqual(8, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(2, rv.length) self.assertEqual([1, 2], rv.list) def test_add_new_types(self): counter_type = '''["ip4_fib_counter", ["u32", "address"], ["u8", "address_length"], ["u64", "packets"], ["u64", "bytes"], {"crc" : "0xb2739495"} ]''' with_type_msg = '''["with_type_msg", ["u32", "length"], ["u16", "list", 0, "length"], ["vl_api_ip4_fib_counter_t", "counter"] ]''' # Add new type msglist = VPP(testmode=True) p = json.loads(counter_type) msglist.add_type(p[0], p[1:]) p = json.loads(with_type_msg) msgdef = msglist.add_message(p[0], p[1:]) b = msglist.encode(msgdef, {'length': 2, 'list': [1, 2], 'counter': {'address': 4, 'address_length': 12, 'packets': 1235, 'bytes': 5678}}) self.assertEqual(29, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(2, rv.length) self.assertEqual(5678, rv.counter.bytes) def test_add_two_new_types(self): mock_
# Copyright (c) 2018 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

##############################################################################
# Check for memfd_create headers and libs
##############################################################################
check_c_source_compiles("
  #define _GNU_SOURCE
  #include <sys/mman.h>
  int main() { return memfd_create (\"/dev/false\", 0); }
" HAVE_MEMFD_CREATE)

if (HAVE_MEMFD_CREATE)
    add_definitions(-DHAVE_MEMFD_CREATE)
endif()

check_c_source_compiles("
  #define _GNU_SOURCE
  #include <sched.h>
  int main() { return getcpu (0, 0); }
" HAVE_GETCPU)

if (HAVE_GETCPU)
    add_definitions(-DHAVE_GETCPU)
endif()

check_c_source_compiles("
  #define _GNU_SOURCE
  #include <fcntl.h>
  int main() { return fcntl64 (0, 0); }
" HAVE_FCNTL64)

if (HAVE_FCNTL64)
    add_definitions(-DHAVE_FCNTL64)
endif()
s"], ["u8", "address_length"], ["u64", "packets"], ["u64", "bytes"], {"crc" : "0xb2739495"} ]''' vla_byte_array = '''["vla_byte_array", ["vl_api_ip4_fib_counter_t", "counter", 0], {"crc" : "0xb2739495"} ]''' p = json.loads(counter_type) msglist.add_type(p[0], p[1:]) p = json.loads(vla_byte_array) with self.assertRaises(NotImplementedError): msgdef = msglist.add_message(p[0], p[1:]) def test_array_count_not_previous(self): msglist = VPP(testmode=True) # VLA vla_byte_array = '''["vla_byte_array", ["u32", "count"], ["u32", "filler"], ["u32", "lst", 0, "count"], {"crc" : "0xb2739495"} ]''' p = json.loads(vla_byte_array) msgdef = msglist.add_message(p[0], p[1:]) b = msglist.encode(msgdef, {'count': 3, 'lst': [1, 2, 3], 'filler': 1}) rv = msglist.decode(msgdef, b) self.assertEqual(rv.lst, [1, 2, 3]) def test_argument_name(self): msglist = VPP(testmode=True) simple_name = '''["simple_name", ["u32", "length"], ["u8", "name"] ]''' p = json.loads(simple_name) msgdef = msglist.add_message(p[0], p[1:]) b = msglist.encode(msgdef, {'length': 6, 'name': 1}) self.assertEqual(5, len(b)) rv = msglist.decode(msgdef, b) self.assertEqual(6, rv.length) self.assertEqual(1, rv.name)