Skip to content

Commit

Permalink
ruff'ing the test folder
Browse files Browse the repository at this point in the history
  • Loading branch information
@thomas-mangin
thomas-mangin committed Oct 25, 2024
1 parent 9dadff8 commit bf7f1b9
Show file tree
Hide file tree
Showing 10 changed files with 222 additions and 169 deletions.
24 changes: 18 additions & 6 deletions tests/bgpls_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,37 +8,49 @@


class TestTlvs(unittest.TestCase):
def test_ip_reach_ipv4(self,):
def test_ip_reach_ipv4(
self,
):
data = b'\n\n\x00'

tlv = IpReach.unpack(data, 3)
self.assertEqual(tlv.json(), '"ip-reachability-tlv": "10.0.0.0", "ip-reach-prefix": "10.0.0.0/10"')

def test_ip_reach_ipv6(self,):
def test_ip_reach_ipv6(
self,
):
data = b'\x7f \x01\x07\x00\x00\x00\x80'
tlv = IpReach.unpack(data, 4)
self.assertEqual(
tlv.json(), '"ip-reachability-tlv": "2001:700:0:8000::", "ip-reach-prefix": "2001:700:0:8000::/127"'
)

def test_igp_tags(self,):
def test_igp_tags(
self,
):
data = b'\x00\x00\xff\xfe'
tlv = IgpTags.unpack(data)
self.assertEqual(tlv.json(), '"igp-route-tags": [65534]')

def test_prefix_metric(self,):
def test_prefix_metric(
self,
):
data = b'\x00\x00\x00\x14'
tlv = PrefixMetric.unpack(data)
self.assertEqual(tlv.json(), '"prefix-metric": 20')

def test_ospf_route_type(self,):
def test_ospf_route_type(
self,
):
data = b'\x04'
tlv = OspfRoute.unpack(data)
self.assertEqual(tlv.json(), '"ospf-route-type": 4')


class TestDescriptors(unittest.TestCase):
def test_node_descriptor(self,):
def test_node_descriptor(
self,
):
data = b'\x02\x00\x00\x04\x00\x00\xff\xfd\x02\x01\x00\x04\x00\x00\x00\x00\x02\x03\x00\x04\nq?\xf0'
igp_type = 3
descriptor, remain = NodeDescriptor.unpack(data, igp_type)
Expand Down
6 changes: 3 additions & 3 deletions tests/cache_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ def __init__(self, data):
time1 = end - start

print(COUNT, 'iterations of', klass.__name__, 'with', UNIQUE, 'uniques classes')
print("time instance %d" % time1)
print('time instance %d' % time1)

cache = Cache()
start = time.time()
Expand All @@ -122,6 +122,6 @@ def __init__(self, data):
end = time.time()
time2 = end - start

print("time cached %d" % time2)
print("speedup %.3f" % (time1 / time2))
print('time cached %d' % time2)
print('speedup %.3f' % (time1 / time2))
print()
4 changes: 2 additions & 2 deletions tests/connection.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,10 @@ def test():
OPEN = ''.join(
[
chr(int(_, 16))
for _ in "FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF 00 1D 01 04 78 14 00 5A 52 DB 00 45 00".split()
for _ in 'FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF 00 1D 01 04 78 14 00 5A 52 DB 00 45 00'.split()
]
)
KEEP = ''.join([chr(int(_, 16)) for _ in "FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF 00 00 04".split()])
KEEP = ''.join([chr(int(_, 16)) for _ in 'FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF FF 00 00 04'.split()])

from exabgp.reactor.network.outgoing import Outgoing

Expand Down
73 changes: 41 additions & 32 deletions tests/decode_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,15 +11,12 @@
# pylint: ignore=E131
# pylama:ignore=E131

import sys
import unittest

# INET include is required as it perform some AFI/SAFI registration
from exabgp.bgp.message.update.nlri import INET # noqa

from exabgp.bgp.message.open.routerid import RouterID
from exabgp.protocol.ip import IPv4
from exabgp.bgp.message.open.asn import ASN
from exabgp.bgp.message.open.holdtime import HoldTime

from exabgp.bgp.message import Update
from exabgp.bgp.message import Open
Expand Down Expand Up @@ -306,37 +303,37 @@

class FakeNeighbor(dict):
def __init__(self):
self.update({
'description': 'a test neighbor',
'router-id': RouterID('127.0.0.1'),
'local-address': IPv4('127.0.0.1'),
'peer-address': IPv4('127.0.0.1'),
'host-name': 'localhost',
'domain-name': 'localdomain',
'peer-as': ASN('65500'),
'local-as': ASN('65500'),
'hold-time': HoldTime(180),
'capability': {
'asn4': False,
'add-path': 0,
'extended_message': False,
'nexthop': None,
'route-refresh': False,
'graceful-restart': False,
'multi-session': None,
'add-path': None,
'aigp': None,
'operational': None,
'extended-message': True,
'software-version': False,
},
})
self.update(
{
'description': 'a test neighbor',
'router-id': RouterID('127.0.0.1'),
'local-address': IPv4('127.0.0.1'),
'peer-address': IPv4('127.0.0.1'),
'host-name': 'localhost',
'domain-name': 'localdomain',
'peer-as': ASN('65500'),
'local-as': ASN('65500'),
'hold-time': HoldTime(180),
'capability': {
'asn4': False,
'add-path': 0,
'extended_message': False,
'nexthop': None,
'route-refresh': False,
'graceful-restart': False,
'multi-session': None,
'aigp': None,
'operational': None,
'extended-message': True,
'software-version': False,
},
}
)

