<s> import warnings <EOL> warnings . simplefilter ( '<STR_LIT:ignore>' , DeprecationWarning ) <EOL> from eventlet import saranwrap <EOL> warnings . simplefilter ( '<STR_LIT:default>' , DeprecationWarning ) <EOL> from eventlet import greenpool , sleep <EOL> import os <EOL> import eventlet <EOL> import sys <EOL> import tempfile <EOL> import time <EOL> from tests import LimitedTestCase , main , skip_on_windows , skip_with_pyevent <EOL> import re <EOL> import StringIO <EOL> def list_maker ( ) : <EOL> return [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] <EOL> one = <NUM_LIT:1> <EOL> two = <NUM_LIT:2> <EOL> three = <NUM_LIT:3> <EOL> class CoroutineCallingClass ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _my_dict = { } <EOL> def run_coroutine ( self ) : <EOL> eventlet . spawn_n ( self . _add_random_key ) <EOL> def _add_random_key ( self ) : <EOL> self . _my_dict [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> def get_dict ( self ) : <EOL> return self . _my_dict <EOL> class TestSaranwrap ( LimitedTestCase ) : <EOL> TEST_TIMEOUT = <NUM_LIT:8> <EOL> def assert_server_exists ( self , prox ) : <EOL> self . assert_ ( saranwrap . status ( prox ) ) <EOL> prox . foo = <NUM_LIT:0> <EOL> self . assertEqual ( <NUM_LIT:0> , prox . foo ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_tuple ( self ) : <EOL> my_tuple = ( <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> prox = saranwrap . wrap ( my_tuple ) <EOL> self . assertEqual ( prox [ <NUM_LIT:0> ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( prox [ <NUM_LIT:1> ] , <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( my_tuple ) , <NUM_LIT:2> ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_string ( self ) : <EOL> my_object = "<STR_LIT>" <EOL> prox = saranwrap . wrap ( my_object ) <EOL> self . assertEqual ( str ( my_object ) , str ( prox ) ) <EOL> self . assertEqual ( len ( my_object ) , len ( prox ) ) <EOL> self . assertEqual ( my_object . join ( [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) , prox . join ( [ '<STR_LIT:a>' , '<STR_LIT:b>' ] ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_uniterable ( self ) : <EOL> prox = saranwrap . wrap ( FloatingPointError ( ) ) <EOL> def index ( ) : <EOL> prox [ <NUM_LIT:0> ] <EOL> def key ( ) : <EOL> prox [ '<STR_LIT:a>' ] <EOL> self . assertRaises ( IndexError , index ) <EOL> self . assertRaises ( TypeError , key ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_dict ( self ) : <EOL> my_object = { '<STR_LIT:a>' : <NUM_LIT:1> } <EOL> prox = saranwrap . wrap ( my_object ) <EOL> self . assertEqual ( '<STR_LIT:a>' , prox . keys ( ) [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , prox [ '<STR_LIT:a>' ] ) <EOL> self . assertEqual ( str ( my_object ) , str ( prox ) ) <EOL> self . assertEqual ( '<STR_LIT>' + repr ( my_object ) , repr ( prox ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_module_class ( self ) : <EOL> prox = saranwrap . wrap ( re ) <EOL> self . assertEqual ( saranwrap . Proxy , type ( prox ) ) <EOL> exp = prox . compile ( '<STR_LIT:.>' ) <EOL> self . assertEqual ( exp . flags , <NUM_LIT:0> ) <EOL> self . assert_ ( repr ( prox . compile ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_eq ( self ) : <EOL> prox = saranwrap . wrap ( re ) <EOL> exp1 = prox . compile ( '<STR_LIT:.>' ) <EOL> exp2 = prox . compile ( exp1 . pattern ) <EOL> self . assertEqual ( exp1 , exp2 ) <EOL> exp3 = prox . compile ( '<STR_LIT:/>' ) <EOL> self . assert_ ( exp1 != exp3 ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_nonzero ( self ) : <EOL> prox = saranwrap . wrap ( re ) <EOL> exp1 = prox . compile ( '<STR_LIT:.>' ) <EOL> self . assert_ ( bool ( exp1 ) ) <EOL> prox2 = saranwrap . Proxy ( [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] ) <EOL> self . assert_ ( bool ( prox2 ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_multiple_wraps ( self ) : <EOL> prox1 = saranwrap . wrap ( re ) <EOL> prox2 = saranwrap . wrap ( re ) <EOL> x1 = prox1 . compile ( '<STR_LIT:.>' ) <EOL> x2 = prox1 . compile ( '<STR_LIT:.>' ) <EOL> del x2 <EOL> x3 = prox2 . compile ( '<STR_LIT:.>' ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_dict_passthru ( self ) : <EOL> prox = saranwrap . wrap ( StringIO ) <EOL> x = prox . StringIO ( '<STR_LIT:a>' ) <EOL> self . assertEqual ( type ( x . __dict__ ) , saranwrap . ObjectProxy ) <EOL> self . assertEqual ( type ( saranwrap . wrap ( StringIO ) . StringIO ( '<STR_LIT:a>' ) . __dict__ ) , saranwrap . ObjectProxy ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_is_value ( self ) : <EOL> server = saranwrap . Server ( None , None , None ) <EOL> self . assert_ ( server . is_value ( None ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_getitem ( self ) : <EOL> prox = saranwrap . wrap ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> self . assertEqual ( prox [ <NUM_LIT:0> ] , <NUM_LIT:0> ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_wrap_setitem ( self ) : <EOL> prox = saranwrap . wrap ( [ <NUM_LIT:0> , <NUM_LIT:1> , <NUM_LIT:2> ] ) <EOL> prox [ <NUM_LIT:1> ] = <NUM_LIT:2> <EOL> self . assertEqual ( prox [ <NUM_LIT:1> ] , <NUM_LIT:2> ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_raising_exceptions ( self ) : <EOL> prox = saranwrap . wrap ( re ) <EOL> def nofunc ( ) : <EOL> prox . never_name_a_function_like_this ( ) <EOL> self . assertRaises ( AttributeError , nofunc ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_unpicklable_server_exception ( self ) : <EOL> prox = saranwrap . wrap ( saranwrap ) <EOL> def unpickle ( ) : <EOL> prox . raise_an_unpicklable_error ( ) <EOL> self . assertRaises ( saranwrap . UnrecoverableError , unpickle ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_pickleable_server_exception ( self ) : <EOL> prox = saranwrap . wrap ( saranwrap ) <EOL> def fperror ( ) : <EOL> prox . raise_standard_error ( ) <EOL> self . assertRaises ( FloatingPointError , fperror ) <EOL> self . assert_server_exists ( prox ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_print_does_not_break_wrapper ( self ) : <EOL> prox = saranwrap . wrap ( saranwrap ) <EOL> prox . print_string ( '<STR_LIT:hello>' ) <EOL> self . assert_server_exists ( prox ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_stderr_does_not_break_wrapper ( self ) : <EOL> prox = saranwrap . wrap ( saranwrap ) <EOL> prox . err_string ( '<STR_LIT>' ) <EOL> self . assert_server_exists ( prox ) <EOL> def assertLessThan ( self , a , b ) : <EOL> self . assert_ ( a < b , "<STR_LIT>" % ( a , b ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_status ( self ) : <EOL> prox = saranwrap . wrap ( time ) <EOL> a = prox . gmtime ( <NUM_LIT:0> ) <EOL> status = saranwrap . status ( prox ) <EOL> self . assertEqual ( status [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( status [ '<STR_LIT>' ] , <NUM_LIT:2> ) <EOL> self . assert_ ( status [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( saranwrap . status ( a ) , status ) <EOL> prox . gmtime ( <NUM_LIT:1> ) <EOL> is_id = prox . ctime ( <NUM_LIT:1> ) <EOL> status = saranwrap . status ( prox ) <EOL> self . assertEqual ( status [ '<STR_LIT>' ] , <NUM_LIT:1> ) <EOL> self . assertEqual ( status [ '<STR_LIT>' ] , <NUM_LIT:3> ) <EOL> prox2 = saranwrap . wrap ( re ) <EOL> self . assert_ ( status [ '<STR_LIT>' ] != saranwrap . status ( prox2 ) [ '<STR_LIT>' ] ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_del ( self ) : <EOL> prox = saranwrap . wrap ( time ) <EOL> delme = prox . gmtime ( <NUM_LIT:0> ) <EOL> status_before = saranwrap . status ( prox ) <EOL> del delme <EOL> prox . ctime ( <NUM_LIT:1> ) <EOL> status_after = saranwrap . status ( prox ) <EOL> self . assertLessThan ( status_after [ '<STR_LIT>' ] , status_before [ '<STR_LIT>' ] ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_contains ( self ) : <EOL> prox = saranwrap . wrap ( { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> self . assert_ ( '<STR_LIT:a>' in prox ) <EOL> self . assert_ ( '<STR_LIT:x>' not in prox ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_variable_and_keyword_arguments_with_function_calls ( self ) : <EOL> import optparse <EOL> prox = saranwrap . wrap ( optparse ) <EOL> parser = prox . OptionParser ( ) <EOL> z = parser . add_option ( '<STR_LIT>' , action = '<STR_LIT:store>' , type = '<STR_LIT:string>' , dest = '<STR_LIT:n>' ) <EOL> opts , args = parser . parse_args ( [ "<STR_LIT>" ] ) <EOL> self . assertEqual ( opts . n , '<STR_LIT:foo>' ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_original_proxy_going_out_of_scope ( self ) : <EOL> def make_re ( ) : <EOL> prox = saranwrap . wrap ( re ) <EOL> return prox . compile ( '<STR_LIT:.>' ) <EOL> tid = make_re ( ) <EOL> self . assertEqual ( tid . flags , <NUM_LIT:0> ) <EOL> def make_list ( ) : <EOL> from tests import saranwrap_test <EOL> prox = saranwrap . wrap ( saranwrap_test . list_maker ) <EOL> return prox ( ) <EOL> proxl = make_list ( ) <EOL> self . assertEqual ( proxl [ <NUM_LIT:2> ] , <NUM_LIT:2> ) <EOL> def test_status_of_none ( self ) : <EOL> try : <EOL> saranwrap . status ( None ) <EOL> self . assert_ ( False ) <EOL> except AttributeError , e : <EOL> pass <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_not_inheriting_pythonpath ( self ) : <EOL> temp_dir = tempfile . mkdtemp ( "<STR_LIT>" ) <EOL> fp = open ( os . path . join ( temp_dir , "<STR_LIT>" ) , "<STR_LIT:w>" ) <EOL> fp . write ( """<STR_LIT>""" ) <EOL> fp . close ( ) <EOL> prox = saranwrap . wrap_module ( '<STR_LIT>' ) <EOL> try : <EOL> prox . pypath <EOL> self . fail ( ) <EOL> except ImportError : <EOL> pass <EOL> sys . path . append ( temp_dir ) <EOL> try : <EOL> import tempmod <EOL> prox = saranwrap . wrap ( tempmod ) <EOL> self . assert_ ( prox . pypath . count ( temp_dir ) ) <EOL> self . assert_ ( prox . sys_path . count ( temp_dir ) ) <EOL> finally : <EOL> import shutil <EOL> shutil . rmtree ( temp_dir ) <EOL> sys . path . remove ( temp_dir ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_contention ( self ) : <EOL> from tests import saranwrap_test <EOL> prox = saranwrap . wrap ( saranwrap_test ) <EOL> pool = greenpool . GreenPool ( <NUM_LIT:4> ) <EOL> pool . spawn_n ( lambda : self . assertEquals ( prox . one , <NUM_LIT:1> ) ) <EOL> pool . spawn_n ( lambda : self . assertEquals ( prox . two , <NUM_LIT:2> ) ) <EOL> pool . spawn_n ( lambda : self . assertEquals ( prox . three , <NUM_LIT:3> ) ) <EOL> pool . waitall ( ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_copy ( self ) : <EOL> import copy <EOL> compound_object = { '<STR_LIT:a>' : [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ] } <EOL> prox = saranwrap . wrap ( compound_object ) <EOL> def make_assertions ( copied ) : <EOL> self . assert_ ( isinstance ( copied , dict ) ) <EOL> self . assert_ ( isinstance ( copied [ '<STR_LIT:a>' ] , list ) ) <EOL> self . assertEquals ( copied , compound_object ) <EOL> self . assertNotEqual ( id ( compound_object ) , id ( copied ) ) <EOL> make_assertions ( copy . copy ( prox ) ) <EOL> make_assertions ( copy . deepcopy ( prox ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_list_of_functions ( self ) : <EOL> return <EOL> from tests import saranwrap_test <EOL> prox = saranwrap . wrap ( [ saranwrap_test . list_maker ] ) <EOL> self . assertEquals ( list_maker ( ) , prox [ <NUM_LIT:0> ] ( ) ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_under_the_hood_coroutines ( self ) : <EOL> from tests import saranwrap_test <EOL> mod_proxy = saranwrap . wrap ( saranwrap_test ) <EOL> obj_proxy = mod_proxy . CoroutineCallingClass ( ) <EOL> obj_proxy . run_coroutine ( ) <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> self . assert_ ( <EOL> '<STR_LIT>' in obj_proxy . get_dict ( ) , <EOL> '<STR_LIT>' ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_child_process_death ( self ) : <EOL> prox = saranwrap . wrap ( { } ) <EOL> pid = saranwrap . getpid ( prox ) <EOL> self . assertEqual ( os . kill ( pid , <NUM_LIT:0> ) , None ) <EOL> del prox <EOL> sleep ( <NUM_LIT:0.1> ) <EOL> self . assertRaises ( OSError , os . kill , pid , <NUM_LIT:0> ) <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_detection_of_server_crash ( self ) : <EOL> pass <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_equality_with_local_object ( self ) : <EOL> pass <EOL> @ skip_on_windows <EOL> @ skip_with_pyevent <EOL> def test_non_blocking ( self ) : <EOL> pass <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
