exabgp.protocol.family._SAFI.__repr__() - Code Metrics - Inspection of "replace _index() with Family.index()" - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 0517dc...c6929c )

by Thomas

created 2020-05-12 09:06 UTC

exabgp.protocol.family._SAFI.repr() A

↳ Parent: exabgp.protocol.family

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0

# encoding: utf-8
"""
family.py

Created by Thomas Mangin on 2010-01-19.
Copyright (c) 2009-2017 Exa Networks. All rights reserved.
License: 3-clause BSD. (See the COPYRIGHT file)
"""

from struct import pack
from struct import unpack

from exabgp.util import character
from exabgp.util import ordinal
from exabgp.protocol.resource import Resource


# ======================================================================== AFI
# https://www.iana.org/assignments/address-family-numbers/


class _AFI(int):
    UNDEFINED = 0x00  # internal
    IPv4 = 0x01
    IPv6 = 0x02
    L2VPN = 0x19
    BGPLS = 0x4004

    _names = {
        UNDEFINED: 'undefined',
        IPv4: 'ipv4',
        IPv6: 'ipv6',
        L2VPN: 'l2vpn',
        BGPLS: 'bgp-ls',
    }

    _masks = {
        IPv4: 32,
        IPv6: 128,
    }

    def pack(self):
        return pack('!H', self)

    def name(self):
        return self._names.get(self, 'unknown-afi-0x%s' % hex(self))

    def mask(self):
        return self._masks.get(self, 'invalid request for this family')

    def __repr__(self):
        return self.name()

    def __str__(self):
        return self.name()


class AFI(Resource):
    undefined = _AFI(_AFI.UNDEFINED)
    ipv4 = _AFI(_AFI.IPv4)
    ipv6 = _AFI(_AFI.IPv6)
    l2vpn = _AFI(_AFI.L2VPN)
    bgpls = _AFI(_AFI.BGPLS)

    common = {
        undefined.pack(): undefined,
        ipv4.pack(): ipv4,
        ipv6.pack(): ipv6,
        l2vpn.pack(): l2vpn,
        bgpls.pack(): bgpls,
    }

    codes = dict(
        (k.lower().replace('_', '-'), v)
        for (k, v) in {'ipv4': ipv4, 'ipv6': ipv6, 'l2vpn': l2vpn, 'bgp-ls': bgpls,}.items()
    )

    cache = dict([(r, r) for (l, r) in codes.items()])
    names = dict([(r, l) for (l, r) in codes.items()])

    inet_names = dict([(r, l.replace('ipv', 'inet')) for (l, r) in codes.items()])

    def name(self):
        return self.inet_names.get(self, "unknown afi")

    @staticmethod
    def unpack(data):
        return AFI.common.get(data, _AFI(unpack('!H', data)[0]))

    @classmethod
    def value(cls, name):
        return cls.codes.get(name, None)

    @staticmethod
    def implemented_safi(afi):
        if afi == 'ipv4':
            return ['unicast', 'multicast', 'nlri-mpls', 'mpls-vpn', 'flow', 'flow-vpn']
        if afi == 'ipv6':
            return ['unicast', 'mpls-vpn', 'flow', 'flow-vpn']
        if afi == 'l2vpn':
            return ['vpls', 'evpn']
        if afi == 'bgp-ls':
            return ['bgp-ls', 'bgp-ls-vpn']
        return []

    @classmethod
    def fromString(cls, string):
        return cls.codes.get(string, cls.undefined)

    @classmethod
    def create(cls, value):
        return cls.cache.get(value, _AFI(value))


# ======================================================================= SAFI
# https://www.iana.org/assignments/safi-namespace


