<s> from Queue import Queue <EOL> import random <EOL> import threading <EOL> import unittest <EOL> from coapclient import HelperClient <EOL> from coapserver import CoAPServer <EOL> from coapthon import defines <EOL> from coapthon . messages . message import Message <EOL> from coapthon . messages . option import Option <EOL> from coapthon . messages . request import Request <EOL> from coapthon . messages . response import Response <EOL> __author__ = '<STR_LIT>' <EOL> __version__ = "<STR_LIT>" <EOL> class Tests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . server_address = ( "<STR_LIT>" , <NUM_LIT> ) <EOL> self . current_mid = random . randint ( <NUM_LIT:1> , <NUM_LIT:1000> ) <EOL> self . server_mid = random . randint ( <NUM_LIT:1000> , <NUM_LIT> ) <EOL> self . server = CoAPServer ( "<STR_LIT>" , <NUM_LIT> , multicast = True ) <EOL> self . server_thread = threading . Thread ( target = self . server . listen , args = ( <NUM_LIT:10> , ) ) <EOL> self . server_thread . start ( ) <EOL> self . queue = Queue ( ) <EOL> def tearDown ( self ) : <EOL> self . server . close ( ) <EOL> self . server_thread . join ( timeout = <NUM_LIT> ) <EOL> self . server = None <EOL> def _test_with_client ( self , message_list ) : <EOL> client = HelperClient ( self . server_address ) <EOL> for message , expected in message_list : <EOL> if message is not None : <EOL> received_message = client . send_request ( message ) <EOL> if expected is not None : <EOL> if expected . type is not None : <EOL> self . assertEqual ( received_message . type , expected . type ) <EOL> if expected . mid is not None : <EOL> self . assertEqual ( received_message . mid , expected . mid ) <EOL> self . assertEqual ( received_message . code , expected . code ) <EOL> if expected . source is not None : <EOL> self . assertEqual ( received_message . source , self . server_address ) <EOL> if expected . token is not None : <EOL> self . assertEqual ( received_message . token , expected . token ) <EOL> if expected . payload is not None : <EOL> self . assertEqual ( received_message . payload , expected . payload ) <EOL> if expected . options : <EOL> self . assertEqual ( len ( received_message . options ) , len ( expected . options ) ) <EOL> for o in expected . options : <EOL> assert isinstance ( o , Option ) <EOL> option_value = getattr ( expected , o . name . lower ( ) . replace ( "<STR_LIT:->" , "<STR_LIT:_>" ) ) <EOL> option_value_rec = getattr ( received_message , o . name . lower ( ) . replace ( "<STR_LIT:->" , "<STR_LIT:_>" ) ) <EOL> self . assertEqual ( option_value , option_value_rec ) <EOL> client . stop ( ) <EOL> def _test_with_client_observe ( self , message_list ) : <EOL> client = HelperClient ( self . server_address ) <EOL> for message , expected in message_list : <EOL> if message is not None : <EOL> client . send_request ( message , self . client_callback ) <EOL> if expected is not None : <EOL> received_message = self . queue . get ( ) <EOL> if expected . type is not None : <EOL> self . assertEqual ( received_message . type , expected . type ) <EOL> if expected . mid is not None : <EOL> self . assertEqual ( received_message . mid , expected . mid ) <EOL> self . assertEqual ( received_message . code , expected . code ) <EOL> if expected . source is not None : <EOL> self . assertEqual ( received_message . source , self . server_address ) <EOL> if expected . token is not None : <EOL> self . assertEqual ( received_message . token , expected . token ) <EOL> if expected . payload is not None : <EOL> self . assertEqual ( received_message . payload , expected . payload ) <EOL> if expected . options : <EOL> self . assertEqual ( len ( received_message . options ) , len ( expected . options ) ) <EOL> for o in expected . options : <EOL> assert isinstance ( o , Option ) <EOL> option_value = getattr ( expected , o . name . lower ( ) . replace ( "<STR_LIT:->" , "<STR_LIT:_>" ) ) <EOL> option_value_rec = getattr ( received_message , o . name . lower ( ) . replace ( "<STR_LIT:->" , "<STR_LIT:_>" ) ) <EOL> self . assertEqual ( option_value , option_value_rec ) <EOL> client . stop ( ) <EOL> def client_callback ( self , response ) : <EOL> print "<STR_LIT>" <EOL> self . queue . put ( response ) <EOL> def test_not_allowed ( self ) : <EOL> print "<STR_LIT>" <EOL> path = "<STR_LIT>" <EOL> req = Request ( ) <EOL> req . code = defines . Codes . GET . number <EOL> req . uri_path = path <EOL> req . type = defines . Types [ "<STR_LIT>" ] <EOL> req . _mid = self . current_mid <EOL> req . destination = self . server_address <EOL> expected = Response ( ) <EOL> expected . type = defines . Types [ "<STR_LIT>" ] <EOL> expected . _mid = self . current_mid <EOL> expected . code = defines . Codes . METHOD_NOT_ALLOWED . number <EOL> expected . token = None <EOL> exchange1 = ( req , expected ) <EOL> self . current_mid += <NUM_LIT:1> <EOL> req = Request ( ) <EOL> req . code = defines . Codes . POST . number <EOL> req . uri_path = path <EOL> req . type = defines . Types [ "<STR_LIT>" ] <EOL> req . _mid = self . current_mid <EOL> req . destination = self . server_address <EOL> expected = Response ( ) <EOL> expected . type = defines . Types [ "<STR_LIT>" ] <EOL> expected . _mid = self . current_mid <EOL> expected . code = defines . Codes . METHOD_NOT_ALLOWED . number <EOL> expected . token = None <EOL> exchange2 = ( req , expected ) <EOL> self . current_mid += <NUM_LIT:1> <EOL> req = Request ( ) <EOL> req . code = defines . Codes . PUT . number <EOL> req . uri_path = path <EOL> req . type = defines . Types [ "<STR_LIT>" ] <EOL> req . _mid = self . current_mid <EOL> req . destination = self . server_address <EOL> expected = Response ( ) <EOL> expected . type = defines . Types [ "<STR_LIT>" ] <EOL> expected . _mid = self . current_mid <EOL> expected . code = defines . Codes . METHOD_NOT_ALLOWED . number <EOL> expected . token = None <EOL> exchange3 = ( req , expected ) <EOL> self . current_mid += <NUM_LIT:1> <EOL> req = Request ( ) <EOL> req . code = defines . Codes . DELETE . number <EOL> req . uri_path = path <EOL> req . type = defines . Types [ "<STR_LIT>" ] <EOL> req . _mid = self . current_mid <EOL> req . destination = self . server_address <EOL> expected = Response ( ) <EOL> expected . type = defines . Types [ "<STR_LIT>" ] <EOL> expected . _mid = self . current_mid <EOL> expected . code = defines . Codes . METHOD_NOT_ALLOWED . number <EOL> expected . token = None <EOL> exchange4 = ( req , expected ) <EOL> self . current_mid += <NUM_LIT:1> <EOL> self . _test_with_client ( [ exchange1 , exchange2 , exchange3 , exchange4 ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
