<s> import pytest <EOL> import sys <EOL> import time <EOL> from . test_base_class import TestBaseClass <EOL> from aerospike import exception as e <EOL> aerospike = pytest . importorskip ( "<STR_LIT>" ) <EOL> try : <EOL> import aerospike <EOL> except : <EOL> print ( "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> class TestUdfPut ( TestBaseClass ) : <EOL> def setup_class ( cls ) : <EOL> """<STR_LIT>""" <EOL> hostlist , user , password = TestBaseClass . get_hosts ( ) <EOL> config = { '<STR_LIT>' : hostlist } <EOL> if user is None and password is None : <EOL> TestUdfPut . client = aerospike . client ( config ) . connect ( ) <EOL> else : <EOL> TestUdfPut . client = aerospike . client ( <EOL> config ) . connect ( user , password ) <EOL> def teardown_class ( cls ) : <EOL> TestUdfPut . client . close ( ) <EOL> def setup_method ( self , method ) : <EOL> """<STR_LIT>""" <EOL> def teardown_method ( self , method ) : <EOL> """<STR_LIT>""" <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> udf_list = TestUdfPut . client . udf_list ( { '<STR_LIT>' : <NUM_LIT:100> } ) <EOL> for udf in udf_list : <EOL> if udf [ '<STR_LIT:name>' ] == '<STR_LIT>' : <EOL> TestUdfPut . client . udf_remove ( "<STR_LIT>" ) <EOL> def test_udf_put_without_parameters ( self ) : <EOL> with pytest . raises ( TypeError ) as typeError : <EOL> TestUdfPut . client . udf_put ( ) <EOL> assert "<STR_LIT>" in str ( <EOL> typeError . value ) <EOL> def test_udf_put_with_proper_parameters ( self ) : <EOL> policy = { } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> status = TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> assert status == <NUM_LIT:0> <EOL> udf_list = TestUdfPut . client . udf_list ( { } ) <EOL> present = False <EOL> for udf in udf_list : <EOL> if '<STR_LIT>' == udf [ '<STR_LIT:name>' ] : <EOL> present = True <EOL> assert True if present else False <EOL> def test_udf_put_with_invalid_timeout_policy_value ( self ) : <EOL> policy = { '<STR_LIT>' : <NUM_LIT:0.1> } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> try : <EOL> TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> except e . ParamError as exception : <EOL> assert exception . code == - <NUM_LIT:2> <EOL> assert exception . msg == "<STR_LIT>" <EOL> def test_udf_put_with_proper_timeout_policy_value ( self ) : <EOL> policy = { '<STR_LIT>' : <NUM_LIT:1000> } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> status = TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> assert status == <NUM_LIT:0> <EOL> udf_list = TestUdfPut . client . udf_list ( { } ) <EOL> for udf in udf_list : <EOL> if '<STR_LIT>' == udf [ '<STR_LIT:name>' ] : <EOL> _ = True <EOL> def test_udf_put_with_non_existent_filename ( self ) : <EOL> policy = { } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> try : <EOL> TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> except e . LuaFileNotFound as exception : <EOL> assert exception . code == <NUM_LIT> <EOL> def test_udf_put_with_non_lua_udf_type_and_lua_script_file ( self ) : <EOL> policy = { '<STR_LIT>' : <NUM_LIT:0> } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:1> <EOL> try : <EOL> TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> except e . ClientError as exception : <EOL> assert exception . code == - <NUM_LIT:1> <EOL> assert exception . msg == "<STR_LIT>" <EOL> def test_udf_put_with_all_none_parameters ( self ) : <EOL> with pytest . raises ( TypeError ) as exception : <EOL> TestUdfPut . client . udf_put ( None , None , None ) <EOL> assert "<STR_LIT>" in str ( exception . value ) <EOL> def test_udf_put_with_filename_unicode ( self ) : <EOL> policy = { } <EOL> filename = u"<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> status = TestUdfPut . client . udf_put ( filename , udf_type , policy ) <EOL> assert status == <NUM_LIT:0> <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> udf_list = TestUdfPut . client . udf_list ( { } ) <EOL> present = False <EOL> for udf in udf_list : <EOL> if '<STR_LIT>' == udf [ '<STR_LIT:name>' ] : <EOL> present = True <EOL> assert True if present else False <EOL> def test_udf_put_with_proper_parameters_without_connection ( self ) : <EOL> policy = { } <EOL> filename = "<STR_LIT>" <EOL> udf_type = <NUM_LIT:0> <EOL> config = { '<STR_LIT>' : [ ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ] } <EOL> client1 = aerospike . client ( config ) <EOL> try : <EOL> client1 . udf_put ( filename , udf_type , policy ) <EOL> except e . ClusterError as exception : <EOL> assert exception . code == <NUM_LIT:11> <EOL> assert exception . msg == '<STR_LIT>' </s>