class _SAFI(int):
    UNDEFINED = 0  # internal
    UNICAST = 1  # [RFC4760]
    MULTICAST = 2  # [RFC4760]
    NLRI_MPLS = 4  # [RFC3107]
    VPLS = 65  # [RFC4761]
    EVPN = 70  # [draft-ietf-l2vpn-evpn]
    BGPLS = 71  # [RFC7752]
    BGPLS_VPN = 72  # [RFC7752]
    MPLS_VPN = 128  # [RFC4364]
    RTC = 132  # [RFC4684]
    FLOW_IP = 133  # [RFC5575]
    FLOW_VPN = 134  # [RFC5575]
    # deprecated = 3            # [RFC4760]
    # mcast_vpn = 5             # [draft-ietf-l3vpn-2547bis-mcast-bgp] (TEMPORARY - Expires 2008-06-19)
    # pseudowire = 6            # [draft-ietf-pwe3-dynamic-ms-pw] (TEMPORARY - Expires 2008-08-23) Dynamic Placement of Multi-Segment Pseudowires
    # encapsulation = 7         # [RFC5512]
    # tunel = 64                # [Nalawade]
    # bgp_mdt = 66              # [Nalawade]
    # bgp_4over6 = 67           # [Cui]
    # bgp_6over4 = 67           # [Cui]
    # vpn_adi = 69              # [RFC-ietf-l1vpn-bgp-auto-discovery-05.txt]
    # mcast_bgp_mpls_vpn = 129  # [RFC2547]
    # rt = 132                  # [RFC4684]
    # vpn_ad = 140              # [draft-ietf-l3vpn-bgpvpn-auto]
    # private = [_ for _ in range(241,254)]   # [RFC4760]
    # unassigned = [_ for _ in range(8,64)] + [_ for _ in range(70,128)]
    # reverved = [0,3] + [130,131] + [_ for _ in range(135,140)] + [_ for _ in range(141,241)] + [255,]    # [RFC4760]

    _names = {
        UNICAST: 'unicast',
        MULTICAST: 'multicast',
        NLRI_MPLS: 'nlri-mpls',
        VPLS: 'vpls',
        EVPN: 'evpn',
        BGPLS: 'bgp-ls',
        BGPLS_VPN: 'bgp-ls-vpn',
        MPLS_VPN: 'mpls-vpn',
        RTC: 'rtc',
        FLOW_IP: 'flow',
        FLOW_VPN: 'flow-vpn',
    }

    def pack(self):
        return character(self)

    def name(self):
        return self._names.get(self, 'unknown safi %d' % int(self))

    def has_label(self):
        return self in (SAFI.nlri_mpls, SAFI.mpls_vpn)

    def has_rd(self):
        return self in (SAFI.nlri_mpls, SAFI.mpls_vpn, SAFI.flow_vpn)
        # technically self.flow_vpn and self.vpls has an RD but it is not an NLRI

    def has_path(self):
        return self in (SAFI.unicast, SAFI.nlri_mpls)
        # technically self.flow_vpn and self.vpls has an RD but it is not an NLRI

    def __str__(self):
        return self.name()

    def __repr__(self):
        return str(self)


class SAFI(Resource):
    undefined = _SAFI(_SAFI.UNDEFINED)
    unicast = _SAFI(_SAFI.UNICAST)
    multicast = _SAFI(_SAFI.MULTICAST)
    nlri_mpls = _SAFI(_SAFI.NLRI_MPLS)
    vpls = _SAFI(_SAFI.VPLS)
    evpn = _SAFI(_SAFI.EVPN)
    bgp_ls = _SAFI(_SAFI.BGPLS)
    bgp_ls_vpn = _SAFI(_SAFI.BGPLS_VPN)
    mpls_vpn = _SAFI(_SAFI.MPLS_VPN)
    rtc = _SAFI(_SAFI.RTC)
    flow_ip = _SAFI(_SAFI.FLOW_IP)
    flow_vpn = _SAFI(_SAFI.FLOW_VPN)

    common = {
        undefined.pack(): undefined,
        unicast.pack(): unicast,
        multicast.pack(): multicast,
        nlri_mpls.pack(): nlri_mpls,
        vpls.pack(): vpls,
        evpn.pack(): evpn,
        bgp_ls.pack(): bgp_ls,
        bgp_ls_vpn.pack(): bgp_ls_vpn,
        mpls_vpn.pack(): mpls_vpn,
        rtc.pack(): rtc,
        flow_ip.pack(): flow_ip,
        flow_vpn.pack(): flow_vpn,
    }

    codes = dict(
        (k.lower().replace('_', '-'), v)
        for (k, v) in {
            'unicast': unicast,
            'multicast': multicast,
            'nlri-mpls': nlri_mpls,
            'vpls': vpls,
            'evpn': evpn,
            'bgp-ls': bgp_ls,
            'bgp-ls-vpn': bgp_ls_vpn,
            'mpls-vpn': mpls_vpn,
            'rtc': rtc,
            'flow': flow_ip,
            'flow-vpn': flow_vpn,
        }.items()
    )

    names = _SAFI._names

    cache = dict([(r, r) for (l, r) in codes.items()])

    @staticmethod
    def unpack(data):
        return SAFI.common.get(data, _SAFI(ordinal(data)))

    @classmethod
    def value(cls, name):
        return cls.codes.get(name, None)

    @classmethod
    def fromString(cls, string):
        return cls.codes.get(string, cls.undefined)

    @classmethod
    def create(cls, value):
        return cls.cache.get(value, _SAFI(value))


