vpp-api/python/tests/test_vpp_papi2.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import print_function
   4 import unittest, sys, threading, struct, logging, os
   5 from vpp_papi import VPP
   6 from ipaddress import *
   7 import glob, json
   8
   9 papi_event = threading.Event()
  10 import glob
  11
  12 import fnmatch
  13 import os
  14
  15 jsonfiles = []
  16 for root, dirnames, filenames in os.walk('../../../build-root/'):
  17     if root.find('install-') == -1: continue
  18     for filename in fnmatch.filter(filenames, '*.api.json'):
  19         jsonfiles.append(os.path.join(root, filename))
  20
  21 class TestPAPI(unittest.TestCase):
  22     show_version_msg = '''["show_version",
  23               ["u16", "_vl_msg_id"],
  24               ["u32", "client_index"],
  25               ["u32", "context"],
  26               {"crc" : "0xf18f9480"}
  27         ]'''
  28
  29     ip_address_details_msg = '''["ip_address_details",
  30             ["u16", "_vl_msg_id"],
  31             ["u32", "client_index"],
  32             ["u32", "context"],
  33             ["u8", "ip", 16],
  34             ["u8", "prefix_length"],
  35             {"crc" : "0x87d522a1"}
  36         ]'''
  37
  38     cli_inband_msg = '''["cli_inband",
  39             ["u16", "_vl_msg_id"],
  40             ["u32", "client_index"],
  41             ["u32", "context"],
  42             ["u32", "length"],
  43             ["u8", "cmd", 0, "length"],
  44             {"crc" : "0x22345937"}
  45         ]'''
  46
  47     def test_adding_new_message_object(self):
  48         p = json.loads(TestPAPI.show_version_msg)
  49         msglist = VPP([])
  50         msgdef = msglist.add_message(p[0], p[1:])
  51
  52         # Verify that message can be retrieved
  53         self.assertTrue(msglist['show_version'])
  54         self.assertFalse(msglist['foobar'])
  55
  56         # Test duplicate
  57         self.assertRaises(ValueError, msglist.add_message, p[0], p[1:])
  58
  59         # Look at return tuple
  60         self.assertTrue(msglist.ret_tup('show_version'))
  61
  62     def test_adding_new_message_object_with_array(self):
  63         p = json.loads(TestPAPI.ip_address_details_msg)
  64         msglist = VPP([])
  65         msglist.add_message(p[0], p[1:])
  66
  67         self.assertTrue(msglist['ip_address_details'])
  68
  69     def test_message_to_bytes(self):
  70         msglist = VPP([])
  71         p = json.loads(TestPAPI.show_version_msg)
  72         msgdef = msglist.add_message(p[0], p[1:])
  73
  74         # Give me a byte string for given message and given arguments
  75
  76         b = msglist.encode(msgdef, {'_vl_msg_id' : 50, 'context' : 123 })
  77         self.assertEqual(10, len(b))
  78         rv = msglist.decode(msgdef, b)
  79         self.assertEqual(rv._0, 50)
  80         self.assertEqual(rv.context, 123)
  81
  82
  83         p = json.loads(TestPAPI.ip_address_details_msg)
  84         msgdef = msglist.add_message(p[0], p[1:])
  85
  86         # Give me a byte string for given message and given arguments
  87         b = msglist.encode(msgdef, {'_vl_msg_id' : 50, 'context' : 123,
  88                                     'ip' : b'\xf0\xf1\xf2',
  89                                     'prefix_length' :  12})
  90         self.assertEqual(27, len(b))
  91         rv = msglist.decode(msgdef, b)
  92
  93         self.assertEqual(rv.context, 123)
  94         self.assertEqual(rv.ip, b'\xf0\xf1\xf2\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00')
  95         self.assertEqual(rv.prefix_length, 12)
  96
  97         p = json.loads(TestPAPI.cli_inband_msg)
  98         msgdef = msglist.add_message(p[0], p[1:])
  99
 100         # Give me a byte string for given message and given arguments
 101         b = msglist.encode(msgdef, { '_vl_msg_id' : 50, 'context' : 123,
 102                                      'length' : 20, 'cmd' : 'show version verbose'})
 103         self.assertEqual(34, len(b))
 104         rv = msglist.decode(msgdef, b)
 105         self.assertEqual(rv._0, 50)
 106         self.assertEqual(rv.context, 123)
 107         self.assertEqual(rv.cmd.decode('ascii'), 'show version verbose')
 108
 109         variable_array_16_msg = '''["variable_array_16",
 110             ["u32", "length"],
 111             ["u16", "list", 0, "length"]
 112         ]'''
 113
 114         p = json.loads(variable_array_16_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, { 'list' : [1, 2], 'length' :2})
 119         self.assertEqual(8, len(b))
 120         rv = msglist.decode(msgdef, b)
 121         self.assertEqual(2, rv.length)
 122         self.assertEqual([1,2], rv.list)
 123
 124     def test_add_new_types(self):
 125         counter_type = '''["ip4_fib_counter",
 126             ["u32", "address"],
 127             ["u8", "address_length"],
 128             ["u64", "packets"],
 129             ["u64", "bytes"],
 130             {"crc" : "0xb2739495"}
 131         ]'''
 132
 133         with_type_msg = '''["with_type_msg",
 134             ["u32", "length"],
 135             ["u16", "list", 0, "length"],
 136             ["ip4_fib_counter", "counter"]
 137
 138         ]'''
 139
 140         # Add new type
 141         msglist = VPP([])
 142         p = json.loads(counter_type)
 143         msglist.add_type(p[0], p[1:])
 144         p = json.loads(with_type_msg)
 145         msgdef = msglist.add_message(p[0], p[1:])
 146         b = msglist.encode(msgdef, {'length' : 2, 'list' : [1,2],
 147                            'counter' : { 'address' : 4, 'address_length' : 12,
 148                                        'packets': 1235, 'bytes' : 5678}})
 149         self.assertEqual(29, len(b)) # feil
 150         rv = msglist.decode(msgdef, b)
 151         self.assertEqual(2, rv.length)
 152         self.assertEqual(5678, rv.counter.bytes)
 153
 154     def test_add_new_compound_type_with_array(self):
 155         counter_type = '''["ip4_fib_counter",
 156             ["u32", "address"],
 157             ["u8", "address_length"],
 158             ["u64", "packets"],
 159             ["u64", "bytes"],
 160             {"crc" : "0xb2739495"}
 161         ]'''
 162
 163         with_type_msg = '''["with_type_msg",
 164             ["u32", "length"],
 165             ["u16", "list", 0, "length"],
 166             ["ip4_fib_counter", "counter", 2]
 167
 168         ]'''
 169
 170         # Add new type
 171         msglist = VPP([])
 172         p = json.loads(counter_type)
 173         msglist.add_type(p[0], p[1:])
 174         p = json.loads(with_type_msg)
 175         msgdef = msglist.add_message(p[0], p[1:])
 176         b = msglist.encode(msgdef, {'length' : 2, 'list' : [1,2],
 177                            'counter' : [{ 'address' : 4, 'address_length' : 12,
 178                                         'packets': 1235, 'bytes' : 5678},
 179                                       { 'address' : 111, 'address_length' : 222,
 180                                         'packets': 333, 'bytes' : 444}]})
 181         self.assertEqual(50, len(b))
 182         rv = msglist.decode(msgdef, b)
 183         self.assertEqual([1,2], rv.list)
 184         self.assertEqual(1235, rv.counter[0].packets)
 185
 186         with_type_variable_msg = '''["with_type_variable_msg",
 187             ["u32", "length"],
 188             ["ip4_fib_counter", "counter", 0, "length"]
 189
 190         ]'''
 191
 192         p = json.loads(with_type_variable_msg)
 193         msgdef = msglist.add_message(p[0], p[1:])
 194         b = msglist.encode(msgdef, {'length' : 2,
 195                            'counter' : [{ 'address' : 4, 'address_length' : 12,
 196                                         'packets': 1235, 'bytes' : 5678},
 197                                       { 'address' : 111, 'address_length' : 222,
 198                                         'packets': 333, 'bytes' : 444}]})
 199         self.assertEqual(46, len(b))
 200         rv = msglist.decode(msgdef, b)
 201         self.assertEqual(2, rv.length)
 202         self.assertEqual(1235, rv.counter[0].packets)
 203         self.assertEqual(333, rv.counter[1].packets)
 204
 205
 206     def test_compound_data_type(self):
 207         # Normal field
 208         # Fixed array
 209         # Variable array
 210         pass
 211
 212     def test_return_tuple(self):
 213         # Normal field
 214         # Fixed array
 215         # Variable array
 216         pass
 217
 218     def test_simple_array(self):
 219         msglist = VPP([])
 220
 221         simple_byte_array = '''["simple_byte_array",
 222             ["u32", "length"],
 223             ["u8", "namecommand", 64]
 224
 225         ]'''
 226         p = json.loads(simple_byte_array)
 227         msgdef = msglist.add_message(p[0], p[1:])
 228         b = msglist.encode(msgdef, {'length': 2, 'namecommand': 'foobar'})
 229         self.assertEqual(68, len(b))
 230         rv = msglist.decode(msgdef, b)
 231         self.assertEqual(2, rv.length)
 232
 233         simple_array = '''["simple_array",
 234             ["u32", "length"],
 235             ["u32", "list", 2]
 236
 237         ]'''
 238         p = json.loads(simple_array)
 239         msgdef = msglist.add_message(p[0], p[1:])
 240         b = msglist.encode(msgdef, {'length': 2, 'list': [1,2]})
 241         self.assertEqual(12, len(b))
 242         rv = msglist.decode(msgdef, b)
 243         self.assertEqual(2, rv.length)
 244         self.assertEqual([1,2], rv.list)
 245
 246         simple_variable_array = '''["simple_variable_array",
 247             ["u32", "length"],
 248             ["u32", "list", 0, "length"]
 249
 250         ]'''
 251         p = json.loads(simple_variable_array)
 252         msgdef = msglist.add_message(p[0], p[1:])
 253         b = msglist.encode(msgdef, {'length':2, 'list': [1,2]})
 254         self.assertEqual(12, len(b))
 255         rv = msglist.decode(msgdef, b)
 256         self.assertEqual(2, rv.length)
 257         self.assertEqual([1,2], rv.list)
 258
 259         simple_variable_byte_array = '''["simple_variable_byte_array",
 260             ["u32", "length"],
 261             ["u8", "list", 0, "length"]
 262         ]'''
 263         p = json.loads(simple_variable_byte_array)
 264         msgdef =msglist.add_message(p[0], p[1:])
 265         b = msglist.encode(msgdef, {'length': 6, 'list' : 'foobar'})
 266         self.assertEqual(10, len(b))
 267         rv = msglist.decode(msgdef, b)
 268         self.assertEqual(6, rv.length)
 269         self.assertEqual('foobar', rv.list)
 270
 271     def test_argument_name(self):
 272         msglist = VPP([])
 273
 274
 275         simple_name = '''["simple_name",
 276             ["u32", "length"],
 277             ["u8", "name"]
 278         ]'''
 279         p = json.loads(simple_name)
 280         msgdef = msglist.add_message(p[0], p[1:])
 281         b = msglist.encode(msgdef, {'length': 6, 'name': 1})
 282         self.assertEqual(5, len(b))
 283         rv = msglist.decode(msgdef, b)
 284         self.assertEqual(6, rv.length)
 285         self.assertEqual(1, rv.name)
 286
 287 class TestConnectedPAPI(unittest.TestCase):
 288     def test_request_reply_function(self):
 289         vpp = VPP(jsonfiles)
 290
 291         vpp.connect('test_vpp_papi2')
 292
 293         rv = vpp.show_version()
 294         self.assertEqual(0, rv.retval)
 295         print('RV', rv.program.decode().rstrip('\0x00'))
 296         vpp.disconnect()
 297
 298
 299     def test_dump_details_function(self):
 300         vpp = VPP(jsonfiles)
 301         vpp.connect('test_vpp_papi3')
 302
 303         rv = vpp.sw_interface_dump()
 304         #self.assertEqual(0, rv.retval)
 305         print('RV', rv)
 306         vpp.disconnect()
 307
 308     def test_vla(self):
 309         vpp = VPP(jsonfiles)
 310
 311         vpp.connect('test_vpp_papi3')
 312
 313         cmd = 'show version verbose'
 314         rv = vpp.cli_inband(length=len(cmd), cmd=cmd)
 315         self.assertEqual(0, rv.retval)
 316         print('RV', rv.reply)
 317
 318         cmd = 'show vlib graph'
 319         rv = vpp.cli_inband(length=len(cmd), cmd=cmd)
 320         self.assertEqual(0, rv.retval)
 321         print('RV', rv.reply)
 322         vpp.disconnect()
 323
 324     def test_events(self):
 325         vpp = VPP(jsonfiles)
 326
 327         vpp.connect('test_vpp_papi3')
 328
 329         vpp.register_event_callback(event_handler)
 330
 331         rv = vpp.want_interface_events(enable_disable = True)
 332         self.assertEqual(0, rv.retval)
 333         print('RV', rv)
 334
 335         rv = vpp.create_loopback()
 336         print('RV', rv)
 337         self.assertEqual(0, rv.retval)
 338
 339         rv = vpp.sw_interface_set_flags(sw_if_index = 1, admin_up_down = 1)
 340         print('RV', rv)
 341         self.assertEqual(0, rv.retval)
 342         rv = vpp.sw_interface_set_flags(sw_if_index = 1, admin_up_down = 0)
 343         print('RV', rv)
 344         self.assertEqual(0, rv.retval)
 345         self.assertEqual(papi_event.wait(10), True)
 346
 347         vpp.disconnect()
 348
 349 def event_handler(msgname, result):
 350     print('IN EVENT HANDLER:', msgname, result)
 351     papi_event.set()
 352
 353 class TestACL(unittest.TestCase):
 354     def test_acl_create(self):
 355         vpp = VPP(jsonfiles)
 356
 357         vpp.connect('acl-test')
 358
 359         rv = vpp.acl_plugin_get_version()
 360         print('RV', rv)
 361         self.assertEqual(rv.major, 0)
 362         self.assertEqual(rv.minor, 1)
 363
 364         rv = vpp.acl_add_replace(acl_index = 0xFFFFFFFF,
 365             r = [{
 366                 "is_permit" : 1,
 367                 "is_ipv6" : 0,
 368                 "proto" : 6,
 369                 "srcport_or_icmptype_first" : 80,
 370                 }],
 371             count = 1)
 372         print ('RV', rv)
 373         rv = vpp.acl_add_replace(acl_index = 0xFFFFFFFF,
 374             r = [{
 375                 "is_permit" : 1,
 376                 "is_ipv6" : 0,
 377                 "proto" : 6,
 378                 "srcport_or_icmptype_first" : 81,
 379                 }],
 380             count = 1)
 381         self.assertEqual(rv.retval, 0)
 382         print ('RV', rv)
 383         ai = rv.acl_index
 384         rv = vpp.acl_dump()
 385         print ('RV', rv)
 386
 387         #rv = vpp.acl_del(acl_index = ai)
 388         #self.assertEqual(rv.retval, 0)
 389
 390         #rv = vpp.acl_dump()
 391         #self.assertEqual([], vpp.acl_dump())
 392
 393         vpp.disconnect()
 394
 395     def test_status(self):
 396         vpp = VPP(jsonfiles)
 397         vpp.status()
 398
 399
 400 if __name__ == '__main__':
 401     unittest.main()