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): """ show version """ rv = self.v.show_version() self.assertEqual(rv.retval, 0) def test_show_version_invalid_param(self): """ show version - invalid parameters""" self.assertRaises(ValueError, self.v.show_version, foobar='foo') def test_u8_array(self): """ u8 array """ 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): """ Add new message object """ 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): """ New message with array """ 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): """ Message to byte encoding """ 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): """ Add new types "
#ifndef _NODE_IN_H_
#define _NODE_IN_H_

typedef enum {
  ACL_IN_ERROR_DROP,
  ACL_IN_ETHERNET_INPUT,
  ACL_IN_L2S_INPUT_IP4_ADD,
  ACL_IN_L2S_INPUT_IP6_ADD,
  ACL_IN_N_NEXT,
} acl_in_next_t;

#endif
32", "address"], ["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): """ VLA with aribtrary length field placement """ 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): """ Argument name """ 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)