ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

misc: vpp_papi- add tests, clean up pep8

Browse Source 
Type: test

Change-Id: Ic81bd4a9eba3c89a746e7a9b9e471b59cd87fa40
Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
This commit is contained in:
Paul Vinciguerra
2019-06-07 13:35:37 -04:00
committed by Ole Trøan
parent 283cd2e9af
commit ae8819f0a4
8 changed files with 205 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -94,6 +94,8 @@ GTAGS
/src/vpp-api/python/build
/src/vpp-api/python/dist
/src/vpp-api/python/vpp_papi.egg-info
/src/vpp_api/.tox
/src/vpp_api/.stestr
# Build files in the test directory
/test/*.ok
@ -103,3 +105,4 @@ GTAGS
# No core files
**/core

40
src/vpp-api/python/tox.ini Normal file
View File

@ -0,0 +1,40 @@
[tox]
envlist = py27,py3{6,7}, pep8
minversion = 2.3.1
skipsdist = True
skip_missing_interpreters = True
[testenv]
usedevelop = True
whitelist_externals =
find
deps =
stestr
parameterized
commands =
find . -type f -name "*.pyc" -delete
stestr --test-path {toxinidir}/vpp_papi/tests/ run --slowest {posargs}
[testenv:pep8]
basepython = python3
deps =
flake8
commands =
flake8 --exclude {toxinidir}/vpp_papi/tests {toxinidir}/vpp_papi {posargs}
[testenv:pypi]
basepython = python3
deps =
twine
wheel
passenv =
TWINE_USERNAME
TWINE_PASSWORD
TWINE_REPOSITORY
TWINE_REPOSITORY_URL
TWINE_CERT
commands =
python setup.py bdist_wheel
twine upload {toxinidir}/dist/*

11
src/vpp-api/python/vpp_papi/__init__.py
View File

@ -1 +1,10 @@
from .vpp_papi import *
from .vpp_papi import FuncWrapper, VPP, VppApiDynamicMethodHolder # noqa: F401
from .vpp_papi import VppEnum, VppEnumType # noqa: F401
from .vpp_papi import VPPIOError, VPPRuntimeError, VPPValueError # noqa: F401
from .vpp_papi import VPPApiClient # noqa: F401
from . macaddress import MACAddress, mac_pton, mac_ntop # noqa: F401
# sorted lexicographically
from .vpp_serializer import BaseTypes # noqa: F401
from .vpp_serializer import VPPEnumType, VPPType, VPPTypeAlias # noqa: F401
from .vpp_serializer import VPPMessage, VPPUnionType # noqa: F401

130
src/vpp-api/python/vpp_papi/tests/test_vpp_format.py Normal file
View File

@ -0,0 +1,130 @@
# Copyright (c) 2019. Vinci Consulting Corp. All Rights Reserved.
#
# 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.
import ipaddress
import socket
import unittest
try:
text_type = unicode
except NameError:
text_type = str
from vpp_papi import vpp_format
from parameterized import parameterized
ip4_addr = '1.2.3.4'
ip4_addrn = b'\x01\x02\x03\x04'
ip4_prefix_len = 32
ip4_prefix = '%s/%s' % (ip4_addr, ip4_prefix_len)
ipv4_network = ipaddress.IPv4Network(text_type(ip4_prefix))
ip4_addr_format_vl_api_address_t = {'un': {'ip4': b'\x01\x02\x03\x04'},
'af': 0}
ip4_addr_format_vl_api_prefix_t = {'prefix':
{'un': {'ip4': b'\x01\x02\x03\x04'},
'af': 0},
'len': ip4_prefix_len}
ip4_addr_format_vl_api_prefix_packed_t = {'prefix': b'\x01\x02\x03\x04',
'len': ip4_prefix_len}
ip6_addr = 'dead::'
ip6_addrn = b'\xde\xad\x00\x00\x00\x00\x00\x00' \
b'\x00\x00\x00\x00\x00\x00\x00\x00'
ip6_prefix_len = 127
ip6_prefix = '%s/%s' % (ip6_addr, ip6_prefix_len)
ipv6_network = ipaddress.IPv6Network(text_type(ip6_prefix))
ip6_addr_format_vl_api_address_t = {'un': {'ip6': b'\xde\xad\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00'},
'af': 1}
ip6_addr_format_vl_api_prefix_t = {'prefix':
{'af': 1,
'un': {
'ip6': b'\xde\xad\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00'}},
'len': ip6_prefix_len}
ip6_addr_format_vl_api_prefix_packed_t = {'prefix': b'\xde\xad\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00'
b'\x00\x00\x00\x00',
'len': ip6_prefix_len}
class TestVppFormat(unittest.TestCase):
def test_format_vl_api_address_t(self):
res = vpp_format.format_vl_api_address_t(ip4_addr)
self.assertEqual(res, ip4_addr_format_vl_api_address_t)
# PY2: raises socket.error
# PY3: raises OSError
with self.assertRaises((TypeError,
socket.error,
OSError)):
res = vpp_format.format_vl_api_address_t(ip4_addrn)
res = vpp_format.format_vl_api_address_t(ip6_addr)
self.assertEqual(res, ip6_addr_format_vl_api_address_t)
with self.assertRaises(TypeError):
es = vpp_format.format_vl_api_address_t(ip6_addrn)
@parameterized.expand([('ip4 prefix',
ip4_prefix,
ip4_addr_format_vl_api_prefix_t),
('ip6 prefix',
ip6_prefix,
ip6_addr_format_vl_api_prefix_t),
('IPv4Network',
ipv4_network,
ip4_addr_format_vl_api_prefix_t),
('IPv6Network',
ipv6_network,
ip6_addr_format_vl_api_prefix_t),
])
def test_format_vl_api_prefix_t(self, _, arg, expected):
res = vpp_format.format_vl_api_prefix_t(arg)
self.assertEqual(res, expected)
def test_format_vl_api_ip6_prefix_t(self):
res = vpp_format.format_vl_api_ip6_prefix_t(ip6_prefix)
self.assertEqual(res, ip6_addr_format_vl_api_prefix_packed_t)
res = vpp_format.format_vl_api_ip6_prefix_t(ipv6_network)
self.assertEqual(res, ip6_addr_format_vl_api_prefix_packed_t)
def test_format_vl_api_ip4_prefix_t(self):
res = vpp_format.format_vl_api_ip4_prefix_t(ip4_prefix)
self.assertEqual(res, ip4_addr_format_vl_api_prefix_packed_t)
res = vpp_format.format_vl_api_ip4_prefix_t(ipv4_network)
self.assertEqual(res, ip4_addr_format_vl_api_prefix_packed_t)
def test_format_vl_api_ip6_prefix_t_raises(self):
# PY2: raises socket.error
# PY3: raises OSError
with self.assertRaises((socket.error, OSError)):
res = vpp_format.format_vl_api_ip6_prefix_t(ip4_prefix)
def test_format_vl_api_ip4_prefix_t_raises(self):
# PY2: raises socket.error
# PY3: raises OSError
with self.assertRaises((socket.error, OSError)):
res = vpp_format.format_vl_api_ip4_prefix_t(ip6_prefix)

10
src/vpp-api/python/vpp_papi/vpp_papi.py
View File

@ -21,17 +21,14 @@ import sys
import multiprocessing as mp
import os
import logging
import collections
import struct
import functools
import json
import threading
import fnmatch
import weakref
import atexit
from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes
from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType
from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias
from . macaddress import MACAddress, mac_pton, mac_ntop
logger = logging.getLogger(__name__)
@ -40,6 +37,11 @@ if sys.version[0] == '2':
else:
import queue as queue
__all__ = ('FuncWrapper', 'VPP', 'VppApiDynamicMethodHolder',
'VppEnum', 'VppEnumType',
'VPPIOError', 'VPPRuntimeError', 'VPPValueError',
'VPPApiClient', )
def metaclass(metaclass):
@functools.wraps(metaclass)

17
src/vpp-api/python/vpp_papi/vpp_serializer.py
View File

@ -12,25 +12,24 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import struct
import collections
import sys
import logging
from . import vpp_format
import ipaddress
import socket
import struct
import sys
if sys.version_info <= (3, 4):
from aenum import IntEnum
from aenum import IntEnum # noqa: F401
else:
from enum import IntEnum
from enum import IntEnum # noqa: F401
if sys.version_info <= (3, 6):
from aenum import IntFlag
from aenum import IntFlag # noqa: F401
else:
from enum import IntFlag
from enum import IntFlag # noqa: F401
from . import vpp_format # noqa: E402
#
# Set log-level in application by doing e.g.:

10
src/vpp-api/python/vpp_papi/vpp_stats.py
View File

@ -86,7 +86,7 @@ char *stat_segment_index_to_name_r (uint32_t index, stat_client_main_t * sm);
uint64_t stat_segment_version(void);
uint64_t stat_segment_version_r(stat_client_main_t *sm);
void free(void *ptr);
""")
""") # noqa: E501
# Utility functions
@ -130,12 +130,14 @@ def combined_counter_vec_list(api, e):
vec.append(if_per_thread)
return vec
def error_vec_list(api, e):
vec = []
for thread in range(api.stat_segment_vec_len(e)):
vec.append(e[thread])
return vec
def name_vec_list(api, e):
return [ffi.string(e[i]).decode('utf-8') for i in
range(api.stat_segment_vec_len(e)) if e[i] != ffi.NULL]
@ -175,7 +177,7 @@ class VPPStatsIOError(IOError):
if not message:
try:
message = self.message % kwargs
except Exception as e:
except Exception:
message = self.message
else:
message = message % kwargs
@ -258,7 +260,7 @@ class VPPStats(object):
.format(name))
k, v = s.popitem()
return v
except VPPStatsIOError as e:
except VPPStatsIOError:
if retries > 10:
return None
retries += 1
@ -280,7 +282,7 @@ class VPPStats(object):
error_names = self.ls(['/err/'])
error_counters = self.dump(error_names)
break
except VPPStatsIOError as e:
except VPPStatsIOError:
if retries > 10:
return None
retries += 1

2
src/vpp-api/python/vpp_papi/vpp_transport_socket.py
View File

@ -188,6 +188,8 @@ class VppTransport(object):
header = self.header.pack(0, len(buf), 0)
n = self.socket.send(header)
n = self.socket.send(buf)
if n == 0:
raise VppTransportSocketIOError(1, 'Not connected')
def _read(self):
hdr = self.socket.recv(16)