1 # Copyright (C) 2014 Nippon Telegraph and Telephone Corporation.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
17 from nose.tools import eq_
18 from nose.tools import ok_
21 from ryu.lib.packet import bmp
22 from ryu.lib.packet import bgp
23 from ryu.lib.packet import afi
24 from ryu.lib.packet import safi
27 class Test_bmp(unittest.TestCase):
28 """ Test case for ryu.lib.packet.bmp
38 # time() can give sub-microsecond precision, which results
39 # in an assertion failure
40 return round(time(), 6)
42 def test_route_monitoring(self):
43 update = bgp.BGPUpdate()
44 msg = bmp.BMPRouteMonitoring(bgp_update=update,
45 peer_type=bmp.BMP_PEER_TYPE_GLOBAL,
48 peer_address='192.0.2.1',
50 peer_bgp_id='192.0.2.1',
51 timestamp=self._time())
52 binmsg = msg.serialize()
53 msg2, rest = bmp.BMPMessage.parser(binmsg)
54 eq_(msg.to_jsondict(), msg2.to_jsondict())
57 def test_statistics_report(self):
58 stats = [{'type': bmp.BMP_STAT_TYPE_REJECTED, 'value': 100},
59 {'type': bmp.BMP_STAT_TYPE_DUPLICATE_PREFIX, 'value': 200},
60 {'type': bmp.BMP_STAT_TYPE_DUPLICATE_WITHDRAW, 'value': 300},
61 {'type': bmp.BMP_STAT_TYPE_ADJ_RIB_IN, 'value': 100000},
62 {'type': bmp.BMP_STAT_TYPE_LOC_RIB, 'value': 500000}]
63 msg = bmp.BMPStatisticsReport(stats=stats,
64 peer_type=bmp.BMP_PEER_TYPE_GLOBAL,
67 peer_address='192.0.2.1',
69 peer_bgp_id='192.0.2.1',
70 timestamp=self._time())
71 binmsg = msg.serialize()
72 msg2, rest = bmp.BMPMessage.parser(binmsg)
73 eq_(msg.to_jsondict(), msg2.to_jsondict())
76 def test_peer_down_notification(self):
77 reason = bmp.BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION
79 data = bgp.BGPNotification(error_code=1, error_subcode=2, data=data)
80 msg = bmp.BMPPeerDownNotification(reason=reason, data=data,
81 peer_type=bmp.BMP_PEER_TYPE_GLOBAL,
84 peer_address='192.0.2.1',
86 peer_bgp_id='192.0.2.1',
87 timestamp=self._time())
88 binmsg = msg.serialize()
89 msg2, rest = bmp.BMPMessage.parser(binmsg)
90 eq_(msg.to_jsondict(), msg2.to_jsondict())
93 def test_peer_up_notification(self):
94 opt_param = [bgp.BGPOptParamCapabilityUnknown(cap_code=200,
96 bgp.BGPOptParamCapabilityRouteRefresh(),
97 bgp.BGPOptParamCapabilityMultiprotocol(
98 afi=afi.IP, safi=safi.MPLS_VPN)]
99 open_message = bgp.BGPOpen(my_as=40000, bgp_identifier='192.0.2.2',
101 msg = bmp.BMPPeerUpNotification(local_address='192.0.2.2',
104 sent_open_message=open_message,
105 received_open_message=open_message,
106 peer_type=bmp.BMP_PEER_TYPE_GLOBAL,
108 peer_distinguisher=0,
109 peer_address='192.0.2.1',
111 peer_bgp_id='192.0.2.1',
112 timestamp=self._time())
113 binmsg = msg.serialize()
114 msg2, rest = bmp.BMPMessage.parser(binmsg)
115 eq_(msg.to_jsondict(), msg2.to_jsondict())
118 def test_initiation(self):
119 initiation_info = [{'type': bmp.BMP_INIT_TYPE_STRING,
120 'value': u'This is Ryu BGP BMP message'}]
121 msg = bmp.BMPInitiation(info=initiation_info)
122 binmsg = msg.serialize()
123 msg2, rest = bmp.BMPMessage.parser(binmsg)
124 eq_(msg.to_jsondict(lambda v: v), msg2.to_jsondict(lambda v: v))
127 def test_termination(self):
128 termination_info = [{'type': bmp.BMP_TERM_TYPE_STRING,
129 'value': u'Session administatively closed'},
130 {'type': bmp.BMP_TERM_TYPE_REASON,
131 'value': bmp.BMP_TERM_REASON_ADMIN}]
132 msg = bmp.BMPTermination(info=termination_info)
133 binmsg = msg.serialize()
134 msg2, rest = bmp.BMPMessage.parser(binmsg)
135 eq_(msg.to_jsondict(lambda v: v), msg2.to_jsondict(lambda v: v))