exabgp.configuration.check.check_generation() - Code Metrics - Inspection of "move configuration testing in its own section" - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( ea37f3...9aacac )

by Thomas

created 2020-05-17 00:13 UTC

exabgp.configuration.check.check_generation() F

↳ Parent: exabgp.configuration.check

Complexity

Conditions

Size

Total Lines	115
Code Lines	83

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	17
eloc	83
nop	1
dl	0
loc	115
rs	1.8
c	0
b	0
f	0

How to fix Long Method Complexity

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

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

# common

import copy

from exabgp.bgp.message import Update
from exabgp.bgp.message import Open
from exabgp.bgp.message.open import Version
from exabgp.bgp.message.open import ASN
from exabgp.bgp.message.open import RouterID
from exabgp.bgp.message.open import HoldTime
from exabgp.bgp.message.open.capability import Capabilities
from exabgp.bgp.message.open.capability import Capability
from exabgp.bgp.message.open.capability import Negotiated
from exabgp.bgp.message import Notify
from exabgp.bgp.message.update.nlri import NLRI

from exabgp.logger import log

# check_neighbor

from exabgp.util.od import od
from exabgp.rib.change import Change

# check_update

from exabgp.reactor.api.response import Response

# check_notification

from exabgp.bgp.message import Notification

# JSON version

from exabgp.version import json as json_version


# =============================================================== check_neighbor
# ...


def check_generation(neighbors):
    log._option['parser'] = True

    for name in neighbors.keys():
        neighbor = copy.deepcopy(neighbors[name])
        neighbor.local_as = neighbor.peer_as

        path = {}
        for f in NLRI.known_families():
            if neighbor.add_path:
                path[f] = neighbor.add_path

        capa = Capabilities().new(neighbor, False)
        if path:
            capa[Capability.CODE.ADD_PATH] = path
        capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()

        routerid_1 = str(neighbor.router_id)
        routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))

        o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
        o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
        negotiated = Negotiated(neighbor)
        negotiated.sent(o1)
        negotiated.received(o2)
        # grouped = False

        for _ in neighbor.rib.outgoing.updates(False):
            pass

        for change1 in neighbor.rib.outgoing.cached_changes():
            str1 = change1.extensive()
            packed = list(Update([change1.nlri], change1.attributes).messages(negotiated))
            pack1 = packed[0]

            log.debug('parsed route requires %d updates' % len(packed), 'parser')
            log.debug('update size is %d' % len(pack1), 'parser')

            log.debug('parsed route %s' % str1, 'parser')
            log.debug('parsed hex   %s' % od(pack1), 'parser')

            # This does not take the BGP header - let's assume we will not break that :)
            try:
                log.debug('')  # new line

                pack1s = pack1[19:] if pack1.startswith(b'\xFF' * 16) else pack1
                update = Update.unpack_message(pack1s, negotiated)

                change2 = Change(update.nlris[0], update.attributes)
                str2 = change2.extensive()
                pack2 = list(Update([update.nlris[0]], update.attributes).messages(negotiated))[0]

                log.debug('recoded route %s' % str2, 'parser')
                log.debug('recoded hex   %s' % od(pack2), 'parser')

                str1 = str1.replace('attribute [ 0x04 0x80 0x00000064 ]', 'med 100')
                str1r = (
                    str1.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
                )
                str2r = (
                    str2.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
                )
                str2r = str2r.replace(
                    'large-community [ 1:2:3 10:11:12 ]',
                    'attribute [ 0x20 0xc0 0x0000000100000002000000030000000a0000000b0000000c ]',
                )

                if 'next-hop self' in str1r:
                    if ':' in str1r:
                        str1r = str1r.replace('next-hop self', 'next-hop ::1')
                    else:
                        str1r = str1r.replace('next-hop self', 'next-hop %s' % neighbor.local_address)

                if ' name ' in str1r:
                    parts = str1r.split(' ')
                    pos = parts.index('name')
                    str1r = ' '.join(parts[:pos] + parts[pos + 2 :])

                skip = False

                if str1r != str2r:
                    if 'attribute [' in str1r and ' 0x00 ' in str1r:
                        # we do not decode non-transitive attributes
                        log.debug('skipping string check on update with non-transitive attribute(s)', 'parser')
                        skip = True
                    else:
                        log.debug('strings are different:', 'parser')
                        log.debug('[%s]' % (str1r), 'parser')
                        log.debug('[%s]' % (str2r), 'parser')
                        return False
                else:
                    log.debug('strings are fine', 'parser')

                if skip:
                    log.debug('skipping encoding for update with non-transitive attribute(s)', 'parser')
                elif pack1 != pack2:
                    log.debug('encoding are different', 'parser')
                    log.debug('[%s]' % (od(pack1)), 'parser')
                    log.debug('[%s]' % (od(pack2)), 'parser')
                    return False
                else:
                    log.debug('encoding is fine', 'parser')
                    log.debug('----------------------------------------', 'parser')

                log.debug('JSON nlri %s' % change1.nlri.json(), 'parser')
                log.debug('JSON attr %s' % change1.attributes.json(), 'parser')

            except Notify as exc:
                log.debug('----------------------------------------', 'parser')
                log.debug(str(exc), 'parser')
                log.debug('----------------------------------------', 'parser')
                return False
        neighbor.rib.clear()

    return True


