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