<s> import os <EOL> import sys <EOL> import socket <EOL> import unittest <EOL> import pyev <EOL> from whizzer import server , client , protocol <EOL> from whizzer . rpc import dispatch , proxy , msgpackrpc <EOL> from common import loop <EOL> class TestDispatch ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . count = <NUM_LIT:0> <EOL> def tearDown ( self ) : <EOL> self . count = <NUM_LIT:0> <EOL> def func ( self ) : <EOL> self . count += <NUM_LIT:1> <EOL> def add ( self , a , b ) : <EOL> self . count += <NUM_LIT:1> <EOL> return a + b <EOL> def test_add_noname ( self ) : <EOL> a = dispatch . Dispatch ( ) <EOL> a . add ( self . func ) <EOL> self . assertEqual ( len ( a . functions ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( self . func in a . functions . values ( ) ) <EOL> self . assertTrue ( self . func . __name__ in a . functions . keys ( ) ) <EOL> def test_add_with_name ( self ) : <EOL> a = dispatch . Dispatch ( ) <EOL> a . add ( self . func , "<STR_LIT>" ) <EOL> self . assertEqual ( len ( a . functions ) , <NUM_LIT:1> ) <EOL> self . assertTrue ( self . func in a . functions . values ( ) ) <EOL> self . assertTrue ( "<STR_LIT>" in a . functions . keys ( ) ) <EOL> def test_call_noargs ( self ) : <EOL> a = dispatch . Dispatch ( ) <EOL> a . add ( self . func ) <EOL> a . call ( self . func . __name__ , ( ) ) <EOL> self . assertEqual ( self . count , <NUM_LIT:1> ) <EOL> def test_call_args ( self ) : <EOL> a = dispatch . Dispatch ( ) <EOL> a . add ( self . add ) <EOL> self . assertEqual ( a . call ( self . add . __name__ , ( <NUM_LIT:1> , <NUM_LIT:2> ) ) , <NUM_LIT:3> ) <EOL> self . assertEqual ( self . count , <NUM_LIT:1> ) <EOL> class MockService ( object ) : <EOL> def __init__ ( self ) : <EOL> self . last_called = None <EOL> @ dispatch . remote <EOL> def add ( self , a , b ) : <EOL> self . last_called = self . add <EOL> return a + b <EOL> @ dispatch . remote <EOL> def exception ( self ) : <EOL> self . last_called = self . exception <EOL> raise Exception ( ) <EOL> @ dispatch . remote <EOL> def rpc_error ( self ) : <EOL> self . last_called = self . rpc_error <EOL> raise Exception ( ) <EOL> @ dispatch . remote <EOL> def tuple_ret ( self , a , b , c ) : <EOL> self . last_called = self . tuple_ret <EOL> return ( a , b , c ) <EOL> @ dispatch . remote <EOL> def dict_ret ( self , a , b , c ) : <EOL> self . last_called = self . dict_ret <EOL> return { '<STR_LIT:a>' : a , '<STR_LIT:b>' : b , '<STR_LIT:c>' : c } <EOL> @ dispatch . remote <EOL> def list_ret ( self , a , b , c ) : <EOL> self . last_called = self . list_ret <EOL> return [ a , b , c ] <EOL> @ dispatch . remote <EOL> def not_set_future_ret ( self ) : <EOL> self . last_called = self . not_set_future_ret <EOL> return future . Future ( ) <EOL> @ dispatch . remote <EOL> def set_future_ret ( self ) : <EOL> self . last_called = self . set_future_ret <EOL> f = future . Future ( ) <EOL> f . set_result ( True ) <EOL> return f <EOL> def local_only ( self ) : <EOL> self . last_called = self . local_only <EOL> pass <EOL> class TestMsgPackProtocol ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . factory = msgpackrpc . MsgPackProtocolFactory ( dispatch . ObjectDispatch ( MockService ( ) ) ) <EOL> self . protocol = self . factory . build ( loop ) <EOL> def tearDown ( self ) : <EOL> self . factory = None <EOL> self . protocol = None <EOL> def runTest ( self ) : <EOL> unittest . TestCase . runTest ( self ) <EOL> def mock_send_response ( self , msgid , error , result ) : <EOL> """<STR_LIT>""" <EOL> self . response = ( msgid , error , result ) <EOL> print "<STR_LIT>" + str ( self . response ) <EOL> def test_connection_made ( self ) : <EOL> future_proxy = self . protocol . proxy ( ) <EOL> self . protocol . connection_made ( None ) <EOL> self . assertTrue ( isinstance ( future_proxy . result ( ) , proxy . Proxy ) ) <EOL> def test_handle_request ( self ) : <EOL> self . protocol . send_response = self . mock_send_response <EOL> self . protocol . request ( <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" , ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> self . assertTrue ( self . response == ( <NUM_LIT:0> , None , <NUM_LIT:3> ) ) <EOL> def test_handle_unknown_request ( self ) : <EOL> self . protocol . send_response = self . mock_send_response <EOL> self . protocol . request ( <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" , ( <NUM_LIT:1> , <NUM_LIT:2> ) ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:0> ] == <NUM_LIT:0> ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] == KeyError . __name__ ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:2> ] == None ) <EOL> def test_handle_badargs_request ( self ) : <EOL> self . protocol . send_response = self . mock_send_response <EOL> self . protocol . request ( <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" , ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:0> ] == <NUM_LIT:0> ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] == TypeError . __name__ ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:2> ] == None ) <EOL> def test_handle_exceptioned_request ( self ) : <EOL> self . assertRaises ( Exception , self . protocol . request , <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def test_handle_rpcexceptioned_request ( self ) : <EOL> self . protocol . send_response = self . mock_send_response <EOL> self . protocol . request ( <NUM_LIT:0> , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> self . assertTrue ( self . response [ <NUM_LIT:1> ] != None ) <EOL> def test_notify ( self ) : <EOL> pass <EOL> def test_unknown_notify ( self ) : <EOL> pass <EOL> def test_badargs_notify ( self ) : <EOL> pass <EOL> def test_exceptioned_notify ( self ) : <EOL> pass <EOL> def test_rpcexceptioned_notify ( self ) : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
