#NEWLINE# def test_recv_token( self ) : #NEWLINE# #INDENT# expected_reply = { #NEWLINE# #INDENT# 'cmd' : 'TOKEN_REPLY' , #NEWLINE# 'port_id' : self . port . id , #NEWLINE# 'peer_port_id' : self . peer_port . id , #NEWLINE# 'sequencenbr' : 0 , #NEWLINE# 'value' : 'ACK' , #NEWLINE# #UNINDENT# } #NEWLINE# payload = { #NEWLINE# #INDENT# 'port_id' : self . port . id , #NEWLINE# 'peer_port_id' : self . peer_port . id , #NEWLINE# 'sequencenbr' : 0 , #NEWLINE# 'token' : { #NEWLINE# #INDENT# 'type' : 'Token' , #NEWLINE# 'data' : 5 , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } #NEWLINE# self . tunnel_in . recv_token ( payload ) #NEWLINE# assert self . trigger_loop . called #NEWLINE# assert ( self . port . fifo . fifo [ 0 ] . value == 5 ) #NEWLINE# self . tunnel . send . assert_called_with ( expected_reply ) #NEWLINE# self . trigger_loop . reset_mock ( ) #NEWLINE# self . tunnel . send . reset_mock ( ) #NEWLINE# payload [ 'sequencenbr' ] = 100 #NEWLINE# self . tunnel_in . recv_token ( payload ) #NEWLINE# assert ( not self . trigger_loop . called ) #NEWLINE# expected_reply [ 'sequencenbr' ] = 100 #NEWLINE# expected_reply [ 'value' ] = 'NACK' #NEWLINE# self . tunnel . send . assert_called_with ( expected_reply ) #NEWLINE# self . trigger_loop . reset_mock ( ) #NEWLINE# self . tunnel . send . reset_mock ( ) #NEWLINE# payload [ 'sequencenbr' ] = 0 #NEWLINE# self . tunnel_in . recv_token ( payload ) #NEWLINE# assert ( not self . trigger_loop . called ) #NEWLINE# expected_reply [ 'sequencenbr' ] = 0 #NEWLINE# expected_reply [ 'value' ] = 'ACK' #NEWLINE# self . tunnel . send . assert_called_with ( expected_reply )