# ================================================================ check_message
#


def check_message(neighbor, message):
    message = message.replace(':', '')
    msghexa = [message[i * 2 : (i * 2) + 2] for i in range(len(message) // 2)]
    raw = bytes([int(_, 16) for _ in msghexa])

    if raw.startswith(b'\xff' * 16):
        kind = raw[18]
        # XXX: FIXME: check size
        # size = (raw[16] << 16) + raw[17]

        if kind == 1:
            return check_open(neighbor, raw[18:])
        elif kind == 2:
            return check_update(neighbor, raw)
        elif kind == 3:
            return check_notification(raw)
    else:
        return check_update(neighbor, raw)


# ================================================================= check_update
#


def check_open(neighbor, raw):
    pass


# ================================================================= check_update
#


def check_update(neighbor, raw):
    log._option['parser'] = True

    neighbor = neighbor[list(neighbor)[0]]

    path = {}
    for f in NLRI.known_families():
        if neighbor.add_path:
            path[f] = neighbor.add_path

    capa = Capabilities().new(neighbor, False)
    capa[Capability.CODE.ADD_PATH] = path
    capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
    # capa[Capability.CODE.FOUR_BYTES_ASN] = True

    routerid_1 = str(neighbor.router_id)
    routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))

    o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
    o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
    negotiated = Negotiated(neighbor)
    negotiated.sent(o1)
    negotiated.received(o2)
    # grouped = False

    while raw:
        if raw.startswith(b'\xff' * 16):
            kind = raw[18]
            size = (raw[16] << 16) + raw[17]

            injected, raw = raw[19:size], raw[size:]

            if kind == 2:
                log.debug('the message is an update', 'parser')
                decoding = 'update'
            else:
                log.debug('the message is not an update (%d) - aborting' % kind, 'parser')
                return False
        else:
            log.debug('header missing, assuming this message is ONE update', 'parser')
            decoding = 'update'
            injected, raw = raw, ''

        try:
            # This does not take the BGP header - let's assume we will not break that :)
            update = Update.unpack_message(injected, negotiated)
        except Notify:
            import traceback
            log.error('could not parse the message', 'parser')
            log.error(traceback.format_exc(), 'parser')
            return False
        except Exception:
            import traceback
            log.error('could not parse the message', 'parser')
            log.error(traceback.format_exc(), 'parser')
            return False

        log.debug('', 'parser')  # new line
        for number in range(len(update.nlris)):
            change = Change(update.nlris[number], update.attributes)
            log.info('decoded %s %s %s' % (decoding, change.nlri.action, change.extensive()), 'parser')
        log.info(
            'update json %s' % Response.JSON(json_version).update(neighbor, 'in', update, None, '', ''), 'parser'
        )

    return True


# ================================================================= check_update
#


def check_notification(raw):
    notification = Notification.unpack_message(raw[18:], None)
    # XXX: FIXME: should be using logger here
    print(notification)
    return True