@staticmethod
def families():
return NLRI.known_families()

import unittest

# from contextlib import contextmanager
# @contextmanager
Expand Down Expand Up @@ -371,8 +368,20 @@ def setUp(self):
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)
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)
Expand Down
22 changes: 12 additions & 10 deletions tests/flow_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,8 +28,8 @@ def setUp(self):

def test_rule(self):
components = {
'destination': Flow4Destination(IPv4.pton("192.0.2.0"), 24),
'source': Flow4Source(IPv4.pton("10.1.2.0"), 24),
'destination': Flow4Destination(IPv4.pton('192.0.2.0'), 24),
'source': Flow4Source(IPv4.pton('10.1.2.0'), 24),
'anyport_1': FlowAnyPort(NumericOperator.EQ, 25),
}
messages = {
Expand All @@ -41,13 +41,15 @@ def test_rule(self):
for key in ['destination', 'source', 'anyport_1']:
component = components[key].pack()
message = bytes(messages[key])
_ = component
_ = message
# if component != message:
# self.fail('content mismatch\n%s\n%s' % (['0x%02X' % _ for _ in component],['0x%02X' % _ for _ in message]))

def test_rule_and(self):
components = {
'destination': Flow4Destination(IPv4.pton("192.0.2.0"), 24),
'source': Flow4Source(IPv4.pton("10.1.2.0"), 24),
'destination': Flow4Destination(IPv4.pton('192.0.2.0'), 24),
'source': Flow4Source(IPv4.pton('10.1.2.0'), 24),
'anyport_1': FlowAnyPort(NumericOperator.EQ | NumericOperator.GT, 25),
'anyport_2': FlowAnyPort(NumericOperator.EQ | NumericOperator.LT, 80),
}
Expand All @@ -59,7 +61,7 @@ def test_rule_and(self):
}

flow = Flow()
message = b""
message = b''
for key in ['destination', 'source', 'anyport_1', 'anyport_2']:
flow.add(components[key])
message += bytes(messages[key])
Expand All @@ -70,8 +72,8 @@ def test_rule_and(self):

def test_nlri(self):
components = {
'destination': Flow4Destination(IPv4.pton("192.0.2.0"), 24),
'source': Flow4Source(IPv4.pton("10.1.2.0"), 24),
'destination': Flow4Destination(IPv4.pton('192.0.2.0'), 24),
'source': Flow4Source(IPv4.pton('10.1.2.0'), 24),
'anyport_1': FlowAnyPort(NumericOperator.EQ | NumericOperator.GT, 25),
'anyport_2': FlowAnyPort(NumericOperator.EQ | NumericOperator.LT, 80),
}
Expand All @@ -83,7 +85,7 @@ def test_nlri(self):
}

flow = Flow()
message = b""
message = b''
for key in ['destination', 'source', 'anyport_1', 'anyport_2']:
flow.add(components[key])
message += bytes(messages[key])
Expand All @@ -99,8 +101,8 @@ def test_nlri(self):

def test_compare(self):
components = {
'destination': Flow4Destination(IPv4.pton("192.0.2.0"), 24),
'source': Flow4Source(IPv4.pton("10.1.2.0"), 24),
'destination': Flow4Destination(IPv4.pton('192.0.2.0'), 24),
'source': Flow4Source(IPv4.pton('10.1.2.0'), 24),
'anyport_1': FlowAnyPort(NumericOperator.EQ | NumericOperator.GT, 25),
'anyport_2': FlowAnyPort(NumericOperator.EQ | NumericOperator.LT, 80),
'anyport_3': FlowAnyPort(NumericOperator.EQ, 80),
Expand Down
6 changes: 3 additions & 3 deletions tests/l2vpn_test.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -34,15 +34,15 @@ def setUp(self):
l2vpn:endpoint:3:base:262145:offset:1:size:8: route-distinguisher 172.30.5.4:13
l2vpn:endpoint:3:base:262145:offset:1:size:8: route-distinguisher 172.30.5.3:11
"""
self.encoded_l2vpn_nlri1 = bytearray.fromhex(u'0011 0001 AC1E 0504 000D 0003 0001 0008 4000 11')
self.encoded_l2vpn_nlri2 = bytearray.fromhex(u'0011 0001 AC1E 0503 000B 0003 0001 0008 4000 11')
self.encoded_l2vpn_nlri1 = bytearray.fromhex('0011 0001 AC1E 0504 000D 0003 0001 0008 4000 11')
self.encoded_l2vpn_nlri2 = bytearray.fromhex('0011 0001 AC1E 0503 000B 0003 0001 0008 4000 11')
self.decoded_l2vpn_nlri1 = VPLS(TestL2VPN.generate_rd('172.30.5.4:13'), 3, 262145, 1, 8)
self.decoded_l2vpn_nlri2 = VPLS(TestL2VPN.generate_rd('172.30.5.3:11'), 3, 262145, 1, 8)
"""
output from Juniper
Communities: target:54591:6 Layer2-info: encaps: VPLS, control flags:[0x0] , mtu: 0, site preference: 100
"""
self.encoded_ext_community = bytearray.fromhex(u'0002 D53F 0000 0006 800A 1300 0000 0064')
self.encoded_ext_community = bytearray.fromhex('0002 D53F 0000 0006 800A 1300 0000 0064')

def test_l2vpn_decode(self):
"""
Expand Down
Loading

0 comments on commit bf7f1b9

Please sign in to comment.