<s> import unittest <EOL> from nose . tools import eq_ <EOL> from nose . tools import ok_ <EOL> from time import time <EOL> from ryu . lib . packet import bmp <EOL> from ryu . lib . packet import bgp <EOL> from ryu . lib . packet import afi <EOL> from ryu . lib . packet import safi <EOL> class Test_bmp ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> pass <EOL> def tearDown ( self ) : <EOL> pass <EOL> def _time ( self ) : <EOL> return round ( time ( ) , <NUM_LIT:6> ) <EOL> def test_route_monitoring ( self ) : <EOL> update = bgp . BGPUpdate ( ) <EOL> msg = bmp . BMPRouteMonitoring ( bgp_update = update , <EOL> peer_type = bmp . BMP_PEER_TYPE_GLOBAL , <EOL> is_post_policy = True , <EOL> peer_distinguisher = <NUM_LIT:0> , <EOL> peer_address = '<STR_LIT>' , <EOL> peer_as = <NUM_LIT> , <EOL> peer_bgp_id = '<STR_LIT>' , <EOL> timestamp = self . _time ( ) ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( ) , msg2 . to_jsondict ( ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) <EOL> def test_statistics_report ( self ) : <EOL> stats = [ { '<STR_LIT:type>' : bmp . BMP_STAT_TYPE_REJECTED , '<STR_LIT:value>' : <NUM_LIT:100> } , <EOL> { '<STR_LIT:type>' : bmp . BMP_STAT_TYPE_DUPLICATE_PREFIX , '<STR_LIT:value>' : <NUM_LIT:200> } , <EOL> { '<STR_LIT:type>' : bmp . BMP_STAT_TYPE_DUPLICATE_WITHDRAW , '<STR_LIT:value>' : <NUM_LIT> } , <EOL> { '<STR_LIT:type>' : bmp . BMP_STAT_TYPE_ADJ_RIB_IN , '<STR_LIT:value>' : <NUM_LIT> } , <EOL> { '<STR_LIT:type>' : bmp . BMP_STAT_TYPE_LOC_RIB , '<STR_LIT:value>' : <NUM_LIT> } ] <EOL> msg = bmp . BMPStatisticsReport ( stats = stats , <EOL> peer_type = bmp . BMP_PEER_TYPE_GLOBAL , <EOL> is_post_policy = True , <EOL> peer_distinguisher = <NUM_LIT:0> , <EOL> peer_address = '<STR_LIT>' , <EOL> peer_as = <NUM_LIT> , <EOL> peer_bgp_id = '<STR_LIT>' , <EOL> timestamp = self . _time ( ) ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( ) , msg2 . to_jsondict ( ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) <EOL> def test_peer_down_notification ( self ) : <EOL> reason = bmp . BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION <EOL> data = b'<STR_LIT>' <EOL> data = bgp . BGPNotification ( error_code = <NUM_LIT:1> , error_subcode = <NUM_LIT:2> , data = data ) <EOL> msg = bmp . BMPPeerDownNotification ( reason = reason , data = data , <EOL> peer_type = bmp . BMP_PEER_TYPE_GLOBAL , <EOL> is_post_policy = True , <EOL> peer_distinguisher = <NUM_LIT:0> , <EOL> peer_address = '<STR_LIT>' , <EOL> peer_as = <NUM_LIT> , <EOL> peer_bgp_id = '<STR_LIT>' , <EOL> timestamp = self . _time ( ) ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( ) , msg2 . to_jsondict ( ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) <EOL> def test_peer_up_notification ( self ) : <EOL> opt_param = [ bgp . BGPOptParamCapabilityUnknown ( cap_code = <NUM_LIT:200> , <EOL> cap_value = b'<STR_LIT>' ) , <EOL> bgp . BGPOptParamCapabilityRouteRefresh ( ) , <EOL> bgp . BGPOptParamCapabilityMultiprotocol ( <EOL> afi = afi . IP , safi = safi . MPLS_VPN ) ] <EOL> open_message = bgp . BGPOpen ( my_as = <NUM_LIT> , bgp_identifier = '<STR_LIT>' , <EOL> opt_param = opt_param ) <EOL> msg = bmp . BMPPeerUpNotification ( local_address = '<STR_LIT>' , <EOL> local_port = <NUM_LIT> , <EOL> remote_port = <NUM_LIT> , <EOL> sent_open_message = open_message , <EOL> received_open_message = open_message , <EOL> peer_type = bmp . BMP_PEER_TYPE_GLOBAL , <EOL> is_post_policy = True , <EOL> peer_distinguisher = <NUM_LIT:0> , <EOL> peer_address = '<STR_LIT>' , <EOL> peer_as = <NUM_LIT> , <EOL> peer_bgp_id = '<STR_LIT>' , <EOL> timestamp = self . _time ( ) ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( ) , msg2 . to_jsondict ( ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) <EOL> def test_initiation ( self ) : <EOL> initiation_info = [ { '<STR_LIT:type>' : bmp . BMP_INIT_TYPE_STRING , <EOL> '<STR_LIT:value>' : u'<STR_LIT>' } ] <EOL> msg = bmp . BMPInitiation ( info = initiation_info ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( lambda v : v ) , msg2 . to_jsondict ( lambda v : v ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) <EOL> def test_termination ( self ) : <EOL> termination_info = [ { '<STR_LIT:type>' : bmp . BMP_TERM_TYPE_STRING , <EOL> '<STR_LIT:value>' : u'<STR_LIT>' } , <EOL> { '<STR_LIT:type>' : bmp . BMP_TERM_TYPE_REASON , <EOL> '<STR_LIT:value>' : bmp . BMP_TERM_REASON_ADMIN } ] <EOL> msg = bmp . BMPTermination ( info = termination_info ) <EOL> binmsg = msg . serialize ( ) <EOL> msg2 , rest = bmp . BMPMessage . parser ( binmsg ) <EOL> eq_ ( msg . to_jsondict ( lambda v : v ) , msg2 . to_jsondict ( lambda v : v ) ) <EOL> eq_ ( rest , b'<STR_LIT>' ) </s>