# ===================================================================== FAMILY


class Family(object):
    size = {
        # family                   next-hop   RD
        (AFI.ipv4, SAFI.unicast): ((4,), 0),
        (AFI.ipv4, SAFI.multicast): ((4,), 0),
        (AFI.ipv4, SAFI.nlri_mpls): ((4,), 0),
        (AFI.ipv4, SAFI.mpls_vpn): ((12,), 8),
        (AFI.ipv4, SAFI.flow_ip): ((0, 4), 0),
        (AFI.ipv4, SAFI.flow_vpn): ((0, 4), 0),
        (AFI.ipv4, SAFI.rtc): ((4, 16), 0),
        (AFI.ipv6, SAFI.unicast): ((16, 32), 0),
        (AFI.ipv6, SAFI.nlri_mpls): ((16, 32), 0),
        (AFI.ipv6, SAFI.mpls_vpn): ((24, 40), 8),
        (AFI.ipv6, SAFI.flow_ip): ((0, 16, 32), 0),
        (AFI.ipv6, SAFI.flow_vpn): ((0, 16, 32), 0),
        (AFI.l2vpn, SAFI.vpls): ((4,), 0),
        (AFI.l2vpn, SAFI.evpn): ((4,), 0),
        (AFI.bgpls, SAFI.bgp_ls): ((4,), 0),
    }

    __slots__ = ['afi', 'safi']

    def __init__(self, afi, safi):
        self.afi = AFI.create(afi)
        self.safi = SAFI.create(safi)

    def has_label(self):
        return self.safi.has_label()

    def has_rd(self):
        return self.safi.has_rd()

    def has_path(self):
        return self.safi.has_path()

    def __eq__(self, other):
        return self.afi == other.afi and self.safi == other.safi

    def __neq__(self, other):
        return self.afi != other.afi or self.safi != other.safi

    def __lt__(self, other):
        raise RuntimeError('comparing Family for ordering does not make sense')

    def __le__(self, other):
        raise RuntimeError('comparing Family for ordering does not make sense')

    def __gt__(self, other):
        raise RuntimeError('comparing Family for ordering does not make sense')

    def __ge__(self, other):
        raise RuntimeError('comparing Family for ordering does not make sense')

    def family(self):
        return (self.afi, self.safi)

    def extensive(self):
        return 'afi %s safi %s' % (self.afi, self.safi)

    def index(self):
        return b'%02x%02x' % (self.afi, self.safi)

    def __repr__(self):
        return "%s %s" % (str(self.afi), str(self.safi))


