ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/packet/test_bmp.py

   1 # Copyright (C) 2014 Nippon Telegraph and Telephone Corporation.
   2 #
   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
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   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
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import unittest
  17 from nose.tools import eq_
  18 from nose.tools import ok_
  19 from time import time
  20
  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
  25
  26
  27 class Test_bmp(unittest.TestCase):
  28     """ Test case for ryu.lib.packet.bmp
  29     """
  30
  31     def setUp(self):
  32         pass
  33
  34     def tearDown(self):
  35         pass
  36
  37     def _time(self):
  38         # time() can give sub-microsecond precision, which results
  39         # in an assertion failure
  40         return round(time(), 6)
  41
  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,
  46                                      is_post_policy=True,
  47                                      peer_distinguisher=0,
  48                                      peer_address='192.0.2.1',
  49                                      peer_as=30000,
  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())
  55         eq_(rest, b'')
  56
  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,
  65                                       is_post_policy=True,
  66                                       peer_distinguisher=0,
  67                                       peer_address='192.0.2.1',
  68                                       peer_as=30000,
  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())
  74         eq_(rest, b'')
  75
  76     def test_peer_down_notification(self):
  77         reason = bmp.BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION
  78         data = b'hoge'
  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,
  82                                           is_post_policy=True,
  83                                           peer_distinguisher=0,
  84                                           peer_address='192.0.2.1',
  85                                           peer_as=30000,
  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())
  91         eq_(rest, b'')
  92
  93     def test_peer_up_notification(self):
  94         opt_param = [bgp.BGPOptParamCapabilityUnknown(cap_code=200,
  95                                                       cap_value=b'hoge'),
  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',
 100                                    opt_param=opt_param)
 101         msg = bmp.BMPPeerUpNotification(local_address='192.0.2.2',
 102                                         local_port=179,
 103                                         remote_port=11089,
 104                                         sent_open_message=open_message,
 105                                         received_open_message=open_message,
 106                                         peer_type=bmp.BMP_PEER_TYPE_GLOBAL,
 107                                         is_post_policy=True,
 108                                         peer_distinguisher=0,
 109                                         peer_address='192.0.2.1',
 110                                         peer_as=30000,
 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())
 116         eq_(rest, b'')
 117
 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))
 125         eq_(rest, b'')
 126
 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))
 136         eq_(rest, b'')