test/test_papi.py

   1 import binascii
   2 from framework import VppTestCase
   3 from vpp_papi import VPP
   4 import json
   5
   6 """ TestPAPI is a subclass of  VPPTestCase classes.
   7
   8 Basic test for sanity check of the Python API binding.
   9
  10 """
  11
  12
  13 class TestPAPI(VppTestCase):
  14     """ PAPI Test Case """
  15
  16     @classmethod
  17     def setUpClass(cls):
  18         super(TestPAPI, cls).setUpClass()
  19         cls.v = cls.vapi.papi
  20
  21     def test_show_version(self):
  22         """ show version """
  23         rv = self.v.show_version()
  24         self.assertEqual(rv.retval, 0)
  25
  26     def test_show_version_invalid_param(self):
  27         """ show version - invalid parameters"""
  28         self.assertRaises(ValueError, self.v.show_version, foobar='foo')
  29
  30     def test_u8_array(self):
  31         """ u8 array """
  32         rv = self.v.get_node_index(node_name='ip4-lookup')
  33         self.assertEqual(rv.retval, 0)
  34         node_name = 'X' * 100
  35         self.assertRaises(ValueError, self.v.get_node_index,
  36                           node_name=node_name)
  37
  38
  39 class TestPAPIMessageParsing(VppTestCase):
  40     """ PAPI Message parsing Test Case """
  41
  42     show_version_msg = '''["show_version",
  43               ["u16", "_vl_msg_id"],
  44               ["u32", "client_index"],
  45               ["u32", "context"],
  46               {"crc" : "0xf18f9480"}
  47         ]'''
  48
  49     ip_address_details_msg = '''["ip_address_details",
  50             ["u16", "_vl_msg_id"],
  51             ["u32", "client_index"],
  52             ["u32", "context"],
  53             ["u8", "ip", 16],
  54             ["u8", "prefix_length"],
  55             {"crc" : "0x87d522a1"}
  56         ]'''
  57
  58     cli_inband_msg = '''["cli_inband",
  59             ["u16", "_vl_msg_id"],
  60             ["u32", "client_index"],
  61             ["u32", "context"],
  62             ["u32", "length"],
  63             ["u8", "cmd", 0, "length"],
  64             {"crc" : "0x22345937"}
  65         ]'''
  66
  67     def test_adding_new_message_object(self):
  68         """ Add new message object """
  69         p = json.loads(TestPAPIMessageParsing.show_version_msg)
  70         msglist = VPP(testmode=json)
  71         msgdef = msglist.add_message(p[0], p[1:])
  72
  73         # Verify that message can be retrieved
  74         self.assertTrue(msglist['show_version'])
  75         self.assertFalse(msglist['foobar'])
  76
  77         # Test duplicate
  78         self.assertRaises(ValueError, msglist.add_message, p[0], p[1:])
  79
  80         # Look at return tuple
  81         self.assertTrue(msglist.ret_tup('show_version'))
  82
  83     def test_adding_new_message_object_with_array(self):
  84         """ New message with array """
  85         p = json.loads(TestPAPIMessageParsing.ip_address_details_msg)
  86         msglist = VPP(testmode=True)
  87         msglist.add_message(p[0], p[1:])
  88
  89         self.assertTrue(msglist['ip_address_details'])
  90
  91     def test_message_to_bytes(self):
  92         """ Message to byte encoding """
  93         msglist = VPP(testmode=True)
  94         p = json.loads(TestPAPIMessageParsing.show_version_msg)
  95         msgdef = msglist.add_message(p[0], p[1:])
  96
  97         # Give me a byte string for given message and given arguments
  98
  99         b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123})
 100         self.assertEqual(10, len(b))
 101         rv = msglist.decode(msgdef, b)
 102         self.assertEqual(rv._0, 50)
 103         self.assertEqual(rv.context, 123)
 104
 105         p = json.loads(TestPAPIMessageParsing.ip_address_details_msg)
 106         msgdef = msglist.add_message(p[0], p[1:])
 107
 108         # Give me a byte string for given message and given arguments
 109         b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123,
 110                                     'ip': b'\xf0\xf1\xf2',
 111                                     'prefix_length': 12})
 112         self.assertEqual(27, len(b))
 113         rv = msglist.decode(msgdef, b)
 114
 115         self.assertEqual(rv.context, 123)
 116         self.assertEqual(rv.ip, b'\xf0\xf1\xf2\x00\x00\x00' +
 117                          '\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
 118         self.assertEqual(rv.prefix_length, 12)
 119
 120         p = json.loads(TestPAPIMessageParsing.cli_inband_msg)
 121         msgdef = msglist.add_message(p[0], p[1:])
 122
 123         # Give me a byte string for given message and given arguments
 124         b = msglist.encode(msgdef, {'_vl_msg_id': 50, 'context': 123,
 125                                     'length': 20,
 126                                     'cmd': 'show version verbose'})
 127         self.assertEqual(34, len(b))
 128         rv = msglist.decode(msgdef, b)
 129         self.assertEqual(rv._0, 50)
 130         self.assertEqual(rv.context, 123)
 131         self.assertEqual(rv.cmd.decode('ascii'), 'show version verbose')
 132
 133         variable_array_16_msg = '''["variable_array_16",
 134             ["u32", "length"],
 135             ["u16", "list", 0, "length"]
 136         ]'''
 137
 138         p = json.loads(variable_array_16_msg)
 139         msgdef = msglist.add_message(p[0], p[1:])
 140
 141         # Give me a byte string for given message and given arguments
 142         b = msglist.encode(msgdef, {'list': [1, 2], 'length': 2})
 143         self.assertEqual(8, len(b))
 144         rv = msglist.decode(msgdef, b)
 145         self.assertEqual(2, rv.length)
 146         self.assertEqual([1, 2], rv.list)
 147
 148     def test_add_new_types(self):
 149         """ Add new types """
 150         counter_type = '''["ip4_fib_counter",
 151             ["u32", "address"],
 152             ["u8", "address_length"],
 153             ["u64", "packets"],
 154             ["u64", "bytes"],
 155             {"crc" : "0xb2739495"}
 156         ]'''
 157
 158         with_type_msg = '''["with_type_msg",
 159             ["u32", "length"],
 160             ["u16", "list", 0, "length"],
 161             ["vl_api_ip4_fib_counter_t", "counter"]
 162         ]'''
 163
 164         # Add new type
 165         msglist = VPP(testmode=True)
 166         p = json.loads(counter_type)
 167         msglist.add_type(p[0], p[1:])
 168         p = json.loads(with_type_msg)
 169         msgdef = msglist.add_message(p[0], p[1:])
 170         b = msglist.encode(msgdef, {'length': 2, 'list': [1, 2],
 171                                     'counter': {'address': 4,
 172                                                 'address_length': 12,
 173                                                 'packets': 1235,
 174                                                 'bytes': 5678}})
 175         self.assertEqual(29, len(b))
 176         rv = msglist.decode(msgdef, b)
 177         self.assertEqual(2, rv.length)
 178         self.assertEqual(5678, rv.counter.bytes)
 179
 180     def test_add_two_new_types(self):
 181         """ Add new types 2 """
 182         mock_r1 = '''["mock_r1",
 183             ["u32", "a1"],
 184             {"crc" : "0xb2739495"}
 185         ]'''
 186         mock_r2 = '''["mock_r2",
 187             ["u32", "a1"],
 188             {"crc" : "0xb2739495"}
 189         ]'''
 190
 191         mock_msg = '''["mock_msg",
 192             ["u32", "context"],
 193             ["i32", "retval"],
 194             ["vl_api_mock_r1_t", "r1"],
 195             ["vl_api_mock_r2_t", "r2"],
 196             {"crc" : "0xb2739495"}
 197         ]'''
 198
 199         # Add new type
 200         msglist = VPP(testmode=True)
 201         p = json.loads(mock_r1)
 202         msglist.add_type(p[0], p[1:])
 203         p = json.loads(mock_r2)
 204         msglist.add_type(p[0], p[1:])
 205         p = json.loads(mock_msg)
 206         msgdef = msglist.add_message(p[0], p[1:])
 207         b = msglist.encode(msgdef, {'context': 2, 'retval': 0,
 208                                     'r1': {'a1': 4}, 'r2': {'a1': 12}})
 209
 210         self.assertEqual(16, len(b))
 211         rv = msglist.decode(msgdef, b)
 212         self.assertEqual(4, rv.r1.a1)
 213
 214     def test_nested_array_type(self):
 215         """ Nested array type """
 216         bier_type = '''["bier_table_id",
 217             ["u8", "bt_set"],
 218             ["u8", "bt_sub_domain"],
 219             ["u8", "bt_hdr_len_id"],
 220             {"crc" : "0xb2739495"}
 221         ]'''
 222         fib_path3 = '''["fib_path3",
 223             ["u32", "sw_if_index"],
 224             ["u8", "n_labels"],
 225             ["u32", "label_stack", 0, "n_labels"],
 226             {"crc" : "0xb2739495"}
 227         ]'''
 228
 229         bier_route_details = '''["bier_route_details",
 230             ["u32", "client_index"],
 231             ["vl_api_bier_table_id_t", "br_tbl_id"],
 232             ["u32", "br_n_paths"],
 233             ["vl_api_fib_path3_t", "br_paths", 0, "br_n_paths"],
 234             {"crc" : "0xb2739495"}
 235         ]'''
 236
 237         # Add new type
 238         msglist = VPP(testmode=True)
 239
 240         p = json.loads(bier_type)
 241         msglist.add_type(p[0], p[1:])
 242         p = json.loads(fib_path3)
 243         msglist.add_type(p[0], p[1:])
 244
 245         p = json.loads(bier_route_details)
 246         msgdef = msglist.add_message(p[0], p[1:])
 247
 248         bt_tbl_id = {'bt_set': 1, 'bt_sub_domain': 2, 'bt_hdr_len_id': 3}
 249         fib_path = {'sw_if_index': 1, 'n_labels': 2,
 250                     'label_stack': [123, 456]}
 251
 252         b = msglist.encode(msgdef, {'client_index': 2,
 253                                     'br_tbl_id': bt_tbl_id,
 254                                     'br_n_paths': 2,
 255                                     'br_paths': [fib_path, fib_path]})
 256         self.assertEqual(37, len(b))
 257         rv = msglist.decode(msgdef, b)
 258         self.assertEqual([123, 456], rv.br_paths[1].label_stack)
 259         self.assertEqual(bt_tbl_id['bt_set'], rv.br_tbl_id.bt_set)
 260
 261     def test_add_new_compound_type_with_array(self):
 262         """ New compound type with array """
 263         counter_type = '''["ip4_fib_counter",
 264             ["u32", "address"],
 265             ["u8", "address_length"],
 266             ["u64", "packets"],
 267             ["u64", "bytes"],
 268             {"crc" : "0xb2739495"}
 269         ]'''
 270
 271         with_type_msg = '''["with_type_msg",
 272             ["u32", "length"],
 273             ["u16", "list", 0, "length"],
 274             ["vl_api_ip4_fib_counter_t", "counter", 2]
 275
 276         ]'''
 277
 278         # Add new type
 279         msglist = VPP(testmode=True)
 280         p = json.loads(counter_type)
 281         msglist.add_type(p[0], p[1:])
 282         p = json.loads(with_type_msg)
 283         msgdef = msglist.add_message(p[0], p[1:])
 284         b = msglist.encode(msgdef, {'length': 2, 'list': [1, 2],
 285                                     'counter': [{'address': 4,
 286                                                  'address_length': 12,
 287                                                  'packets': 1235,
 288                                                  'bytes': 5678},
 289                                                 {'address': 111,
 290                                                  'address_length': 222,
 291                                                  'packets': 333,
 292                                                  'bytes': 444}]})
 293         self.assertEqual(50, len(b))
 294         rv = msglist.decode(msgdef, b)
 295         self.assertEqual([1, 2], rv.list)
 296         self.assertEqual(1235, rv.counter[0].packets)
 297
 298         with_type_variable_msg = '''["with_type_variable_msg",
 299             ["u32", "length"],
 300             ["vl_api_ip4_fib_counter_t", "counter", 0, "length"]
 301
 302         ]'''
 303
 304         p = json.loads(with_type_variable_msg)
 305         msgdef = msglist.add_message(p[0], p[1:])
 306         b = msglist.encode(msgdef, {'length': 2,
 307                                     'counter': [{'address': 4,
 308                                                  'address_length': 12,
 309                                                  'packets': 1235,
 310                                                  'bytes': 5678},
 311                                                 {'address': 111,
 312                                                  'address_length': 222,
 313                                                  'packets': 333,
 314                                                  'bytes': 444}]})
 315         self.assertEqual(46, len(b))
 316         rv = msglist.decode(msgdef, b)
 317         self.assertEqual(2, rv.length)
 318         self.assertEqual(1235, rv.counter[0].packets)
 319         self.assertEqual(333, rv.counter[1].packets)
 320
 321     def test_simple_array(self):
 322         """ Simple array """
 323         msglist = VPP(testmode=True)
 324
 325         simple_byte_array = '''["simple_byte_array",
 326             ["u32", "length"],
 327             ["u8", "namecommand", 64]
 328
 329         ]'''
 330         p = json.loads(simple_byte_array)
 331         msgdef = msglist.add_message(p[0], p[1:])
 332         b = msglist.encode(msgdef, {'length': 2, 'namecommand': 'foobar'})
 333         self.assertEqual(68, len(b))
 334         rv = msglist.decode(msgdef, b)
 335         self.assertEqual(2, rv.length)
 336
 337         simple_array = '''["simple_array",
 338             ["u32", "length"],
 339             ["u32", "list", 2]
 340
 341         ]'''
 342         p = json.loads(simple_array)
 343         msgdef = msglist.add_message(p[0], p[1:])
 344         b = msglist.encode(msgdef, {'length': 2, 'list': [1, 2]})
 345         self.assertEqual(12, len(b))
 346         rv = msglist.decode(msgdef, b)
 347         self.assertEqual(2, rv.length)
 348         self.assertEqual([1, 2], rv.list)
 349
 350         simple_variable_array = '''["simple_variable_array",
 351             ["u32", "length"],
 352             ["u32", "list", 0, "length"]
 353
 354         ]'''
 355         p = json.loads(simple_variable_array)
 356         msgdef = msglist.add_message(p[0], p[1:])
 357         b = msglist.encode(msgdef, {'length': 2, 'list': [1, 2]})
 358         self.assertEqual(12, len(b))
 359         rv = msglist.decode(msgdef, b)
 360         self.assertEqual(2, rv.length)
 361         self.assertEqual([1, 2], rv.list)
 362
 363         simple_variable_byte_array = '''["simple_variable_byte_array",
 364             ["u32", "length"],
 365             ["u8", "list", 0, "length"]
 366         ]'''
 367         p = json.loads(simple_variable_byte_array)
 368         msgdef = msglist.add_message(p[0], p[1:])
 369         b = msglist.encode(msgdef, {'length': 6, 'list': 'foobar'})
 370         self.assertEqual(10, len(b))
 371         rv = msglist.decode(msgdef, b)
 372         self.assertEqual(6, rv.length)
 373         self.assertEqual('foobar', rv.list)
 374
 375     def test_old_vla_array(self):
 376         """ Old style VLA array """
 377         msglist = VPP(testmode=True)
 378
 379         # VLA
 380         vla_byte_array = '''["vla_byte_array",
 381             ["u32", "foobar"],
 382             ["u32", "list", 2],
 383             ["u32", "propercount"],
 384             ["u8", "propermask", 0, "propercount"],
 385             ["u8", "oldmask", 0],
 386             {"crc" : "0xb2739495"}
 387         ]'''
 388         p = json.loads(vla_byte_array)
 389         msgdef = msglist.add_message(p[0], p[1:])
 390         b = msglist.encode(msgdef, {'list': [123, 456], 'oldmask': b'foobar',
 391                                     'propercount': 2,
 392                                     'propermask': [8, 9]})
 393         self.assertEqual(24, len(b))
 394         rv = msglist.decode(msgdef, b)
 395         self.assertEqual(b'foobar', rv.oldmask)
 396
 397     def test_old_vla_array_not_last_member(self):
 398         """ Old VLA array arbitrary placement """
 399         msglist = VPP(testmode=True)
 400
 401         # VLA
 402         vla_byte_array = '''["vla_byte_array",
 403             ["u8", "oldmask", 0],
 404             ["u32", "foobar"],
 405             {"crc" : "0xb2739495"}
 406         ]'''
 407         p = json.loads(vla_byte_array)
 408         self.assertRaises(ValueError, msglist.add_message, p[0], p[1:])
 409
 410     def test_old_vla_array_u32(self):
 411         """ Old VLA u32 """
 412         msglist = VPP(testmode=True)
 413
 414         # VLA
 415         vla_byte_array = '''["vla_byte_array",
 416             ["u32", "foobar"],
 417             ["u32", "oldmask", 0],
 418             {"crc" : "0xb2739495"}
 419         ]'''
 420         p = json.loads(vla_byte_array)
 421         msgdef = msglist.add_message(p[0], p[1:])
 422         b = msglist.encode(msgdef, {'foobar': 123,
 423                                     'oldmask': [123, 456, 789]})
 424         self.assertEqual(16, len(b))
 425         rv = msglist.decode(msgdef, b)
 426         self.assertEqual([123, 456, 789], rv.oldmask)
 427
 428     def test_old_vla_array_compound(self):
 429         """ Old VLA compound type """
 430         msglist = VPP(testmode=True)
 431
 432         # VLA
 433         counter_type = '''["ip4_fib_counter",
 434             ["u32", "address"],
 435             ["u8", "address_length"],
 436             ["u64", "packets"],
 437             ["u64", "bytes"],
 438             {"crc" : "0xb2739495"}
 439         ]'''
 440
 441         vla_byte_array = '''["vla_byte_array",
 442             ["vl_api_ip4_fib_counter_t", "counter", 0],
 443             {"crc" : "0xb2739495"}
 444         ]'''
 445
 446         p = json.loads(counter_type)
 447         msglist.add_type(p[0], p[1:])
 448
 449         p = json.loads(vla_byte_array)
 450         with self.assertRaises(NotImplementedError):
 451             msgdef = msglist.add_message(p[0], p[1:])
 452
 453     def test_array_count_not_previous(self):
 454         """ VLA with aribtrary length field placement """
 455         msglist = VPP(testmode=True)
 456
 457         # VLA
 458         vla_byte_array = '''["vla_byte_array",
 459             ["u32", "count"],
 460             ["u32", "filler"],
 461             ["u32", "lst", 0, "count"],
 462             {"crc" : "0xb2739495"}
 463         ]'''
 464
 465         p = json.loads(vla_byte_array)
 466         msgdef = msglist.add_message(p[0], p[1:])
 467         b = msglist.encode(msgdef, {'count': 3, 'lst': [1, 2, 3],
 468                                     'filler': 1})
 469         rv = msglist.decode(msgdef, b)
 470         self.assertEqual(rv.lst, [1, 2, 3])
 471
 472     def test_argument_name(self):
 473         """ Argument name """
 474         msglist = VPP(testmode=True)
 475
 476         simple_name = '''["simple_name",
 477             ["u32", "length"],
 478             ["u8", "name"]
 479         ]'''
 480         p = json.loads(simple_name)
 481         msgdef = msglist.add_message(p[0], p[1:])
 482         b = msglist.encode(msgdef, {'length': 6, 'name': 1})
 483         self.assertEqual(5, len(b))
 484         rv = msglist.decode(msgdef, b)
 485         self.assertEqual(6, rv.length)
 486         self.assertEqual(1, rv.name)