packet/bgp: Reduce Pylint warnings

This patch removes the following Pylint warnings:
  - anomalous-backslash-in-string
  - arguments-differ
  - bad-builtin
  - bad-staticmethod-argument
  - superfluous-parens
  - super-init-not-called
  - unidiomatic-typecheck
  - unused-variable
  - wrong-import-order

Signed-off-by: IWASE Yusuke <iwase.yusuke0@gmail.com>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>
This commit is contained in:
IWASE Yusuke 2016-05-25 15:27:37 +09:00 committed by FUJITA Tomonori
parent 171b35f0db
commit b3b8238e6c
2 changed files with 87 additions and 66 deletions

View File

@ -26,10 +26,11 @@ import abc
import copy
import functools
import numbers
import six
import socket
import struct
import six
from ryu.lib.stringify import StringifyMixin
from ryu.lib.packet import afi as addr_family
from ryu.lib.packet import safi as subaddr_family
@ -158,14 +159,14 @@ class _Value(object):
_VALUE_FIELDS = ['value']
@staticmethod
def do_init(cls, self, kwargs, **extra_kwargs):
def do_init(cls_type, self, kwargs, **extra_kwargs):
ourfields = {}
for f in cls._VALUE_FIELDS:
for f in cls_type._VALUE_FIELDS:
v = kwargs[f]
del kwargs[f]
ourfields[f] = v
kwargs.update(extra_kwargs)
super(cls, self).__init__(**kwargs)
super(cls_type, self).__init__(**kwargs)
self.__dict__.update(ourfields)
@classmethod
@ -232,6 +233,7 @@ class BgpExc(Exception):
"""Flag if set indicates Notification message should be sent to peer."""
def __init__(self, data=''):
super(BgpExc, self).__init__()
self.data = data
def __str__(self):
@ -256,6 +258,7 @@ class BadLen(BgpExc):
SUB_CODE = BGP_ERROR_SUB_BAD_MESSAGE_LENGTH
def __init__(self, msg_type_code, message_length):
super(BadLen, self).__init__()
self.msg_type_code = msg_type_code
self.length = message_length
self.data = struct.pack('!H', self.length)
@ -275,6 +278,7 @@ class BadMsg(BgpExc):
SUB_CODE = BGP_ERROR_SUB_BAD_MESSAGE_TYPE
def __init__(self, msg_type):
super(BadMsg, self).__init__()
self.msg_type = msg_type
self.data = struct.pack('B', msg_type)
@ -313,6 +317,7 @@ class UnsupportedVersion(BgpExc):
SUB_CODE = BGP_ERROR_SUB_UNSUPPORTED_VERSION_NUMBER
def __init__(self, locally_support_version):
super(UnsupportedVersion, self).__init__()
self.data = struct.pack('H', locally_support_version)
@ -399,6 +404,7 @@ class MissingWellKnown(BgpExc):
SUB_CODE = BGP_ERROR_SUB_MISSING_WELL_KNOWN_ATTRIBUTE
def __init__(self, pattr_type_code):
super(MissingWellKnown, self).__init__()
self.pattr_type_code = pattr_type_code
self.data = struct.pack('B', pattr_type_code)
@ -607,9 +613,9 @@ def get_rf(afi, safi):
return _rf_map[(afi, safi)]
def pad(bin, len_):
assert len(bin) <= len_
return bin + b'\0' * (len_ - len(bin))
def pad(binary, len_):
assert len(binary) <= len_
return binary + b'\0' * (len_ - len(binary))
class _RouteDistinguisher(StringifyMixin, _TypeDisp, _Value):
@ -618,7 +624,9 @@ class _RouteDistinguisher(StringifyMixin, _TypeDisp, _Value):
IPV4_ADDRESS = 1
FOUR_OCTET_AS = 2
def __init__(self, type_, admin=0, assigned=0):
def __init__(self, admin=0, assigned=0, type_=None):
if type_ is None:
type_ = self._rev_lookup_type(self.__class__)
self.type = type_
self.admin = admin
self.assigned = assigned
@ -629,7 +637,7 @@ class _RouteDistinguisher(StringifyMixin, _TypeDisp, _Value):
(type_,) = struct.unpack_from(cls._PACK_STR, six.binary_type(buf))
rest = buf[struct.calcsize(cls._PACK_STR):]
subcls = cls._lookup_type(type_)
return subcls(type_=type_, **subcls.parse_value(rest))
return subcls(**subcls.parse_value(rest))
@classmethod
def from_str(cls, str_):
@ -645,7 +653,7 @@ class _RouteDistinguisher(StringifyMixin, _TypeDisp, _Value):
type_ = cls.TWO_OCTET_AS
first = int(first)
subcls = cls._lookup_type(type_)
return subcls(type_=type_, admin=first, assigned=int(second))
return subcls(admin=first, assigned=int(second))
def serialize(self):
value = self.serialize_value()
@ -663,8 +671,9 @@ class BGPTwoOctetAsRD(_RouteDistinguisher):
_VALUE_PACK_STR = '!HI'
_VALUE_FIELDS = ['admin', 'assigned']
def __init__(self, type_=_RouteDistinguisher.TWO_OCTET_AS, **kwargs):
self.do_init(BGPTwoOctetAsRD, self, kwargs, type_=type_)
def __init__(self, **kwargs):
super(BGPTwoOctetAsRD, self).__init__()
self.do_init(BGPTwoOctetAsRD, self, kwargs)
@_RouteDistinguisher.register_type(_RouteDistinguisher.IPV4_ADDRESS)
@ -677,8 +686,9 @@ class BGPIPv4AddressRD(_RouteDistinguisher):
]
}
def __init__(self, type_=_RouteDistinguisher.IPV4_ADDRESS, **kwargs):
self.do_init(BGPIPv4AddressRD, self, kwargs, type_=type_)
def __init__(self, **kwargs):
super(BGPIPv4AddressRD, self).__init__()
self.do_init(BGPIPv4AddressRD, self, kwargs)
@classmethod
def parse_value(cls, buf):
@ -703,16 +713,16 @@ class BGPFourOctetAsRD(_RouteDistinguisher):
_VALUE_PACK_STR = '!IH'
_VALUE_FIELDS = ['admin', 'assigned']
def __init__(self, type_=_RouteDistinguisher.FOUR_OCTET_AS,
**kwargs):
self.do_init(BGPFourOctetAsRD, self, kwargs, type_=type_)
def __init__(self, **kwargs):
super(BGPFourOctetAsRD, self).__init__()
self.do_init(BGPFourOctetAsRD, self, kwargs)
@six.add_metaclass(abc.ABCMeta)
class _AddrPrefix(StringifyMixin):
_PACK_STR = '!B' # length
def __init__(self, length, addr, prefixes=None):
def __init__(self, length, addr, prefixes=None, **kwargs):
# length is on-wire bit length of prefixes+addr.
assert prefixes != ()
if isinstance(addr, tuple):
@ -724,14 +734,14 @@ class _AddrPrefix(StringifyMixin):
addr = prefixes + (addr,)
self.addr = addr
@staticmethod
@classmethod
@abc.abstractmethod
def _to_bin(addr):
def _to_bin(cls, addr):
pass
@staticmethod
@classmethod
@abc.abstractmethod
def _from_bin(addr):
def _from_bin(cls, addr):
pass
@classmethod
@ -764,12 +774,12 @@ class _AddrPrefix(StringifyMixin):
class _BinAddrPrefix(_AddrPrefix):
@staticmethod
def _to_bin(addr):
@classmethod
def _to_bin(cls, addr):
return addr
@staticmethod
def _from_bin(addr):
@classmethod
def _from_bin(cls, addr):
return addr
@ -811,10 +821,10 @@ class _LabelledAddrPrefix(_AddrPrefix):
return buf
@classmethod
def _label_from_bin(cls, bin):
def _label_from_bin(cls, label):
(b1, b2, b3) = struct.unpack_from(cls._LABEL_PACK_STR,
six.binary_type(bin))
rest = bin[struct.calcsize(cls._LABEL_PACK_STR):]
six.binary_type(label))
rest = label[struct.calcsize(cls._LABEL_PACK_STR):]
return (b1 << 16) | (b2 << 8) | b3, rest
@classmethod
@ -824,7 +834,7 @@ class _LabelledAddrPrefix(_AddrPrefix):
labels = [x << 4 for x in labels]
if labels and labels[-1] != cls._WITHDRAW_LABEL:
labels[-1] |= 1 # bottom of stack
bin_labels = list(map(cls._label_to_bin, labels))
bin_labels = list(cls._label_to_bin(l) for l in labels)
return bytes(reduce(lambda x, y: x + y, bin_labels,
bytearray()) + cls._prefix_to_bin(rest))
@ -880,7 +890,7 @@ class _IPAddrPrefix(_AddrPrefix):
@staticmethod
def _prefix_from_bin(addr):
return (addrconv.ipv4.bin_to_text(pad(addr, 4)),)
return addrconv.ipv4.bin_to_text(pad(addr, 4)),
class _IP6AddrPrefix(_AddrPrefix):
@ -891,7 +901,7 @@ class _IP6AddrPrefix(_AddrPrefix):
@staticmethod
def _prefix_from_bin(addr):
return (addrconv.ipv6.bin_to_text(pad(addr, 16)),)
return addrconv.ipv6.bin_to_text(pad(addr, 16)),
class _VPNAddrPrefix(_AddrPrefix):
@ -1112,7 +1122,7 @@ class RouteTargetMembershipNLRI(StringifyMixin):
return cls(origin_as, route_target)
def serialize(self):
rt_nlri = ''
rt_nlri = b''
if not self.is_default_rtnlri():
rt_nlri += struct.pack('!I', self.origin_as)
# Encode route target
@ -1123,7 +1133,7 @@ class RouteTargetMembershipNLRI(StringifyMixin):
def _addr_class_key(route_family):
return (route_family.afi, route_family.safi)
return route_family.afi, route_family.safi
_ADDR_CLASSES = {
@ -1164,7 +1174,7 @@ class _OptParam(StringifyMixin, _TypeDisp, _Value):
rest = rest[length:]
subcls = cls._lookup_type(type_)
caps = subcls.parse_value(value)
if type(caps) != list:
if not isinstance(caps, list):
caps = [subcls(type_=type_, length=length, **caps[0])]
return caps, rest
@ -1293,7 +1303,6 @@ class BGPOptParamCapabilityGracefulRestart(_OptParamCapability):
def serialize_cap_value(self):
buf = bytearray()
msg_pack_into(self._CAP_PACK_STR, buf, 0, self.flags << 12 | self.time)
i = 0
offset = 2
for i in self.tuples:
afi, safi, flags = i
@ -1522,7 +1531,7 @@ class _BGPPathAttributeAsPathCommon(_PathAttribute):
six.binary_type(buf))
buf = buf[struct.calcsize(cls._SEG_HDR_PACK_STR):]
l = []
for i in range(0, num_as):
for _ in range(0, num_as):
(as_number,) = struct.unpack_from(as_pack_str,
six.binary_type(buf))
buf = buf[struct.calcsize(as_pack_str):]
@ -1532,7 +1541,8 @@ class _BGPPathAttributeAsPathCommon(_PathAttribute):
elif type_ == cls._AS_SEQUENCE:
result.append(l)
else:
assert(0) # protocol error
# protocol error
raise struct.error('Unsupported segment type: %s' % type_)
return {
'value': result,
'as_pack_str': as_pack_str,
@ -1546,6 +1556,10 @@ class _BGPPathAttributeAsPathCommon(_PathAttribute):
type_ = self._AS_SET
elif isinstance(e, list):
type_ = self._AS_SEQUENCE
else:
raise struct.error(
'Element of %s.value must be of type set or list' %
self.__class__.__name__)
l = list(e)
num_as = len(l)
if num_as == 0:
@ -1904,7 +1918,9 @@ class _ExtendedCommunity(StringifyMixin, _TypeDisp, _Value):
FOUR_OCTET_AS_SPECIFIC = 0x02
OPAQUE = 0x03
def __init__(self, type_):
def __init__(self, type_=None):
if type_ is None:
type_ = self._rev_lookup_type(self.__class__)
self.type = type_
@classmethod
@ -1929,10 +1945,9 @@ class BGPTwoOctetAsSpecificExtendedCommunity(_ExtendedCommunity):
_VALUE_PACK_STR = '!BHI' # sub type, as number, local adm
_VALUE_FIELDS = ['subtype', 'as_number', 'local_administrator']
def __init__(self, type_=_ExtendedCommunity.TWO_OCTET_AS_SPECIFIC,
**kwargs):
self.do_init(BGPTwoOctetAsSpecificExtendedCommunity, self, kwargs,
type_=type_)
def __init__(self, **kwargs):
super(BGPTwoOctetAsSpecificExtendedCommunity, self).__init__()
self.do_init(BGPTwoOctetAsSpecificExtendedCommunity, self, kwargs)
@_ExtendedCommunity.register_type(_ExtendedCommunity.IPV4_ADDRESS_SPECIFIC)
@ -1945,10 +1960,9 @@ class BGPIPv4AddressSpecificExtendedCommunity(_ExtendedCommunity):
]
}
def __init__(self, type_=_ExtendedCommunity.IPV4_ADDRESS_SPECIFIC,
**kwargs):
self.do_init(BGPIPv4AddressSpecificExtendedCommunity, self, kwargs,
type_=type_)
def __init__(self, **kwargs):
super(BGPIPv4AddressSpecificExtendedCommunity, self).__init__()
self.do_init(BGPIPv4AddressSpecificExtendedCommunity, self, kwargs)
@classmethod
def parse_value(cls, buf):
@ -1974,10 +1988,9 @@ class BGPFourOctetAsSpecificExtendedCommunity(_ExtendedCommunity):
_VALUE_PACK_STR = '!BIH' # sub type, as number, local adm
_VALUE_FIELDS = ['subtype', 'as_number', 'local_administrator']
def __init__(self, type_=_ExtendedCommunity.FOUR_OCTET_AS_SPECIFIC,
**kwargs):
self.do_init(BGPFourOctetAsSpecificExtendedCommunity, self, kwargs,
type_=type_)
def __init__(self, **kwargs):
super(BGPFourOctetAsSpecificExtendedCommunity, self).__init__()
self.do_init(BGPFourOctetAsSpecificExtendedCommunity, self, kwargs)
@_ExtendedCommunity.register_type(_ExtendedCommunity.OPAQUE)
@ -1985,18 +1998,18 @@ class BGPOpaqueExtendedCommunity(_ExtendedCommunity):
_VALUE_PACK_STR = '!7s' # opaque value
_VALUE_FIELDS = ['opaque']
def __init__(self, type_=_ExtendedCommunity.OPAQUE,
**kwargs):
self.do_init(BGPOpaqueExtendedCommunity, self, kwargs,
type_=type_)
def __init__(self, **kwargs):
super(BGPOpaqueExtendedCommunity, self).__init__()
self.do_init(BGPOpaqueExtendedCommunity, self, kwargs)
@_ExtendedCommunity.register_unknown_type()
class BGPUnknownExtendedCommunity(_ExtendedCommunity):
_VALUE_PACK_STR = '!7s' # opaque value
def __init__(self, **kwargs):
self.do_init(BGPUnknownExtendedCommunity, self, kwargs)
def __init__(self, type_, **kwargs):
super(BGPUnknownExtendedCommunity, self).__init__(type_=type_)
self.do_init(BGPUnknownExtendedCommunity, self, kwargs, type_=type_)
@_PathAttribute.register_type(BGP_ATTR_TYPE_MP_REACH_NLRI)
@ -2178,7 +2191,7 @@ class BGPMessage(packet_base.PacketBase, _TypeDisp):
========================== ===============================================
marker Marker field. Ignored when encoding.
len Length field. Ignored when encoding.
type Type field. one of BGP\_MSG\_ constants.
type Type field. one of ``BGP_MSG_*`` constants.
========================== ===============================================
"""
@ -2186,12 +2199,15 @@ class BGPMessage(packet_base.PacketBase, _TypeDisp):
_HDR_LEN = struct.calcsize(_HDR_PACK_STR)
_class_prefixes = ['BGP']
def __init__(self, type_, len_=None, marker=None):
def __init__(self, marker=None, len_=None, type_=None):
super(BGPMessage, self).__init__()
if marker is None:
self._marker = _MARKER
else:
self._marker = marker
self.len = len_
if type_ is None:
type_ = self._rev_lookup_type(self.__class__)
self.type = type_
@classmethod
@ -2211,7 +2227,7 @@ class BGPMessage(packet_base.PacketBase, _TypeDisp):
kwargs = subcls.parser(binmsg)
return subcls(marker=marker, len_=len_, type_=type_, **kwargs), rest
def serialize(self):
def serialize(self, payload=None, prev=None):
# fixup
self._marker = _MARKER
tail = self.serialize_tail()

View File

@ -16,6 +16,8 @@
from __future__ import print_function
import os
import sys
import unittest
from nose.tools import eq_
from nose.tools import ok_
@ -25,6 +27,10 @@ from ryu.lib.packet import afi
from ryu.lib.packet import safi
BGP4_PACKET_DATA_DIR = os.path.join(
os.path.dirname(sys.modules[__name__].__file__), '../../packet_data/bgp4/')
class Test_bgp(unittest.TestCase):
""" Test case for ryu.lib.packet.bgp
"""
@ -112,7 +118,7 @@ class Test_bgp(unittest.TestCase):
]
path_attributes = [
bgp.BGPPathAttributeOrigin(value=1),
bgp.BGPPathAttributeAsPath(value=[[1000], set([1001, 1002]),
bgp.BGPPathAttributeAsPath(value=[[1000], {1001, 1002},
[1003, 1004]]),
bgp.BGPPathAttributeNextHop(value='192.0.2.199'),
bgp.BGPPathAttributeMultiExitDisc(value=2000000000),
@ -124,7 +130,7 @@ class Test_bgp(unittest.TestCase):
bgp.BGPPathAttributeOriginatorId(value='10.1.1.1'),
bgp.BGPPathAttributeClusterList(value=['1.1.1.1', '2.2.2.2']),
bgp.BGPPathAttributeExtendedCommunities(communities=ecommunities),
bgp.BGPPathAttributeAs4Path(value=[[1000000], set([1000001, 1002]),
bgp.BGPPathAttributeAs4Path(value=[[1000000], {1000001, 1002},
[1003, 1000004]]),
bgp.BGPPathAttributeAs4Aggregator(as_number=100040000,
addr='192.0.2.99'),
@ -199,11 +205,10 @@ class Test_bgp(unittest.TestCase):
# 'bgp4-update',
'bgp4-keepalive',
]
dir = '../packet_data/bgp4/'
for f in files:
print('testing %s' % f)
binmsg = open(dir + f, 'rb').read()
binmsg = open(BGP4_PACKET_DATA_DIR + f, 'rb').read()
msg, rest = bgp.BGPMessage.parser(binmsg)
binmsg2 = msg.serialize()
eq_(binmsg, binmsg2)
@ -260,7 +265,7 @@ class Test_bgp(unittest.TestCase):
]
path_attributes = [
bgp.BGPPathAttributeOrigin(value=1),
bgp.BGPPathAttributeAsPath(value=[[1000], set([1001, 1002]),
bgp.BGPPathAttributeAsPath(value=[[1000], {1001, 1002},
[1003, 1004]]),
bgp.BGPPathAttributeNextHop(value='192.0.2.199'),
bgp.BGPPathAttributeMultiExitDisc(value=2000000000),
@ -270,7 +275,7 @@ class Test_bgp(unittest.TestCase):
addr='192.0.2.99'),
bgp.BGPPathAttributeCommunities(communities=communities),
bgp.BGPPathAttributeExtendedCommunities(communities=ecommunities),
bgp.BGPPathAttributeAs4Path(value=[[1000000], set([1000001, 1002]),
bgp.BGPPathAttributeAs4Path(value=[[1000000], {1000001, 1002},
[1003, 1000004]]),
bgp.BGPPathAttributeAs4Aggregator(as_number=100040000,
addr='192.0.2.99'),