1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
#!/usr/bin/env python
# encoding: utf-8
"""
protocol.py
Created by Thomas Mangin on 2009-08-27.
Copyright (c) 2009-2015 Exa Networks. All rights reserved.
License: 3-clause BSD. (See the COPYRIGHT file)
"""
import unittest
from exabgp.bgp.message.open.asn import ASN
from exabgp.bgp.message import Open
from exabgp.bgp.message.open import Capabilities
# from exabgp.bgp.message import Notification
from exabgp.bgp.message import KeepAlive
from exabgp.bgp.message import Update
from exabgp.bgp.message.update import Attributes
from exabgp.reactor.protocol import Protocol
from exabgp.bgp.neighbor import Neighbor
from exabgp.vendoring import six
from exabgp.vendoring.six.moves import StringIO
from exabgp.configuration.environment import environment
environment.setup('')
class Network(StringIO):
def pending(self):
return True
# route1 = Update([],[to_NLRI('10.0.0.1','32')],Attributes())
# route1.next_hop = '10.0.0.254'
#
#
# route2 = Update([],[to_NLRI('10.0.1.1','32')],Attributes())
# route2.next_hop = '10.0.0.254'
#
#
# route3 = Update([],[to_NLRI('10.0.2.1','32')],Attributes())
# route3.next_hop = '10.0.0.254'
#
#
# routes = [route1,route2,route3]
# routes.sort()
class TestProtocol(unittest.TestCase):
def setUp(self):
self.neighbor = Neighbor()
self.neighbor.local_as = ASN(65000)
self.neighbor.peer_as = ASN(65000)
# self.neighbor.peer_address = InetIP('1.2.3.4')
# self.neighbor.local_address = InetIP('5.6.7.8')
def test_1_selfparse_open(self):
# ds = Open(4,65000,'1.2.3.4',Capabilities().default(),30)
#
# txt = ds.message()
# network = Network(txt)
# print [hex(ord(c)) for c in txt]
# bgp = Protocol(self.neighbor,network)
# bgp.follow = False
# o = bgp.read_open('127.0.0.1')
# self.assertEqual(o.version,4)
# self.assertEqual(o.asn,65000)
# self.assertEqual(o.hold_time,30)
# self.assertEqual(str(o.router_id),'1.2.3.4')
pass
def test_2_selfparse_KeepAlive(self):
# ds = KeepAlive()
#
# txt = ds.message()
# network = Network(txt)
# bgp = Protocol(self.neighbor,network)
# message = bgp.read_message()
# self.assertEqual(message.TYPE,KeepAlive.TYPE)
pass
def test_3_parse_update(self):
# txt = ''.join([chr(c) for c in [0x0, 0x0, 0x0, 0x1c, 0x40, 0x1, 0x1, 0x2, 0x40, 0x2, 0x0, 0x40, 0x3, 0x4, 0xc0, 0x0, 0x2, 0xfe, 0x80, 0x4, 0x4, 0x0, 0x0, 0x0, 0x0, 0x40, 0x5, 0x4, 0x0, 0x0, 0x1, 0x23, 0x20, 0x52, 0xdb, 0x0, 0x7, 0x20, 0x52, 0xdb, 0x0, 0x45, 0x20, 0x52, 0xdb, 0x0, 0x47]])
# updates = new_Update(txt)
#
# routes = [str(route) for route in updates.added()]
# routes.sort()
# self.assertEqual(routes[0],'82.219.0.69/32 next-hop 192.0.2.254')
# self.assertEqual(routes[1],'82.219.0.7/32 next-hop 192.0.2.254')
# self.assertEqual(routes[2],'82.219.0.71/32 next-hop 192.0.2.254')
pass
def test_4_parse_update(self):
# txt = ''.join([chr(c) for c in [0x0, 0x0, 0x0, 0x12, 0x40, 0x1, 0x1, 0x0, 0x40, 0x2, 0x4, 0x2, 0x1, 0x78, 0x14, 0x40, 0x3, 0x4, 0x52, 0xdb, 0x2, 0xb5, 0x0]])
# updates = new_Update(txt)
# self.assertEqual(str(updates.added()[0]),'0.0.0.0/0 next-hop 82.219.2.181')
pass
def test_6_holdtime(self):
# class MyPeer(Network):
# _data = StringIO(Open(4,65000,'1.2.3.4',Capabilities().default(),90).message())
# def read (self, l):
# return self._data.read(l)
#
# network = MyPeer('')
#
# bgp = Protocol(self.neighbor,network)
# bgp.follow = False
#
# before = bgp.neighbor.hold_time
# bgp.new_open()
# bgp.read_open('127.0.0.1')
# after = bgp.neighbor.hold_time
#
# self.assertEqual(after,min(before,90))
pass
def test_7_message(self):
# txt = ''.join([chr(_) for _ in [0x0, 0x0, 0x0, 0x30, 0x40, 0x1, 0x1, 0x0, 0x50, 0x2, 0x0, 0x4, 0x2, 0x1, 0xff, 0xfe, 0x80, 0x4, 0x4, 0x0, 0x0, 0x0, 0x0, 0x80, 0xe, 0x1a, 0x0, 0x2, 0x1, 0x10, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x20, 0x12, 0x34, 0x56, 0x78]])
# updates = new_Update(txt)
# print updates
# self.assertEqual(str(updates.added()[0]),'1234:5678::/32 next-hop ::')
pass
def test_7_ipv6(self):
# txt = ''.join([chr(_) for _ in [0x0, 0x0, 0x0, 0x25, 0x40, 0x1, 0x1, 0x0, 0x40, 0x2, 0x4, 0x2, 0x1, 0xfd, 0xe8, 0xc0, 0x8, 0x8, 0x78, 0x14, 0x0, 0x0, 0x78, 0x14, 0x78, 0x14, 0x40, 0xf, 0xc, 0x0, 0x2, 0x1, 0x40, 0x2a, 0x2, 0xb, 0x80, 0x0, 0x0, 0x0, 0x1]])
# updates = new_Update(txt)
pass
if __name__ == '__main__':
unittest.main()
|
