<s> from mock import patch , call <EOL> from unittest import TestCase <EOL> from itertools import product <EOL> from json import dumps <EOL> from sys import stdout <EOL> import json <EOL> from kamaki . clients import ClientError , compute <EOL> rest_pkg = '<STR_LIT>' <EOL> compute_pkg = '<STR_LIT>' <EOL> img_ref = "<STR_LIT>" <EOL> vm_name = "<STR_LIT>" <EOL> fid = <NUM_LIT> <EOL> vm_send = dict ( server = dict ( <EOL> flavorRef = fid , <EOL> name = vm_name , <EOL> imageRef = img_ref , <EOL> metadata = dict ( os = "<STR_LIT>" , users = "<STR_LIT:root>" ) ) ) <EOL> vm_recv = dict ( server = dict ( <EOL> status = "<STR_LIT>" , <EOL> updated = "<STR_LIT>" , <EOL> hostId = "<STR_LIT>" , <EOL> name = vm_name , <EOL> imageRef = img_ref , <EOL> created = "<STR_LIT>" , <EOL> flavorRef = fid , <EOL> adminPass = "<STR_LIT>" , <EOL> suspended = False , <EOL> progress = <NUM_LIT:0> , <EOL> id = <NUM_LIT> , <EOL> metadata = dict ( os = "<STR_LIT>" , users = "<STR_LIT:root>" ) ) ) <EOL> img_recv = dict ( image = dict ( <EOL> status = "<STR_LIT>" , <EOL> updated = "<STR_LIT>" , <EOL> name = "<STR_LIT>" , <EOL> created = "<STR_LIT>" , <EOL> progress = <NUM_LIT:100> , <EOL> id = img_ref , <EOL> metadata = dict ( <EOL> partition_table = "<STR_LIT>" , <EOL> kernel = "<STR_LIT>" , <EOL> osfamily = "<STR_LIT>" , <EOL> users = "<STR_LIT:root>" , <EOL> gui = "<STR_LIT>" , <EOL> sortorder = "<STR_LIT:1>" , <EOL> os = "<STR_LIT>" , <EOL> root_partition = "<STR_LIT:1>" , <EOL> description = "<STR_LIT>" ) ) ) <EOL> vm_list = dict ( servers = [ <EOL> dict ( name = '<STR_LIT>' , id = <NUM_LIT:1> ) , <EOL> dict ( name = '<STR_LIT>' , id = <NUM_LIT:2> ) ] ) <EOL> flavor_list = dict ( flavors = [ <EOL> dict ( id = <NUM_LIT> , name = "<STR_LIT>" ) , <EOL> dict ( id = <NUM_LIT> , name = "<STR_LIT>" ) , <EOL> dict ( id = <NUM_LIT> , name = "<STR_LIT>" ) ] ) <EOL> img_list = dict ( images = [ <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) , <EOL> dict ( name = "<STR_LIT>" , id = "<STR_LIT>" ) ] ) <EOL> class FR ( object ) : <EOL> """<STR_LIT>""" <EOL> json = vm_recv <EOL> headers = { } <EOL> content = json <EOL> status = None <EOL> status_code = <NUM_LIT:200> <EOL> def print_iterations ( old , new ) : <EOL> if new : <EOL> if new % <NUM_LIT:1000> : <EOL> return old <EOL> stdout . write ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % old ) ) <EOL> stdout . write ( '<STR_LIT:%s>' % new ) <EOL> else : <EOL> stdout . write ( '<STR_LIT>' ) <EOL> stdout . flush ( ) <EOL> return new <EOL> class ComputeRestClient ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . url = '<STR_LIT>' <EOL> self . token = '<STR_LIT>' <EOL> self . client = compute . ComputeRestClient ( self . url , self . token ) <EOL> def tearDown ( self ) : <EOL> FR . json = vm_recv <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def _test_get ( self , service , params , get , set_param ) : <EOL> method = getattr ( self . client , '<STR_LIT>' % service ) <EOL> param_args = [ ( { } , { k : k } , { k : v [ <NUM_LIT:1> ] } ) for k , v in params . items ( ) ] <EOL> num_of_its = <NUM_LIT:0> <EOL> for i , args in enumerate ( product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' % service ) , <EOL> ( None , False , True ) , <EOL> ( <NUM_LIT:200> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) , <EOL> * param_args ) ) : <EOL> ( srv_id , detail , success , kwargs ) = args [ : <NUM_LIT:4> ] <EOL> kwargs [ '<STR_LIT:success>' ] = success <EOL> srv_kwargs = dict ( ) <EOL> for param in args [ <NUM_LIT:4> : ] : <EOL> srv_kwargs . update ( param ) <EOL> srv_kwargs . update ( kwargs ) <EOL> method ( * args [ : <NUM_LIT:2> ] , ** srv_kwargs ) <EOL> srv_str = '<STR_LIT>' if detail else ( <EOL> '<STR_LIT>' % srv_id ) if srv_id else '<STR_LIT>' <EOL> self . assertEqual ( <EOL> get . mock_calls [ - <NUM_LIT:1> ] , <EOL> call ( '<STR_LIT>' % ( service , srv_str ) , ** kwargs ) ) <EOL> param_calls = [ ] <EOL> for k , v in params . items ( ) : <EOL> real_v = srv_kwargs . get ( k , v [ <NUM_LIT:1> ] ) if not srv_id else v [ <NUM_LIT:1> ] <EOL> param_calls . append ( call ( v [ <NUM_LIT:0> ] , real_v , iff = real_v ) ) <EOL> actual = set_param . mock_calls [ - len ( param_calls ) : ] <EOL> self . assertEqual ( sorted ( actual ) , sorted ( param_calls ) ) <EOL> num_of_its = print_iterations ( num_of_its , i ) <EOL> print ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % num_of_its ) ) + ( '<STR_LIT:%s>' % i ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def _test_srv_cmd_get ( self , srv , cmd , params , get , set_param ) : <EOL> method = getattr ( self . client , '<STR_LIT>' % ( srv , cmd ) ) <EOL> param_args = [ ( { } , { k : k } , { k : v [ <NUM_LIT:1> ] } ) for k , v in params . items ( ) ] <EOL> num_of_its = <NUM_LIT:0> <EOL> for i , args in enumerate ( product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( None , '<STR_LIT>' ) , <EOL> ( ( <NUM_LIT> , <NUM_LIT:200> ) , ( <NUM_LIT:1000> ) ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) , <EOL> * param_args ) ) : <EOL> srv_id , xtra_id , success , kwargs = args [ : <NUM_LIT:4> ] <EOL> kwargs = dict ( kwargs ) <EOL> kwargs [ '<STR_LIT:success>' ] = success <EOL> srv_kwargs = dict ( ) <EOL> for param in args [ <NUM_LIT:4> : ] : <EOL> srv_kwargs . update ( param ) <EOL> srv_kwargs . update ( kwargs ) <EOL> method ( * args [ : <NUM_LIT:2> ] , ** srv_kwargs ) <EOL> srv_str = '<STR_LIT>' % ( srv , srv_id , cmd ) <EOL> srv_str += ( '<STR_LIT>' % xtra_id ) if xtra_id else '<STR_LIT>' <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( srv_str , ** kwargs ) ) <EOL> param_calls = [ ] <EOL> for k , v in params . items ( ) : <EOL> real_v = srv_kwargs . get ( k , v [ <NUM_LIT:1> ] ) <EOL> param_calls . append ( call ( v [ <NUM_LIT:0> ] , real_v , iff = real_v ) ) <EOL> actual = set_param . mock_calls [ - len ( param_calls ) : ] <EOL> self . assertEqual ( sorted ( actual ) , sorted ( param_calls ) ) <EOL> num_of_its = print_iterations ( num_of_its , i ) <EOL> print ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % num_of_its ) ) + ( '<STR_LIT:%s>' % i ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def _test_post ( self , srv , cmd , params , post , set_param , set_header ) : <EOL> method = getattr ( <EOL> self . client , '<STR_LIT>' % ( srv , ( '<STR_LIT>' % cmd ) if cmd else '<STR_LIT>' ) ) <EOL> param_args = [ ( { } , { k : k } , { k : v [ <NUM_LIT:1> ] } ) for k , v in params . items ( ) ] <EOL> num_of_its = <NUM_LIT:0> <EOL> for i , args in enumerate ( product ( <EOL> ( '<STR_LIT>' % srv , '<STR_LIT>' ) , <EOL> ( <EOL> None , <EOL> { '<STR_LIT>' : { '<STR_LIT:data>' : '<STR_LIT>' } } , <EOL> [ '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , '<STR_LIT>' ] ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) , <EOL> * param_args ) ) : <EOL> srv_id , json_data , success , kwargs = args [ : <NUM_LIT:4> ] <EOL> kwargs = dict ( kwargs ) <EOL> cmd_args = ( srv_id , ) if cmd else ( ) <EOL> kwargs [ '<STR_LIT:success>' ] = success <EOL> srv_kwargs = dict ( ) <EOL> for param in args [ <NUM_LIT:4> : ] : <EOL> srv_kwargs . update ( param ) <EOL> srv_kwargs . update ( kwargs ) <EOL> srv_kwargs [ '<STR_LIT>' ] = json_data <EOL> method ( * cmd_args , ** srv_kwargs ) <EOL> srv_str = '<STR_LIT>' % ( <EOL> srv , ( ( '<STR_LIT>' % ( srv_id , cmd ) ) if cmd else '<STR_LIT>' ) ) <EOL> kwargs [ '<STR_LIT:data>' ] = json . dumps ( json_data ) if json_data else None <EOL> self . assertEqual ( post . mock_calls [ - <NUM_LIT:1> ] , call ( srv_str , ** kwargs ) ) <EOL> param_calls = [ ] <EOL> for k , v in params . items ( ) : <EOL> real_v = srv_kwargs . get ( k , v [ <NUM_LIT:1> ] ) <EOL> param_calls . append ( call ( v [ <NUM_LIT:0> ] , real_v , iff = real_v ) ) <EOL> actual = set_param . mock_calls [ - len ( param_calls ) : ] <EOL> self . assertEqual ( sorted ( actual ) , sorted ( param_calls ) ) <EOL> if json_data : <EOL> self . assertEqual ( set_header . mock_calls [ - <NUM_LIT:2> : ] , [ <EOL> call ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , <EOL> call ( '<STR_LIT>' , len ( kwargs [ '<STR_LIT:data>' ] ) ) ] ) <EOL> num_of_its = print_iterations ( num_of_its , i ) <EOL> print ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % num_of_its ) ) + ( '<STR_LIT:%s>' % i ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def _test_put ( self , srv , cmd , params , put , set_param , set_headers ) : <EOL> method = getattr ( self . client , '<STR_LIT>' % ( <EOL> srv , ( '<STR_LIT>' % cmd ) if cmd else '<STR_LIT>' ) ) <EOL> param_args = [ ( { } , { k : k } , { k : v [ <NUM_LIT:1> ] } ) for k , v in params . items ( ) ] <EOL> num_of_its = <NUM_LIT:0> <EOL> for i , args in enumerate ( product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' % srv ) , <EOL> ( None , [ dict ( json = "<STR_LIT:data>" ) , dict ( data = "<STR_LIT>" ) ] ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) , <EOL> * param_args ) ) : <EOL> srv_id , json_data , success , kwargs = args [ : <NUM_LIT:4> ] <EOL> kwargs = dict ( kwargs ) <EOL> kwargs [ '<STR_LIT:success>' ] = success <EOL> srv_kwargs = dict ( ) <EOL> for param in args [ <NUM_LIT:4> : ] : <EOL> srv_kwargs . update ( param ) <EOL> srv_kwargs . update ( kwargs ) <EOL> srv_kwargs [ '<STR_LIT>' ] = json_data <EOL> method ( srv_id , ** srv_kwargs ) <EOL> srv_str = '<STR_LIT>' % ( srv , srv_id , ( '<STR_LIT>' % cmd if cmd else '<STR_LIT>' ) ) <EOL> if json_data : <EOL> json_data = dumps ( json_data ) <EOL> self . assertEqual ( set_headers . mock_calls [ - <NUM_LIT:2> : ] , [ <EOL> call ( '<STR_LIT:Content-Type>' , '<STR_LIT:application/json>' ) , <EOL> call ( '<STR_LIT>' , len ( json_data ) ) ] ) <EOL> self . assertEqual ( <EOL> put . mock_calls [ - <NUM_LIT:1> ] , <EOL> call ( srv_str , data = json_data , ** kwargs ) ) <EOL> param_calls = [ ] <EOL> for k , v in params . items ( ) : <EOL> real_v = srv_kwargs . get ( k , v [ <NUM_LIT:1> ] ) <EOL> param_calls . append ( call ( v [ <NUM_LIT:0> ] , real_v , iff = real_v ) ) <EOL> actual = set_param . mock_calls [ - len ( param_calls ) : ] <EOL> self . assertEqual ( sorted ( actual ) , sorted ( param_calls ) ) <EOL> num_of_its = print_iterations ( num_of_its , i ) <EOL> print ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % num_of_its ) ) + ( '<STR_LIT:%s>' % i ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def _test_delete ( self , srv , cmd , delete ) : <EOL> method = getattr ( <EOL> self . client , '<STR_LIT>' % ( srv , ( '<STR_LIT>' % cmd ) if cmd else '<STR_LIT>' ) ) <EOL> cmd_params = ( '<STR_LIT>' , '<STR_LIT>' ) if cmd else ( ) <EOL> num_of_its = <NUM_LIT:0> <EOL> for i , args in enumerate ( product ( <EOL> ( '<STR_LIT>' % srv , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) , <EOL> * cmd_params ) ) : <EOL> ( srv_id , success , kwargs ) = args [ : <NUM_LIT:3> ] <EOL> kwargs = dict ( kwargs ) <EOL> kwargs [ '<STR_LIT:success>' ] = success <EOL> cmd_value = args [ - <NUM_LIT:1> ] if cmd else '<STR_LIT>' <EOL> method_args = ( srv_id , cmd_value ) if cmd else ( srv_id , ) <EOL> method ( * method_args , ** kwargs ) <EOL> srv_str = '<STR_LIT>' % ( srv , srv_id ) <EOL> cmd_str = ( '<STR_LIT>' % ( cmd , cmd_value ) ) if cmd else '<STR_LIT>' <EOL> self . assertEqual ( <EOL> delete . mock_calls [ - <NUM_LIT:1> ] , <EOL> call ( '<STR_LIT>' % ( srv_str , cmd_str ) , ** kwargs ) ) <EOL> num_of_its = print_iterations ( num_of_its , i ) <EOL> print ( '<STR_LIT>' * len ( '<STR_LIT:%s>' % num_of_its ) ) + ( '<STR_LIT:%s>' % i ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def test_limits_get ( self , get ) : <EOL> self . client . limits_get ( success = '<STR_LIT>' ) <EOL> get . assert_called_once_with ( '<STR_LIT>' , success = '<STR_LIT>' ) <EOL> def test_servers_get ( self ) : <EOL> params = dict ( <EOL> changes_since = ( '<STR_LIT>' , None ) , <EOL> image = ( '<STR_LIT:image>' , None ) , <EOL> flavor = ( '<STR_LIT>' , None ) , <EOL> name = ( '<STR_LIT:name>' , None ) , <EOL> marker = ( '<STR_LIT>' , None ) , <EOL> limit = ( '<STR_LIT>' , None ) , <EOL> status = ( '<STR_LIT:status>' , None ) , <EOL> host = ( '<STR_LIT:host>' , None ) ) <EOL> self . _test_get ( '<STR_LIT>' , params ) <EOL> def test_servers_post ( self ) : <EOL> params = dict ( <EOL> security_group = ( '<STR_LIT>' , None ) , <EOL> user_data = ( '<STR_LIT>' , None ) , <EOL> availability_zone = ( '<STR_LIT>' , None ) ) <EOL> self . _test_post ( '<STR_LIT>' , None , params ) <EOL> def test_servers_put ( self ) : <EOL> self . _test_put ( '<STR_LIT>' , None , dict ( server_name = ( '<STR_LIT>' , None ) ) ) <EOL> def test_servers_delete ( self ) : <EOL> self . _test_delete ( '<STR_LIT>' , None ) <EOL> def test_servers_metadata_get ( self ) : <EOL> self . _test_srv_cmd_get ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_servers_metadata_post ( self ) : <EOL> self . _test_post ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_servers_metadata_put ( self ) : <EOL> self . _test_put ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_images_metadata_delete ( self ) : <EOL> self . _test_delete ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_servers_action_post ( self ) : <EOL> self . _test_post ( '<STR_LIT>' , '<STR_LIT:action>' , { } ) <EOL> def test_servers_ips_get ( self ) : <EOL> params = dict ( changes_since = ( '<STR_LIT>' , None ) ) <EOL> self . _test_srv_cmd_get ( '<STR_LIT>' , '<STR_LIT>' , params ) <EOL> def test_images_get ( self ) : <EOL> param = dict ( <EOL> changes_since = ( '<STR_LIT>' , None ) , <EOL> server_name = ( '<STR_LIT>' , None ) , <EOL> name = ( '<STR_LIT:name>' , None ) , <EOL> status = ( '<STR_LIT:status>' , None ) , <EOL> marker = ( '<STR_LIT>' , None ) , <EOL> limit = ( '<STR_LIT>' , None ) , <EOL> type = ( '<STR_LIT:type>' , None ) ) <EOL> self . _test_get ( '<STR_LIT>' , param ) <EOL> def test_images_delete ( self ) : <EOL> self . _test_delete ( '<STR_LIT>' , None ) <EOL> def test_images_metadata_get ( self ) : <EOL> self . _test_srv_cmd_get ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_images_metadata_post ( self ) : <EOL> self . _test_post ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_images_metadata_put ( self ) : <EOL> self . _test_put ( '<STR_LIT>' , '<STR_LIT>' , { } ) <EOL> def test_servers_metadata_delete ( self ) : <EOL> self . _test_delete ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_flavors_get ( self ) : <EOL> params = dict ( <EOL> changes_since = ( '<STR_LIT>' , None ) , <EOL> minDisk = ( '<STR_LIT>' , None ) , <EOL> minRam = ( '<STR_LIT>' , None ) , <EOL> marker = ( '<STR_LIT>' , None ) , <EOL> limit = ( '<STR_LIT>' , None ) ) <EOL> self . _test_get ( '<STR_LIT>' , params ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def test_floating_ip_pools_get ( self , get ) : <EOL> for args in product ( <EOL> ( <NUM_LIT:200> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) ) : <EOL> success , kwargs = args <EOL> r = self . client . floating_ip_pools_get ( success , ** kwargs ) <EOL> self . assertTrue ( isinstance ( r , FR ) ) <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( <EOL> '<STR_LIT>' , success = success , ** kwargs ) ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def test_floating_ips_get ( self , get ) : <EOL> for args in product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( <NUM_LIT:200> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) ) : <EOL> ip , success , kwargs = args <EOL> r = self . client . floating_ips_get ( * args [ : <NUM_LIT:2> ] , ** kwargs ) <EOL> self . assertTrue ( isinstance ( r , FR ) ) <EOL> expected = '<STR_LIT>' if not ip else '<STR_LIT>' % ip <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( <EOL> '<STR_LIT>' % expected , success = success , ** kwargs ) ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def test_floating_ips_post ( self , post ) : <EOL> for args in product ( <EOL> ( None , [ dict ( json = '<STR_LIT:data>' ) , dict ( data = '<STR_LIT>' ) ] ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , <NUM_LIT> ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) ) : <EOL> json_data , ip , success , kwargs = args <EOL> self . client . floating_ips_post ( * args [ : <NUM_LIT:3> ] , ** kwargs ) <EOL> expected = '<STR_LIT>' if not ip else '<STR_LIT>' % ip <EOL> self . assertEqual ( post . mock_calls [ - <NUM_LIT:1> ] , call ( <EOL> '<STR_LIT>' % expected , <EOL> json = json_data , success = success , <EOL> ** kwargs ) ) <EOL> @ patch ( '<STR_LIT>' % rest_pkg , return_value = FR ( ) ) <EOL> def test_floating_ips_delete ( self , delete ) : <EOL> for args in product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( <NUM_LIT> , ) , <EOL> ( { } , { '<STR_LIT:k>' : '<STR_LIT:v>' } ) ) : <EOL> ip , success , kwargs = args <EOL> r = self . client . floating_ips_delete ( * args [ : <NUM_LIT:2> ] , ** kwargs ) <EOL> self . assertTrue ( isinstance ( r , FR ) ) <EOL> expected = '<STR_LIT>' if not ip else '<STR_LIT>' % ip <EOL> self . assertEqual ( delete . mock_calls [ - <NUM_LIT:1> ] , call ( <EOL> '<STR_LIT>' % expected , success = success , ** kwargs ) ) <EOL> class ComputeClient ( TestCase ) : <EOL> def assert_dicts_are_equal ( self , d1 , d2 ) : <EOL> for k , v in d1 . items ( ) : <EOL> self . assertTrue ( k in d2 ) <EOL> if isinstance ( v , dict ) : <EOL> self . assert_dicts_are_equal ( v , d2 [ k ] ) <EOL> else : <EOL> self . assertEqual ( unicode ( v ) , unicode ( d2 [ k ] ) ) <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . url = '<STR_LIT>' <EOL> self . token = '<STR_LIT>' <EOL> self . client = compute . ComputeClient ( self . url , self . token ) <EOL> def tearDown ( self ) : <EOL> FR . status_code = <NUM_LIT:200> <EOL> FR . json = vm_recv <EOL> def test_create_server ( self ) : <EOL> with patch . object ( <EOL> compute . ComputeClient , '<STR_LIT>' , <EOL> side_effect = ClientError ( <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT> ) ) : <EOL> self . assertRaises ( <EOL> ClientError , <EOL> self . client . create_server , <EOL> vm_name , fid , img_ref ) <EOL> for params in product ( <EOL> ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , None ) , <EOL> ( '<STR_LIT>' , None ) , <EOL> ( None , { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:root>' } ) ) : <EOL> kwargs = dict ( ) <EOL> for i , k in enumerate ( ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> if params [ i ] : <EOL> kwargs [ k ] = params [ i ] <EOL> with patch . object ( <EOL> compute . ComputeClient , '<STR_LIT>' , <EOL> return_value = FR ( ) ) as post : <EOL> r = self . client . create_server ( vm_name , fid , img_ref , ** kwargs ) <EOL> self . assertEqual ( r , FR . json [ '<STR_LIT>' ] ) <EOL> exp_json = dict ( server = dict ( <EOL> flavorRef = fid , name = vm_name , imageRef = img_ref ) ) <EOL> for k in set ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) . difference ( kwargs ) : <EOL> kwargs [ k ] = None <EOL> self . assertEqual ( <EOL> post . mock_calls [ - <NUM_LIT:1> ] , call ( json_data = exp_json , ** kwargs ) ) <EOL> prsn = '<STR_LIT>' <EOL> self . client . create_server ( <EOL> vm_name , fid , img_ref , personality = prsn , ** kwargs ) <EOL> exp_json [ '<STR_LIT>' ] [ '<STR_LIT>' ] = prsn <EOL> self . assertEqual ( <EOL> post . mock_calls [ - <NUM_LIT:1> ] , call ( json_data = exp_json , ** kwargs ) ) <EOL> kwargs . pop ( '<STR_LIT>' , None ) <EOL> exp_json [ '<STR_LIT>' ] . pop ( '<STR_LIT>' , None ) <EOL> mtdt = '<STR_LIT>' <EOL> self . client . create_server ( <EOL> vm_name , fid , img_ref , metadata = mtdt , ** kwargs ) <EOL> exp_json [ '<STR_LIT>' ] [ '<STR_LIT>' ] = mtdt <EOL> self . assertEqual ( <EOL> post . mock_calls [ - <NUM_LIT:1> ] , call ( json_data = exp_json , ** kwargs ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_list_servers ( self , SG ) : <EOL> FR . json = vm_list <EOL> for args in product ( <EOL> ( False , True ) , <EOL> ( { } , dict ( status = '<STR_LIT:status>' ) ) , <EOL> ( { } , dict ( name = '<STR_LIT:name>' ) ) , <EOL> ( { } , dict ( image = '<STR_LIT:image>' ) ) , <EOL> ( { } , dict ( flavor = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( host = '<STR_LIT:host>' ) ) , <EOL> ( { } , dict ( limit = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( marker = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( changes_since = '<STR_LIT>' ) ) ) : <EOL> detail = args [ <NUM_LIT:0> ] <EOL> kwargs = dict ( ) <EOL> for param in args [ <NUM_LIT:1> : ] : <EOL> kwargs . update ( param ) <EOL> r = self . client . list_servers ( detail , ** kwargs ) <EOL> for k in set ( [ <EOL> '<STR_LIT:status>' , '<STR_LIT:name>' , <EOL> '<STR_LIT:image>' , '<STR_LIT>' , <EOL> '<STR_LIT:host>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) . difference ( kwargs ) : <EOL> kwargs [ k ] = None <EOL> self . assertEqual ( SG . mock_calls [ - <NUM_LIT:1> ] , call ( detail = detail , ** kwargs ) ) <EOL> for i , vm in enumerate ( vm_list [ '<STR_LIT>' ] ) : <EOL> self . assert_dicts_are_equal ( r [ i ] , vm ) <EOL> self . assertEqual ( i + <NUM_LIT:1> , len ( r ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_server_details ( self , SG ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> for args in product ( <EOL> ( { } , dict ( status = '<STR_LIT:status>' ) ) , <EOL> ( { } , dict ( name = '<STR_LIT:name>' ) ) , <EOL> ( { } , dict ( image = '<STR_LIT:image>' ) ) , <EOL> ( { } , dict ( flavor = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( host = '<STR_LIT:host>' ) ) , <EOL> ( { } , dict ( limit = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( marker = '<STR_LIT>' ) ) , <EOL> ( { } , dict ( changes_since = '<STR_LIT>' ) ) ) : <EOL> kwargs = dict ( ) <EOL> for param in args : <EOL> kwargs . update ( param ) <EOL> r = self . client . get_server_details ( vm_id , ** kwargs ) <EOL> for k in set ( [ <EOL> '<STR_LIT:status>' , '<STR_LIT:name>' , <EOL> '<STR_LIT:image>' , '<STR_LIT>' , <EOL> '<STR_LIT:host>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) . difference ( kwargs ) : <EOL> kwargs [ k ] = None <EOL> self . assertEqual ( SG . mock_calls [ - <NUM_LIT:1> ] , call ( vm_id , ** kwargs ) ) <EOL> self . assert_dicts_are_equal ( r , vm_recv [ '<STR_LIT>' ] ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_update_server_name ( self , SP ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> new_name = vm_name + '<STR_LIT>' <EOL> self . client . update_server_name ( vm_id , new_name ) <EOL> SP . assert_called_once_with ( vm_id , json_data = dict ( <EOL> server = dict ( name = new_name ) ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_reboot_server ( self , SP ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> for hard in ( None , True ) : <EOL> self . client . reboot_server ( vm_id , hard = hard ) <EOL> self . assertEqual ( SP . mock_calls [ - <NUM_LIT:1> ] , call ( vm_id , json_data = dict ( <EOL> reboot = dict ( type = '<STR_LIT>' if hard else '<STR_LIT>' ) ) ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_resize_server ( self , SP ) : <EOL> vm_id , flavor = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] , flavor_list [ '<STR_LIT>' ] [ <NUM_LIT:1> ] <EOL> self . client . resize_server ( vm_id , flavor [ '<STR_LIT:id>' ] ) <EOL> exp = dict ( resize = dict ( flavorRef = flavor [ '<STR_LIT:id>' ] ) ) <EOL> SP . assert_called_once_with ( vm_id , json_data = exp ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_create_server_metadata ( self , SP ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> metadata = dict ( m1 = '<STR_LIT>' , m2 = '<STR_LIT>' , m3 = '<STR_LIT>' ) <EOL> FR . json = dict ( meta = vm_recv [ '<STR_LIT>' ] ) <EOL> for k , v in metadata . items ( ) : <EOL> r = self . client . create_server_metadata ( vm_id , k , v ) <EOL> self . assert_dicts_are_equal ( r , vm_recv [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( SP . mock_calls [ - <NUM_LIT:1> ] , call ( <EOL> vm_id , '<STR_LIT:%s>' % k , <EOL> json_data = dict ( meta = { k : v } ) , success = <NUM_LIT> ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_server_metadata ( self , SG ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> metadata = dict ( m1 = '<STR_LIT>' , m2 = '<STR_LIT>' , m3 = '<STR_LIT>' ) <EOL> FR . json = dict ( metadata = metadata ) <EOL> r = self . client . get_server_metadata ( vm_id ) <EOL> FR . json = dict ( meta = metadata ) <EOL> SG . assert_called_once_with ( vm_id , '<STR_LIT>' ) <EOL> self . assert_dicts_are_equal ( r , metadata ) <EOL> for k , v in metadata . items ( ) : <EOL> FR . json = dict ( meta = { k : v } ) <EOL> r = self . client . get_server_metadata ( vm_id , k ) <EOL> self . assert_dicts_are_equal ( r , { k : v } ) <EOL> self . assertEqual ( <EOL> SG . mock_calls [ - <NUM_LIT:1> ] , call ( vm_id , '<STR_LIT:%s>' % k ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_update_server_metadata ( self , SP ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> metadata = dict ( m1 = '<STR_LIT>' , m2 = '<STR_LIT>' , m3 = '<STR_LIT>' ) <EOL> FR . json = dict ( metadata = metadata ) <EOL> r = self . client . update_server_metadata ( vm_id , ** metadata ) <EOL> self . assert_dicts_are_equal ( r , metadata ) <EOL> SP . assert_called_once_with ( <EOL> vm_id , json_data = dict ( metadata = metadata ) , success = <NUM_LIT> ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_delete_server_metadata ( self , SD ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> key = '<STR_LIT>' <EOL> self . client . delete_server_metadata ( vm_id , key ) <EOL> SD . assert_called_once_with ( vm_id , key ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_list_flavors ( self , FG ) : <EOL> FR . json = flavor_list <EOL> for detail in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> r = self . client . list_flavors ( detail = bool ( detail ) ) <EOL> self . assertEqual ( FG . mock_calls [ - <NUM_LIT:1> ] , call ( detail = bool ( detail ) ) ) <EOL> self . assertEqual ( r , flavor_list [ '<STR_LIT>' ] ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_flavor_details ( self , FG ) : <EOL> FR . json = dict ( flavor = flavor_list [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> r = self . client . get_flavor_details ( fid ) <EOL> FG . assert_called_once_with ( fid ) <EOL> self . assert_dicts_are_equal ( r , flavor_list [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_list_images ( self , IG ) : <EOL> FR . json = img_list <EOL> for detail in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> r = self . client . list_images ( detail = detail ) <EOL> self . assertEqual ( IG . mock_calls [ - <NUM_LIT:1> ] , call ( detail = bool ( detail ) ) ) <EOL> expected = img_list [ '<STR_LIT>' ] <EOL> for i in range ( len ( r ) ) : <EOL> self . assert_dicts_are_equal ( expected [ i ] , r [ i ] ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_image_details ( self , IG ) : <EOL> FR . json = img_recv <EOL> r = self . client . get_image_details ( img_ref ) <EOL> IG . assert_called_once_with ( img_ref ) <EOL> self . assert_dicts_are_equal ( r , img_recv [ '<STR_LIT:image>' ] ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_image_metadata ( self , IG ) : <EOL> for key in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> FR . json = dict ( meta = img_recv [ '<STR_LIT:image>' ] ) if ( <EOL> key ) else dict ( metadata = img_recv [ '<STR_LIT:image>' ] ) <EOL> r = self . client . get_image_metadata ( img_ref , key ) <EOL> self . assertEqual ( IG . mock_calls [ - <NUM_LIT:1> ] , call ( img_ref , key or '<STR_LIT>' ) ) <EOL> self . assert_dicts_are_equal ( img_recv [ '<STR_LIT:image>' ] , r ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_delete_server ( self , SD ) : <EOL> vm_id = vm_recv [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . client . delete_server ( vm_id ) <EOL> SD . assert_called_once_with ( vm_id ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_delete_image ( self , ID ) : <EOL> self . client . delete_image ( img_ref ) <EOL> ID . assert_called_once_with ( img_ref ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_update_image_metadata ( self , IP ) : <EOL> metadata = dict ( m1 = '<STR_LIT>' , m2 = '<STR_LIT>' , m3 = '<STR_LIT>' ) <EOL> FR . json = dict ( metadata = metadata ) <EOL> r = self . client . update_image_metadata ( img_ref , ** metadata ) <EOL> IP . assert_called_once_with ( <EOL> img_ref , json_data = dict ( metadata = metadata ) ) <EOL> self . assert_dicts_are_equal ( r , metadata ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_delete_image_metadata ( self , ID ) : <EOL> key = '<STR_LIT>' <EOL> self . client . delete_image_metadata ( img_ref , key ) <EOL> ID . assert_called_once_with ( img_ref , '<STR_LIT:%s>' % key ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_floating_ip_pools ( self , get ) : <EOL> tid = '<STR_LIT>' <EOL> r = self . client . get_floating_ip_pools ( tid ) <EOL> self . assert_dicts_are_equal ( r , FR . json ) <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( tid ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_floating_ips ( self , get ) : <EOL> tid = '<STR_LIT>' <EOL> r = self . client . get_floating_ips ( tid ) <EOL> self . assert_dicts_are_equal ( r , FR . json ) <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( tid ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_alloc_floating_ip ( self , post ) : <EOL> FR . json = dict ( floating_ip = dict ( <EOL> fixed_ip = '<STR_LIT>' , <EOL> id = <NUM_LIT:1> , <EOL> instance_id = '<STR_LIT>' , <EOL> ip = '<STR_LIT>' , <EOL> pool = '<STR_LIT>' ) ) <EOL> for args in product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( None , '<STR_LIT>' ) ) : <EOL> r = self . client . alloc_floating_ip ( * args ) <EOL> tenant_id , pool = args <EOL> self . assert_dicts_are_equal ( r , FR . json [ '<STR_LIT>' ] ) <EOL> expected = dict ( pool = pool ) if pool else dict ( ) <EOL> self . assertEqual ( post . mock_calls [ - <NUM_LIT:1> ] , call ( tenant_id , expected ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_get_floating_ip ( self , get ) : <EOL> FR . json = dict ( floating_ips = [ dict ( <EOL> fixed_ip = '<STR_LIT>' , <EOL> id = <NUM_LIT:1> , <EOL> instance_id = '<STR_LIT>' , <EOL> ip = '<STR_LIT>' , <EOL> pool = '<STR_LIT>' ) , ] ) <EOL> for args in product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( None , '<STR_LIT>' ) ) : <EOL> r = self . client . get_floating_ip ( * args ) <EOL> tenant_id , fip = args <EOL> self . assertEqual ( r , FR . json [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( get . mock_calls [ - <NUM_LIT:1> ] , call ( tenant_id , fip ) ) <EOL> @ patch ( '<STR_LIT>' % compute_pkg , return_value = FR ( ) ) <EOL> def test_delete_floating_ip ( self , delete ) : <EOL> for args in product ( <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( None , '<STR_LIT>' ) ) : <EOL> r = self . client . delete_floating_ip ( * args ) <EOL> tenant_id , fip = args <EOL> self . assertEqual ( r , FR . headers ) <EOL> self . assertEqual ( delete . mock_calls [ - <NUM_LIT:1> ] , call ( tenant_id , fip ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from sys import argv <EOL> from kamaki . clients . test import runTestCase <EOL> not_found = True <EOL> if not argv [ <NUM_LIT:1> : ] or argv [ <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> not_found = False <EOL> runTestCase ( ComputeClient , '<STR_LIT>' , argv [ <NUM_LIT:2> : ] ) <EOL> if not argv [ <NUM_LIT:1> : ] or argv [ <NUM_LIT:1> ] == '<STR_LIT>' : <EOL> not_found = False <EOL> runTestCase ( ComputeRestClient , '<STR_LIT>' , argv [ <NUM_LIT:2> : ] ) <EOL> if not_found : <EOL> print ( '<STR_LIT>' % argv [ <NUM_LIT:1> ] ) </s>
