exabgp.configuration.static.ParseStatic.post() - Code Metrics - Inspection of "large change required to perform better route vali..." - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 25ce4f...b9bc4f )

by Thomas

created 2019-09-24 00:20 UTC

exabgp.configuration.static.ParseStatic.post() A

↳ Parent: exabgp.configuration.static

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
"""
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, logger):
		ParseStaticRoute.__init__(self,tokeniser,scope,error,logger)

	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])

	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			class ParseStatic (ParseStaticRoute):
34			syntax = \
35			'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, logger):
42			ParseStaticRoute.__init__(self,tokeniser,scope,error,logger)
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(
73			nlri,
74			Attributes()
75			)
76
77			while True:
78			command = tokeniser()
79
80			if not command:
81			break
82
83			if command == 'label':
84			nlri.labels = label(tokeniser)
85			continue
86
87			if command == 'rd' or command == 'route-distinguisher':
88			nlri.rd = route_distinguisher(tokeniser)
89			continue
90
91			action = ParseStatic.action.get(command,'')
92
93			if action == 'attribute-add':
94			change.attributes.add(ParseStatic.known[command](tokeniser))
95			elif action == 'nlri-set':
96			change.nlri.assign(ParseStatic.assign[command],ParseStatic.known[command](tokeniser))
97			elif action == 'nexthop-and-attribute':
98			nexthop,attribute = ParseStatic.known[command](tokeniser)
99			change.nlri.nexthop = nexthop
100			change.attributes.add(attribute)
101			else:
102			raise ValueError('unknown command "%s"' % command)
103
104			if not check(change,nlri.afi):
105			raise ValueError('invalid route (missing next-hop, label or rd ?)')
106
107			return list(ParseStatic.split(change))
108
109
110			@ParseStatic.register('attributes','append-route')
111			def attributes (tokeniser):
112			action = OUT.ANNOUNCE if tokeniser.announce else OUT.WITHDRAW
113			ipmask = prefix(lambda: tokeniser.tokens[-1])
114
115			if 'rd' in tokeniser.tokens or 'route-distinguisher' in tokeniser.tokens:
116			nlri = IPVPN(IP.toafi(ipmask.top()), SAFI.mpls_vpn, action)
117			elif 'label' in tokeniser.tokens:
118			nlri = Label(IP.toafi(ipmask.top()), SAFI.nlri_mpls, action)
119			else:
120			nlri = INET(IP.toafi(ipmask.top()), IP.tosafi(ipmask.top()), action)
121
122			nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)
123			attr = Attributes()
124
125			labels = None
126			rd = None
127
128			while True:
129			command = tokeniser()
130
131			if not command:
132			return []
133
134			if command == 'nlri':
135			break
136
137			if command == 'label':
138			labels = label(tokeniser)
139			continue
140
141			if command == 'rd' or command == 'route-distinguisher':
142			rd = route_distinguisher(tokeniser)
143			continue
144
145			action = ParseStatic.action[command]
146
147			if action == 'attribute-add':
148			attr.add(ParseStatic.known[command](tokeniser))
149			elif action == 'nlri-set':
150			nlri.assign(ParseStatic.assign[command],ParseStatic.known[command](tokeniser))
151			elif action == 'nexthop-and-attribute':
152			nexthop,attribute = ParseStatic.known[command](tokeniser)
153			nlri.nexthop = nexthop
154			attr.add(attribute)
155			else:
156			raise ValueError('unknown command "%s"' % command)
157
158			changes = []
159			while True:
160			peeked_nlri = tokeniser.peek()
161			if not peeked_nlri:
162			break
163
164			ipmask = prefix(tokeniser)
165			new = Change(
166			nlri.__class__(
167			nlri.afi,
168			nlri.safi,
169			OUT.UNSET
170			),
171			attr
172			)
173			new.nlri.cidr = CIDR(ipmask.pack(),ipmask.mask)
174			if labels:
175			new.nlri.labels = labels
176			if rd:
177			new.nlri.rd = rd
178			new.nlri.nexthop = nlri.nexthop
179			changes.append(new)
180
181			return changes
182

Exa-Networks / exabgp

Push — master ( 25ce4f...b9bc4f )

exabgp.configuration.static.ParseStatic.post() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like