1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
from __future__ import print_function
import unittest, sys, time, threading, struct, logging, os
import vpp_papi
from ipaddress import *
scriptdir = os.path.dirname(os.path.realpath(__file__))
papi_event = threading.Event()
print(vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS)
def papi_event_handler(result):
if result.vl_msg_id == vpp_papi.vpe.VL_API_SW_INTERFACE_SET_FLAGS:
return
if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_INTERFACE_COUNTERS:
print('Interface counters', result)
return
if result.vl_msg_id == vpp_papi.vpe.VL_API_VNET_IP6_FIB_COUNTERS:
print('IPv6 FIB counters', result)
papi_event.set()
return
print('Unknown message id:', result.vl_msg_id)
import glob, subprocess
class TestPAPI(unittest.TestCase):
@classmethod
def setUpClass(cls):
#
# Start main VPP process
cls.vpp_bin = glob.glob(scriptdir+'/../../../build-root/install-vpp*-native/vpp/bin/vpp')[0]
print("VPP BIN:", cls.vpp_bin)
cls.vpp = subprocess.Popen([cls.vpp_bin, "unix", "nodaemon"], stderr=subprocess.PIPE)
print('Started VPP')
# For some reason unless we let VPP start up the API cannot connect.
time.sleep(0.3)
@classmethod
def tearDownClass(cls):
cls.vpp.terminate()
def setUp(self):
print("Connecting API")
r = vpp_papi.connect("test_papi")
self.assertEqual(r, 0)
def tearDown(self):
r = vpp_papi.disconnect()
self.assertEqual(r, 0)
#
# The tests themselves
#
#
# Basic request / reply
#
def test_show_version(self):
t = vpp_papi.show_version()
print('T', t);
program = t.program.decode().rstrip('\x00')
self.assertEqual('vpe', program)
#
# Details / Dump
#
def test_details_dump(self):
t = vpp_papi.sw_interface_dump(0, b'')
print('Dump/details T', t)
#
# Arrays
#
def test_arrays(self):
t = vpp_papi.vnet_get_summary_stats()
print('Summary stats', t)
print('Packets:', t.total_pkts[0])
print('Packets:', t.total_pkts[1])
#
# Variable sized arrays and counters
#
#@unittest.skip("stats")
def test_want_stats(self):
pid = 123
vpp_papi.register_event_callback(papi_event_handler)
papi_event.clear()
# Need to configure IPv6 to get som IPv6 FIB stats
t = vpp_papi.create_loopback('')
print(t)
self.assertEqual(t.retval, 0)
ifindex = t.sw_if_index
addr = str(IPv6Address(u'1::1').packed)
t = vpp_papi.sw_interface_add_del_address(ifindex, 1, 1, 0, 16, addr)
print(t)
self.assertEqual(t.retval, 0)
# Check if interface is up
# XXX: Add new API to query interface state based on ifindex, instead of dump all.
t = vpp_papi.sw_interface_set_flags(ifindex, 1, 1, 0)
self.assertEqual(t.retval, 0)
t = vpp_papi.want_stats(True, pid)
print (t)
#
# Wait for some stats
#
self.assertEqual(papi_event.wait(15), True)
t = vpp_papi.want_stats(False, pid)
print (t)
#
# Plugins?
#
if __name__ == '__main__':
#logging.basicConfig(level=logging.DEBUG)
unittest.main()
def test_papi():
print('test')
|