1			# encoding: utf-8
2			"""
3			check.py
4
5			Created by Thomas Mangin on 2009-08-25.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			# common
11
12			import copy
13
14			from exabgp.bgp.message import Update
15			from exabgp.bgp.message import Open
16			from exabgp.bgp.message.open import Version
17			from exabgp.bgp.message.open import ASN
18			from exabgp.bgp.message.open import RouterID
19			from exabgp.bgp.message.open import HoldTime
20			from exabgp.bgp.message.open.capability import Capabilities
21			from exabgp.bgp.message.open.capability import Capability
22			from exabgp.bgp.message.open.capability import Negotiated
23			from exabgp.bgp.message import Notify
24			from exabgp.bgp.message.update.nlri import NLRI
25
26			from exabgp.logger import log
27
28			# check_neighbor
29
30			from exabgp.util.od import od
31			from exabgp.rib.change import Change
32
33			# check_update
34
35			from exabgp.reactor.api.response import Response
36
37			# check_notification
38
39			from exabgp.bgp.message import Notification
40
41			# JSON version
42
43			from exabgp.version import json as json_version
44
45
46			# =============================================================== check_neighbor
47			# ...
48
49
50			def check_generation(neighbors):
51			log._option['parser'] = True
52
53			for name in neighbors.keys():
54			neighbor = copy.deepcopy(neighbors[name])
55			neighbor.local_as = neighbor.peer_as
56
57			path = {}
58			for f in NLRI.known_families():
59			if neighbor.add_path:
60			path[f] = neighbor.add_path
61
62			capa = Capabilities().new(neighbor, False)
63			if path:
64			capa[Capability.CODE.ADD_PATH] = path
65			capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
66
67			routerid_1 = str(neighbor.router_id)
68			routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))
69
70			o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
71			o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
72			negotiated = Negotiated(neighbor)
73			negotiated.sent(o1)
74			negotiated.received(o2)
75			# grouped = False
76
77			for _ in neighbor.rib.outgoing.updates(False):
78			pass
79
80			for change1 in neighbor.rib.outgoing.cached_changes():
81			str1 = change1.extensive()
82			packed = list(Update([change1.nlri], change1.attributes).messages(negotiated))
83			pack1 = packed[0]
84
85			log.debug('parsed route requires %d updates' % len(packed), 'parser')
86			log.debug('update size is %d' % len(pack1), 'parser')
87
88			log.debug('parsed route %s' % str1, 'parser')
89			log.debug('parsed hex %s' % od(pack1), 'parser')
90
91			# This does not take the BGP header - let's assume we will not break that :)
92			try:
93			log.debug('') # new line
94
95			pack1s = pack1[19:] if pack1.startswith(b'\xFF' * 16) else pack1
96			update = Update.unpack_message(pack1s, negotiated)
97
98			change2 = Change(update.nlris[0], update.attributes)
99			str2 = change2.extensive()
100			pack2 = list(Update([update.nlris[0]], update.attributes).messages(negotiated))[0]
101
102			log.debug('recoded route %s' % str2, 'parser')
103			log.debug('recoded hex %s' % od(pack2), 'parser')
104
105			str1 = str1.replace('attribute [ 0x04 0x80 0x00000064 ]', 'med 100')
106			str1r = (
107			str1.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
108			)
109			str2r = (
110			str2.lower().replace(' med 100', '').replace(' local-preference 100', '').replace(' origin igp', '')
111			)
112			str2r = str2r.replace(
113			'large-community [ 1:2:3 10:11:12 ]',
114			'attribute [ 0x20 0xc0 0x0000000100000002000000030000000a0000000b0000000c ]',
115			)
116
117			if 'next-hop self' in str1r:
118			if ':' in str1r:
119			str1r = str1r.replace('next-hop self', 'next-hop ::1')
120			else:
121			str1r = str1r.replace('next-hop self', 'next-hop %s' % neighbor.local_address)
122
123			if ' name ' in str1r:
124			parts = str1r.split(' ')
125			pos = parts.index('name')
126			str1r = ' '.join(parts[:pos] + parts[pos + 2 :])
127
128			skip = False
129
130			if str1r != str2r:
131			if 'attribute [' in str1r and ' 0x00 ' in str1r:
132			# we do not decode non-transitive attributes
133			log.debug('skipping string check on update with non-transitive attribute(s)', 'parser')
134			skip = True
135			else:
136			log.debug('strings are different:', 'parser')
137			log.debug('[%s]' % (str1r), 'parser')
138			log.debug('[%s]' % (str2r), 'parser')
139			return False
140			else:
141			log.debug('strings are fine', 'parser')
142
143			if skip:
144			log.debug('skipping encoding for update with non-transitive attribute(s)', 'parser')
145			elif pack1 != pack2:
146			log.debug('encoding are different', 'parser')
147			log.debug('[%s]' % (od(pack1)), 'parser')
148			log.debug('[%s]' % (od(pack2)), 'parser')
149			return False
150			else:
151			log.debug('encoding is fine', 'parser')
152			log.debug('----------------------------------------', 'parser')
153
154			log.debug('JSON nlri %s' % change1.nlri.json(), 'parser')
155			log.debug('JSON attr %s' % change1.attributes.json(), 'parser')
156
157			except Notify as exc:
158			log.debug('----------------------------------------', 'parser')
159			log.debug(str(exc), 'parser')
160			log.debug('----------------------------------------', 'parser')
161			return False
162			neighbor.rib.clear()
163
164			return True
165
166
167			# ================================================================ check_message
168			#
169
170
171			def check_message(neighbor, message):
172			message = message.replace(':', '')
173			msghexa = [message[i * 2 : (i * 2) + 2] for i in range(len(message) // 2)]
174			raw = bytes([int(_, 16) for _ in msghexa])
175
176			if raw.startswith(b'\xff' * 16):
177			kind = raw[18]
178			# XXX: FIXME: check size
179			# size = (raw[16] << 16) + raw[17]
180
181			if kind == 1:
182			return check_open(neighbor, raw[18:])
183			elif kind == 2:
184			return check_update(neighbor, raw)
185			elif kind == 3:
186			return check_notification(raw)
187			else:
188			return check_update(neighbor, raw)
189
190
191			# ================================================================= check_update
192			#
193
194
195			def check_open(neighbor, raw):
196			pass
197
198
199			# ================================================================= check_update
200			#
201
202
203			def check_update(neighbor, raw):
204			log._option['parser'] = True
205
206			neighbor = neighbor[list(neighbor)[0]]
207
208			path = {}
209			for f in NLRI.known_families():
210			if neighbor.add_path:
211			path[f] = neighbor.add_path
212
213			capa = Capabilities().new(neighbor, False)
214			capa[Capability.CODE.ADD_PATH] = path
215			capa[Capability.CODE.MULTIPROTOCOL] = neighbor.families()
216			# capa[Capability.CODE.FOUR_BYTES_ASN] = True
217
218			routerid_1 = str(neighbor.router_id)
219			routerid_2 = '.'.join(str((int(_) + 1) % 250) for _ in str(neighbor.router_id).split('.', -1))
220
221			o1 = Open(Version(4), ASN(neighbor.local_as), HoldTime(180), RouterID(routerid_1), capa)
222			o2 = Open(Version(4), ASN(neighbor.peer_as), HoldTime(180), RouterID(routerid_2), capa)
223			negotiated = Negotiated(neighbor)
224			negotiated.sent(o1)
225			negotiated.received(o2)
226			# grouped = False
227
228			while raw:
229			if raw.startswith(b'\xff' * 16):
230			kind = raw[18]
231			size = (raw[16] << 16) + raw[17]
232
233			injected, raw = raw[19:size], raw[size:]
234
235			if kind == 2:
236			log.debug('the message is an update', 'parser')
237			decoding = 'update'
238			else:
239			log.debug('the message is not an update (%d) - aborting' % kind, 'parser')
240			return False
241			else:
242			log.debug('header missing, assuming this message is ONE update', 'parser')
243			decoding = 'update'
244			injected, raw = raw, ''
245
246			try:
247			# This does not take the BGP header - let's assume we will not break that :)
248			update = Update.unpack_message(injected, negotiated)
249			except Notify:
250			import traceback
251			log.error('could not parse the message', 'parser')
252			log.error(traceback.format_exc(), 'parser')
253			return False
254			except Exception:
255			import traceback
256			log.error('could not parse the message', 'parser')
257			log.error(traceback.format_exc(), 'parser')
258			return False
259
260			log.debug('', 'parser') # new line
261			for number in range(len(update.nlris)):
262			change = Change(update.nlris[number], update.attributes)
263			log.info('decoded %s %s %s' % (decoding, change.nlri.action, change.extensive()), 'parser')
264			log.info(
265			'update json %s' % Response.JSON(json_version).update(neighbor, 'in', update, None, '', ''), 'parser'
266			)
267
268			return True
269
270
271			# ================================================================= check_update
272			#
273
274
275			def check_notification(raw):
276			notification = Notification.unpack_message(raw[18:], None)
277			# XXX: FIXME: should be using logger here
278			print(notification)
279			return True
280

Exa-Networks / exabgp

Push — master ( ea37f3...9aacac )

exabgp.configuration.check.check_generation() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like