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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
#!/usr/bin/env python3
# Copyright (c) 2019 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Python API provider.
"""
import argparse
import binascii
import json
import os
import sys
# Sphinx creates auto-generated documentation by importing the python source
# files and collecting the docstrings from them. The NO_VPP_PAPI flag allows
# the vpp_papi_provider.py file to be importable without having to build
# the whole vpp api if the user only wishes to generate the test documentation.
do_import = True
try:
no_vpp_papi = os.getenv("NO_VPP_PAPI")
if no_vpp_papi == "1":
do_import = False
except:
pass
if do_import:
# TODO: run os.walk once per whole suite and store the path in environmental
# variable
modules_path = None
for root, dirs, files in os.walk('/usr/lib'):
for name in files:
if name == 'vpp_papi.py':
modules_path = os.path.split(root)[0]
break
if modules_path:
sys.path.append(modules_path)
from vpp_papi import VPP
else:
raise RuntimeError('vpp_papi module not found')
# client name
CLIENT_NAME = 'csit_papi'
def papi_init():
"""Construct a VPP instance from VPP JSON API files.
:param vpp_json_dir: Directory containing all the JSON API files. If VPP is
installed in the system it will be in /usr/share/vpp/api/.
:type vpp_json_dir: str
:returns: VPP instance.
:rtype: VPP object
:raises PapiJsonFileError: If no api.json file found.
:raises PapiInitError: If PAPI initialization failed.
"""
try:
vpp = VPP()
return vpp
except Exception as err:
raise RuntimeError('PAPI init failed:\n{err}'.format(err=repr(err)))
def papi_connect(vpp_client, name='vpp_api'):
"""Attach to VPP client.
:param vpp_client: VPP instance to connect to.
:param name: VPP client name.
:type vpp_client: VPP object
:type name: str
:returns: Return code of VPP.connect() method.
:rtype: int
"""
return vpp_client.connect(name)
def papi_disconnect(vpp_client):
"""Detach from VPP client.
:param vpp_client: VPP instance to detach from.
:type vpp_client: VPP object
"""
vpp_client.disconnect()
def papi_run(vpp_client, api_name, api_args):
"""Run PAPI.
:param vpp_client: VPP instance.
:param api_name: VPP API name.
:param api_args: Input arguments of the API.
:type vpp_client: VPP object
:type api_name: str
:type api_args: dict
:returns: VPP API reply.
:rtype: Vpp_serializer reply object
"""
papi_fn = getattr(vpp_client.api, api_name)
return papi_fn(**api_args)
def convert_reply(api_r):
"""Process API reply / a part of API reply for smooth converting to
JSON string.
Apply binascii.hexlify() method for string values.
:param api_r: API reply.
:type api_r: Vpp_serializer reply object (named tuple)
:returns: Processed API reply / a part of API reply.
:rtype: dict
"""
unwanted_fields = ['count', 'index']
reply_dict = dict()
reply_key = repr(api_r).split('(')[0]
reply_value = dict()
for item in dir(api_r):
if not item.startswith('_') and item not in unwanted_fields:
# attr_value = getattr(api_r, item)
# value = binascii.hexlify(attr_value) \
# if isinstance(attr_value, str) else attr_value
value = getattr(api_r, item)
reply_value[item] = value
reply_dict[reply_key] = reply_value
return reply_dict
def process_reply(api_reply):
"""Process API reply for smooth converting to JSON string.
:param api_reply: API reply.
:type api_reply: Vpp_serializer reply object (named tuple) or list of
vpp_serializer reply objects
:returns: Processed API reply.
:rtype: list or dict
"""
if isinstance(api_reply, list):
converted_reply = list()
for r in api_reply:
converted_reply.append(convert_reply(r))
else:
converted_reply = convert_reply(api_reply)
return converted_reply
def main():
"""Main function for the Python API provider.
:raises RuntimeError: If invalid attribute name or invalid value is
used in API call or if PAPI command(s) execution failed.
"""
parser = argparse.ArgumentParser()
parser.add_argument("-j", "--json_data",
required=True,
type=str,
help="JSON string (list) containing API name(s) and "
"its/their input argument(s).")
parser.add_argument("-d", "--json_dir",
type=str,
default='/usr/share/vpp/api/',
help="Directory containing all vpp json api files.")
args = parser.parse_args()
json_string = args.json_data
vpp = papi_init()
reply = list()
json_data = json.loads(json_string)
papi_connect(vpp, CLIENT_NAME)
for data in json_data:
api_name = data['api_name']
api_args_unicode = data['api_args']
api_reply = dict(api_name=api_name)
api_args = dict()
for a_k, a_v in api_args_unicode.items():
value = binascii.unhexlify(a_v) if isinstance(a_v, unicode) else a_v
api_args[str(a_k)] = value
try:
rep = papi_run(vpp, api_name, api_args)
api_reply['api_reply'] = process_reply(rep)
reply.append(api_reply)
except (AttributeError, ValueError) as err:
papi_disconnect(vpp)
raise RuntimeError('PAPI command {api}({args}) input error:\n{err}'.
format(api=api_name,
args=api_args,
err=repr(err)))
except Exception as err:
papi_disconnect(vpp)
raise RuntimeError('PAPI command {api}({args}) error:\n{exc}'.
format(api=api_name,
args=api_args,
exc=repr(err)))
papi_disconnect(vpp)
return json.dumps(reply)
if __name__ == '__main__':
sys.stdout.write(main())
sys.stdout.flush()
sys.exit(0)
|