1			# encoding: utf-8
2			"""
3			family.py
4
5			Created by Thomas Mangin on 2010-01-19.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			from struct import pack
11			from struct import unpack
12
13			from exabgp.util import character
14			from exabgp.util import ordinal
15			from exabgp.protocol.resource import Resource
16
17
18			# ======================================================================== AFI
19			# https://www.iana.org/assignments/address-family-numbers/
20
21
22			class _AFI(int):
23			UNDEFINED = 0x00 # internal
24			IPv4 = 0x01
25			IPv6 = 0x02
26			L2VPN = 0x19
27			BGPLS = 0x4004
28
29			_names = {
30			UNDEFINED: 'undefined',
31			IPv4: 'ipv4',
32			IPv6: 'ipv6',
33			L2VPN: 'l2vpn',
34			BGPLS: 'bgp-ls',
35			}
36
37			_masks = {
38			IPv4: 32,
39			IPv6: 128,
40			}
41
42			def pack(self):
43			return pack('!H', self)
44
45			def name(self):
46			return self._names.get(self, 'unknown-afi-0x%s' % hex(self))
47
48			def mask(self):
49			return self._masks.get(self, 'invalid request for this family')
50
51			def __repr__(self):
52			return self.name()
53
54			def __str__(self):
55			return self.name()
56
57
58			class AFI(Resource):
59			undefined = _AFI(_AFI.UNDEFINED)
60			ipv4 = _AFI(_AFI.IPv4)
61			ipv6 = _AFI(_AFI.IPv6)
62			l2vpn = _AFI(_AFI.L2VPN)
63			bgpls = _AFI(_AFI.BGPLS)
64
65			common = {
66			undefined.pack(): undefined,
67			ipv4.pack(): ipv4,
68			ipv6.pack(): ipv6,
69			l2vpn.pack(): l2vpn,
70			bgpls.pack(): bgpls,
71			}
72
73			codes = dict(
74			(k.lower().replace('_', '-'), v)
75			for (k, v) in {'ipv4': ipv4, 'ipv6': ipv6, 'l2vpn': l2vpn, 'bgp-ls': bgpls,}.items()
76			)
77
78			cache = dict([(r, r) for (l, r) in codes.items()])
79			names = dict([(r, l) for (l, r) in codes.items()])
80
81			inet_names = dict([(r, l.replace('ipv', 'inet')) for (l, r) in codes.items()])
82
83			def name(self):
84			return self.inet_names.get(self, "unknown afi")
85
86			@staticmethod
87			def unpack(data):
88			return AFI.common.get(data, _AFI(unpack('!H', data)[0]))
89
90			@classmethod
91			def value(cls, name):
92			return cls.codes.get(name, None)
93
94			@staticmethod
95			def implemented_safi(afi):
96			if afi == 'ipv4':
97			return ['unicast', 'multicast', 'nlri-mpls', 'mpls-vpn', 'flow', 'flow-vpn']
98			if afi == 'ipv6':
99			return ['unicast', 'mpls-vpn', 'flow', 'flow-vpn']
100			if afi == 'l2vpn':
101			return ['vpls', 'evpn']
102			if afi == 'bgp-ls':
103			return ['bgp-ls', 'bgp-ls-vpn']
104			return []
105
106			@classmethod
107			def fromString(cls, string):
108			return cls.codes.get(string, cls.undefined)
109
110			@classmethod
111			def create(cls, value):
112			return cls.cache.get(value, _AFI(value))
113
114
115			# ======================================================================= SAFI
116			# https://www.iana.org/assignments/safi-namespace
117
118
119			class _SAFI(int):
120			UNDEFINED = 0 # internal
121			UNICAST = 1 # [RFC4760]
122			MULTICAST = 2 # [RFC4760]
123			NLRI_MPLS = 4 # [RFC3107]
124			VPLS = 65 # [RFC4761]
125			EVPN = 70 # [draft-ietf-l2vpn-evpn]
126			BGPLS = 71 # [RFC7752]
127			BGPLS_VPN = 72 # [RFC7752]
128			MPLS_VPN = 128 # [RFC4364]
129			RTC = 132 # [RFC4684]
130			FLOW_IP = 133 # [RFC5575]
131			FLOW_VPN = 134 # [RFC5575]
132			# deprecated = 3 # [RFC4760]
133			# mcast_vpn = 5 # [draft-ietf-l3vpn-2547bis-mcast-bgp] (TEMPORARY - Expires 2008-06-19)
134			# pseudowire = 6 # [draft-ietf-pwe3-dynamic-ms-pw] (TEMPORARY - Expires 2008-08-23) Dynamic Placement of Multi-Segment Pseudowires
135			# encapsulation = 7 # [RFC5512]
136			# tunel = 64 # [Nalawade]
137			# bgp_mdt = 66 # [Nalawade]
138			# bgp_4over6 = 67 # [Cui]
139			# bgp_6over4 = 67 # [Cui]
140			# vpn_adi = 69 # [RFC-ietf-l1vpn-bgp-auto-discovery-05.txt]
141			# mcast_bgp_mpls_vpn = 129 # [RFC2547]
142			# rt = 132 # [RFC4684]
143			# vpn_ad = 140 # [draft-ietf-l3vpn-bgpvpn-auto]
144			# private = [_ for _ in range(241,254)] # [RFC4760]
145			# unassigned = [_ for _ in range(8,64)] + [_ for _ in range(70,128)]
146			# reverved = [0,3] + [130,131] + [_ for _ in range(135,140)] + [_ for _ in range(141,241)] + [255,] # [RFC4760]
147
148			_names = {
149			UNICAST: 'unicast',
150			MULTICAST: 'multicast',
151			NLRI_MPLS: 'nlri-mpls',
152			VPLS: 'vpls',
153			EVPN: 'evpn',
154			BGPLS: 'bgp-ls',
155			BGPLS_VPN: 'bgp-ls-vpn',
156			MPLS_VPN: 'mpls-vpn',
157			RTC: 'rtc',
158			FLOW_IP: 'flow',
159			FLOW_VPN: 'flow-vpn',
160			}
161
162			def pack(self):
163			return character(self)
164
165			def name(self):
166			return self._names.get(self, 'unknown safi %d' % int(self))
167
168			def has_label(self):
169			return self in (SAFI.nlri_mpls, SAFI.mpls_vpn)
170
171			def has_rd(self):
172			return self in (SAFI.nlri_mpls, SAFI.mpls_vpn, SAFI.flow_vpn)
173			# technically self.flow_vpn and self.vpls has an RD but it is not an NLRI
174
175			def has_path(self):
176			return self in (SAFI.unicast, SAFI.nlri_mpls)
177			# technically self.flow_vpn and self.vpls has an RD but it is not an NLRI
178
179			def __str__(self):
180			return self.name()
181
182			def __repr__(self):
183			return str(self)
184
185
186			class SAFI(Resource):
187			undefined = _SAFI(_SAFI.UNDEFINED)
188			unicast = _SAFI(_SAFI.UNICAST)
189			multicast = _SAFI(_SAFI.MULTICAST)
190			nlri_mpls = _SAFI(_SAFI.NLRI_MPLS)
191			vpls = _SAFI(_SAFI.VPLS)
192			evpn = _SAFI(_SAFI.EVPN)
193			bgp_ls = _SAFI(_SAFI.BGPLS)
194			bgp_ls_vpn = _SAFI(_SAFI.BGPLS_VPN)
195			mpls_vpn = _SAFI(_SAFI.MPLS_VPN)
196			rtc = _SAFI(_SAFI.RTC)
197			flow_ip = _SAFI(_SAFI.FLOW_IP)
198			flow_vpn = _SAFI(_SAFI.FLOW_VPN)
199
200			common = {
201			undefined.pack(): undefined,
202			unicast.pack(): unicast,
203			multicast.pack(): multicast,
204			nlri_mpls.pack(): nlri_mpls,
205			vpls.pack(): vpls,
206			evpn.pack(): evpn,
207			bgp_ls.pack(): bgp_ls,
208			bgp_ls_vpn.pack(): bgp_ls_vpn,
209			mpls_vpn.pack(): mpls_vpn,
210			rtc.pack(): rtc,
211			flow_ip.pack(): flow_ip,
212			flow_vpn.pack(): flow_vpn,
213			}
214
215			codes = dict(
216			(k.lower().replace('_', '-'), v)
217			for (k, v) in {
218			'unicast': unicast,
219			'multicast': multicast,
220			'nlri-mpls': nlri_mpls,
221			'vpls': vpls,
222			'evpn': evpn,
223			'bgp-ls': bgp_ls,
224			'bgp-ls-vpn': bgp_ls_vpn,
225			'mpls-vpn': mpls_vpn,
226			'rtc': rtc,
227			'flow': flow_ip,
228			'flow-vpn': flow_vpn,
229			}.items()
230			)
231
232			names = _SAFI._names
233
234			cache = dict([(r, r) for (l, r) in codes.items()])
235
236			@staticmethod
237			def unpack(data):
238			return SAFI.common.get(data, _SAFI(ordinal(data)))
239
240			@classmethod
241			def value(cls, name):
242			return cls.codes.get(name, None)
243
244			@classmethod
245			def fromString(cls, string):
246			return cls.codes.get(string, cls.undefined)
247
248			@classmethod
249			def create(cls, value):
250			return cls.cache.get(value, _SAFI(value))
251
252
253			# ===================================================================== FAMILY
254
255
256			class Family(object):
257			size = {
258			# family next-hop RD
259			(AFI.ipv4, SAFI.unicast): ((4,), 0),
260			(AFI.ipv4, SAFI.multicast): ((4,), 0),
261			(AFI.ipv4, SAFI.nlri_mpls): ((4,), 0),
262			(AFI.ipv4, SAFI.mpls_vpn): ((12,), 8),
263			(AFI.ipv4, SAFI.flow_ip): ((0, 4), 0),
264			(AFI.ipv4, SAFI.flow_vpn): ((0, 4), 0),
265			(AFI.ipv4, SAFI.rtc): ((4, 16), 0),
266			(AFI.ipv6, SAFI.unicast): ((16, 32), 0),
267			(AFI.ipv6, SAFI.nlri_mpls): ((16, 32), 0),
268			(AFI.ipv6, SAFI.mpls_vpn): ((24, 40), 8),
269			(AFI.ipv6, SAFI.flow_ip): ((0, 16, 32), 0),
270			(AFI.ipv6, SAFI.flow_vpn): ((0, 16, 32), 0),
271			(AFI.l2vpn, SAFI.vpls): ((4,), 0),
272			(AFI.l2vpn, SAFI.evpn): ((4,), 0),
273			(AFI.bgpls, SAFI.bgp_ls): ((4,), 0),
274			}
275
276			__slots__ = ['afi', 'safi']
277
278			def __init__(self, afi, safi):
279			self.afi = AFI.create(afi)
280			self.safi = SAFI.create(safi)
281
282			def has_label(self):
283			return self.safi.has_label()
284
285			def has_rd(self):
286			return self.safi.has_rd()
287
288			def has_path(self):
289			return self.safi.has_path()
290
291			def __eq__(self, other):
292			return self.afi == other.afi and self.safi == other.safi
293
294			def __neq__(self, other):
295			return self.afi != other.afi or self.safi != other.safi
296
297			def __lt__(self, other):
298			raise RuntimeError('comparing Family for ordering does not make sense')
299
300			def __le__(self, other):
301			raise RuntimeError('comparing Family for ordering does not make sense')
302
303			def __gt__(self, other):
304			raise RuntimeError('comparing Family for ordering does not make sense')
305
306			def __ge__(self, other):
307			raise RuntimeError('comparing Family for ordering does not make sense')
308
309			def family(self):
310			return (self.afi, self.safi)
311
312			def extensive(self):
313			return 'afi %s safi %s' % (self.afi, self.safi)
314
315			def index(self):
316			return b'%02x%02x' % (self.afi, self.safi)
317
318			def __repr__(self):
319			return "%s %s" % (str(self.afi), str(self.safi))
320

Exa-Networks / exabgp

Push — master ( 0517dc...c6929c )

exabgp.protocol.family._SAFI.__repr__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

exabgp.protocol.family._SAFI.repr() A