exabgp.configuration.static.attributes() - Code Metrics - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

exabgp.configuration.static.attributes() F
last analyzed 2020-07-19 19:06 UTC

↳ Parent: exabgp.configuration.static

Complexity

Conditions

Size

Total Lines	66
Code Lines	51

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	19
eloc	51
nop	1
dl	0
loc	66
rs	0.5999
c	0
b	0
f	0

How to fix Long Method Complexity

# encoding: utf-8
"""
inet/__init__.py

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

from exabgp.configuration.static.route import ParseStaticRoute
from exabgp.configuration.static.parser import prefix

from exabgp.configuration.announce.path import AnnouncePath
from exabgp.configuration.announce.label import AnnounceLabel
from exabgp.configuration.announce.vpn import AnnounceVPN

from exabgp.protocol.ip import IP
from exabgp.protocol.family import SAFI

from exabgp.bgp.message import OUT
from exabgp.bgp.message.update.nlri import CIDR
from exabgp.bgp.message.update.nlri import INET
from exabgp.bgp.message.update.nlri import Label
from exabgp.bgp.message.update.nlri import IPVPN

from exabgp.bgp.message.update.attribute import Attributes

from exabgp.rib.change import Change

from exabgp.configuration.static.mpls import label
from exabgp.configuration.static.mpls import route_distinguisher


class ParseStatic(ParseStaticRoute):
    syntax = 'route <ip>/<netmask> %s;' % ' '.join(ParseStaticRoute.definition)

    action = dict(ParseStaticRoute.action)

    name = 'static'

    def __init__(self, tokeniser, scope, error):
        ParseStaticRoute.__init__(self, tokeniser, scope, error)

    def clear(self):
        return True

    def pre(self):
        return True

    def post(self):
        return True


@ParseStatic.register('route', 'append-route')
def route(tokeniser):
    action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
    ipmask = prefix(tokeniser)
    check = lambda change, afi: True

    if 'rd' in tokeniser.tokens or 'route-distinguisher' in tokeniser.tokens:
        nlri = IPVPN(IP.toafi(ipmask.top()), SAFI.mpls_vpn, action)
        check = AnnounceVPN.check
    elif 'label' in tokeniser.tokens:
        nlri = Label(IP.toafi(ipmask.top()), SAFI.nlri_mpls, action)
        check = AnnounceLabel.check
    else:
        nlri = INET(IP.toafi(ipmask.top()), IP.tosafi(ipmask.top()), action)
        check = AnnouncePath.check

    nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)

    change = Change(nlri, Attributes())

    while True:
        command = tokeniser()

        if not command:
            break

        if command == 'label':
            nlri.labels = label(tokeniser)
            continue

        if command == 'rd' or command == 'route-distinguisher':
            nlri.rd = route_distinguisher(tokeniser)
            continue

        action = ParseStatic.action.get(command, '')

        if action == 'attribute-add':
            change.attributes.add(ParseStatic.known[command](tokeniser))
        elif action == 'nlri-set':
            change.nlri.assign(ParseStatic.assign[command], ParseStatic.known[command](tokeniser))
        elif action == 'nexthop-and-attribute':
            nexthop, attribute = ParseStatic.known[command](tokeniser)
            change.nlri.nexthop = nexthop
            change.attributes.add(attribute)
        else:
            raise ValueError('unknown command "%s"' % command)

    if not check(change, nlri.afi):
        raise ValueError('invalid route (missing next-hop, label or rd ?)')

    return list(ParseStatic.split(change))


@ParseStatic.register('attributes', 'append-route')
def attributes(tokeniser):
    action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
    ipmask = prefix(lambda: tokeniser.tokens[-1])
    tokeniser.afi = ipmask.afi

    if 'rd' in tokeniser.tokens or 'route-distinguisher' in tokeniser.tokens:
        nlri = IPVPN(IP.toafi(ipmask.top()), SAFI.mpls_vpn, action)
    elif 'label' in tokeniser.tokens:
        nlri = Label(IP.toafi(ipmask.top()), SAFI.nlri_mpls, action)
    else:
        nlri = INET(IP.toafi(ipmask.top()), IP.tosafi(ipmask.top()), action)

    nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)
    attr = Attributes()

    labels = None
    rd = None

    while True:
        command = tokeniser()

        if not command:
            return []

        if command == 'nlri':
            break

        if command == 'label':
            labels = label(tokeniser)
            continue

        if command == 'rd' or command == 'route-distinguisher':
            rd = route_distinguisher(tokeniser)
            continue

        action = ParseStatic.action[command]

        if action == 'attribute-add':
            attr.add(ParseStatic.known[command](tokeniser))
        elif action == 'nlri-set':
            nlri.assign(ParseStatic.assign[command], ParseStatic.known[command](tokeniser))
        elif action == 'nexthop-and-attribute':
            nexthop, attribute = ParseStatic.known[command](tokeniser)
            nlri.nexthop = nexthop
            attr.add(attribute)
        else:
            raise ValueError('unknown command "%s"' % command)

    changes = []
    while True:
        peeked_nlri = tokeniser.peek()
        if not peeked_nlri:
            break

        ipmask = prefix(tokeniser)
        new = Change(nlri.__class__(nlri.afi, nlri.safi, OUT.UNSET), attr)
        new.nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)
        if labels:
            new.nlri.labels = labels
        if rd:
            new.nlri.rd = rd
        new.nlri.nexthop = nlri.nexthop
        changes.append(new)

    return changes


1			# encoding: utf-8
2			"""
3			inet/__init__.py
4
5			Created by Thomas Mangin on 2015-06-04.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			from exabgp.configuration.static.route import ParseStaticRoute
11			from exabgp.configuration.static.parser import prefix
12
13			from exabgp.configuration.announce.path import AnnouncePath
14			from exabgp.configuration.announce.label import AnnounceLabel
15			from exabgp.configuration.announce.vpn import AnnounceVPN
16
17			from exabgp.protocol.ip import IP
18			from exabgp.protocol.family import SAFI
19
20			from exabgp.bgp.message import OUT
21			from exabgp.bgp.message.update.nlri import CIDR
22			from exabgp.bgp.message.update.nlri import INET
23			from exabgp.bgp.message.update.nlri import Label
24			from exabgp.bgp.message.update.nlri import IPVPN
25
26			from exabgp.bgp.message.update.attribute import Attributes
27
28			from exabgp.rib.change import Change
29
30			from exabgp.configuration.static.mpls import label
31			from exabgp.configuration.static.mpls import route_distinguisher
32
33
34			class ParseStatic(ParseStaticRoute):
35			syntax = 'route <ip>/<netmask> %s;' % ' '.join(ParseStaticRoute.definition)
36
37			action = dict(ParseStaticRoute.action)
38
39			name = 'static'
40
41			def __init__(self, tokeniser, scope, error):
42			ParseStaticRoute.__init__(self, tokeniser, scope, error)
43
44			def clear(self):
45			return True
46
47			def pre(self):
48			return True
49
50			def post(self):
51			return True
52
53
54			@ParseStatic.register('route', 'append-route')
55			def route(tokeniser):
56			action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
57			ipmask = prefix(tokeniser)
58			check = lambda change, afi: True
59
60			if 'rd' in tokeniser.tokens or 'route-distinguisher' in tokeniser.tokens:
61			nlri = IPVPN(IP.toafi(ipmask.top()), SAFI.mpls_vpn, action)
62			check = AnnounceVPN.check
63			elif 'label' in tokeniser.tokens:
64			nlri = Label(IP.toafi(ipmask.top()), SAFI.nlri_mpls, action)
65			check = AnnounceLabel.check
66			else:
67			nlri = INET(IP.toafi(ipmask.top()), IP.tosafi(ipmask.top()), action)
68			check = AnnouncePath.check
69
70			nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)
71
72			change = Change(nlri, Attributes())
73
74			while True:
75			command = tokeniser()
76
77			if not command:
78			break
79
80			if command == 'label':
81			nlri.labels = label(tokeniser)
82			continue
83
84			if command == 'rd' or command == 'route-distinguisher':
85			nlri.rd = route_distinguisher(tokeniser)
86			continue
87
88			action = ParseStatic.action.get(command, '')
89
90			if action == 'attribute-add':
91			change.attributes.add(ParseStatic.known[command](tokeniser))
92			elif action == 'nlri-set':
93			change.nlri.assign(ParseStatic.assign[command], ParseStatic.known[command](tokeniser))
94			elif action == 'nexthop-and-attribute':
95			nexthop, attribute = ParseStatic.known[command](tokeniser)
96			change.nlri.nexthop = nexthop
97			change.attributes.add(attribute)
98			else:
99			raise ValueError('unknown command "%s"' % command)
100
101			if not check(change, nlri.afi):
102			raise ValueError('invalid route (missing next-hop, label or rd ?)')
103
104			return list(ParseStatic.split(change))
105
106
107			@ParseStatic.register('attributes', 'append-route')
108			def attributes(tokeniser):
109			action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
110			ipmask = prefix(lambda: tokeniser.tokens[-1])
111			tokeniser.afi = ipmask.afi
112
113			if 'rd' in tokeniser.tokens or 'route-distinguisher' in tokeniser.tokens:
114			nlri = IPVPN(IP.toafi(ipmask.top()), SAFI.mpls_vpn, action)
115			elif 'label' in tokeniser.tokens:
116			nlri = Label(IP.toafi(ipmask.top()), SAFI.nlri_mpls, action)
117			else:
118			nlri = INET(IP.toafi(ipmask.top()), IP.tosafi(ipmask.top()), action)
119
120			nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)
121			attr = Attributes()
122
123			labels = None
124			rd = None
125
126			while True:
127			command = tokeniser()
128
129			if not command:
130			return []
131
132			if command == 'nlri':
133			break
134
135			if command == 'label':
136			labels = label(tokeniser)
137			continue
138
139			if command == 'rd' or command == 'route-distinguisher':
140			rd = route_distinguisher(tokeniser)
141			continue
142
143			action = ParseStatic.action[command]
144
145			if action == 'attribute-add':
146			attr.add(ParseStatic.known[command](tokeniser))
147			elif action == 'nlri-set':
148			nlri.assign(ParseStatic.assign[command], ParseStatic.known[command](tokeniser))
149			elif action == 'nexthop-and-attribute':
150			nexthop, attribute = ParseStatic.known[command](tokeniser)
151			nlri.nexthop = nexthop
152			attr.add(attribute)
153			else:
154			raise ValueError('unknown command "%s"' % command)
155
156			changes = []
157			while True:
158			peeked_nlri = tokeniser.peek()
159			if not peeked_nlri:
160			break
161
162			ipmask = prefix(tokeniser)
163			new = Change(nlri.__class__(nlri.afi, nlri.safi, OUT.UNSET), attr)
164			new.nlri.cidr = CIDR(ipmask.pack(), ipmask.mask)
165			if labels:
166			new.nlri.labels = labels
167			if rd:
168			new.nlri.rd = rd
169			new.nlri.nexthop = nlri.nexthop
170			changes.append(new)
171
172			return changes
173

Exa-Networks / exabgp

exabgp.configuration.static.attributes() F last analyzed 2020-07-19 19:06 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

exabgp.configuration.static.attributes() F
last analyzed 2020-07-19 19:06 UTC