<s> from oslo_config import cfg <EOL> import datetime <EOL> from nova import context <EOL> from nova import objects <EOL> from nova . tests . functional . api_sample_tests import api_sample_base <EOL> from nova . tests . functional . api_sample_tests import test_servers <EOL> from nova . tests . unit . api . openstack import fakes <EOL> from nova . tests . unit import fake_block_device <EOL> from nova . tests . unit import fake_instance <EOL> CONF = cfg . CONF <EOL> CONF . import_opt ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> class SnapshotsSampleJsonTests ( api_sample_base . ApiSampleTestBaseV21 ) : <EOL> extension_name = "<STR_LIT>" <EOL> create_subs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> def _get_flags ( self ) : <EOL> f = super ( SnapshotsSampleJsonTests , self ) . _get_flags ( ) <EOL> f [ '<STR_LIT>' ] = CONF . osapi_compute_extension [ : ] <EOL> f [ '<STR_LIT>' ] . append ( <EOL> '<STR_LIT>' ) <EOL> return f <EOL> def setUp ( self ) : <EOL> super ( SnapshotsSampleJsonTests , self ) . setUp ( ) <EOL> self . stub_out ( "<STR_LIT>" , <EOL> fakes . stub_snapshot_get_all ) <EOL> self . stub_out ( "<STR_LIT>" , <EOL> fakes . stub_snapshot_get ) <EOL> def _create_snapshot ( self ) : <EOL> self . stub_out ( "<STR_LIT>" , <EOL> fakes . stub_snapshot_create ) <EOL> response = self . _do_post ( "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> self . create_subs ) <EOL> return response <EOL> def test_snapshots_create ( self ) : <EOL> response = self . _create_snapshot ( ) <EOL> self . _verify_response ( "<STR_LIT>" , <EOL> self . create_subs , response , <NUM_LIT:200> ) <EOL> def test_snapshots_delete ( self ) : <EOL> self . stub_out ( "<STR_LIT>" , <EOL> fakes . stub_snapshot_delete ) <EOL> self . _create_snapshot ( ) <EOL> response = self . _do_delete ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> self . assertEqual ( '<STR_LIT>' , response . content ) <EOL> def test_snapshots_detail ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . _verify_response ( '<STR_LIT>' , { } , response , <NUM_LIT:200> ) <EOL> def test_snapshots_list ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . _verify_response ( '<STR_LIT>' , { } , response , <NUM_LIT:200> ) <EOL> def test_snapshots_show ( self ) : <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> subs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:description>' : '<STR_LIT>' <EOL> } <EOL> self . _verify_response ( '<STR_LIT>' , subs , response , <NUM_LIT:200> ) <EOL> class VolumesSampleJsonTest ( test_servers . ServersSampleBase ) : <EOL> extension_name = "<STR_LIT>" <EOL> def _get_flags ( self ) : <EOL> f = super ( VolumesSampleJsonTest , self ) . _get_flags ( ) <EOL> f [ '<STR_LIT>' ] = CONF . osapi_compute_extension [ : ] <EOL> f [ '<STR_LIT>' ] . append ( <EOL> '<STR_LIT>' ) <EOL> return f <EOL> def _get_volume_id ( self ) : <EOL> return '<STR_LIT>' <EOL> def _stub_volume ( self , id , displayname = "<STR_LIT>" , <EOL> displaydesc = "<STR_LIT>" , size = <NUM_LIT:100> ) : <EOL> volume = { <EOL> '<STR_LIT:id>' : id , <EOL> '<STR_LIT:size>' : size , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : displayname , <EOL> '<STR_LIT>' : displaydesc , <EOL> '<STR_LIT>' : datetime . datetime ( <NUM_LIT> , <NUM_LIT:12> , <NUM_LIT:1> , <NUM_LIT:11> , <NUM_LIT:1> , <NUM_LIT> ) , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : { '<STR_LIT:name>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <EOL> { '<STR_LIT>' : '<STR_LIT:/>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' <EOL> } <EOL> } <EOL> } <EOL> return volume <EOL> def _stub_volume_get ( self , context , volume_id ) : <EOL> return self . _stub_volume ( volume_id ) <EOL> def _stub_volume_delete ( self , context , * args , ** param ) : <EOL> pass <EOL> def _stub_volume_get_all ( self , context , search_opts = None ) : <EOL> id = self . _get_volume_id ( ) <EOL> return [ self . _stub_volume ( id ) ] <EOL> def _stub_volume_create ( self , context , size , name , description , snapshot , <EOL> ** param ) : <EOL> id = self . _get_volume_id ( ) <EOL> return self . _stub_volume ( id ) <EOL> def setUp ( self ) : <EOL> super ( VolumesSampleJsonTest , self ) . setUp ( ) <EOL> fakes . stub_out_networking ( self ) <EOL> fakes . stub_out_rate_limiting ( self . stubs ) <EOL> self . stub_out ( "<STR_LIT>" , <EOL> self . _stub_volume_delete ) <EOL> self . stub_out ( "<STR_LIT>" , self . _stub_volume_get ) <EOL> self . stub_out ( "<STR_LIT>" , <EOL> self . _stub_volume_get_all ) <EOL> def _post_volume ( self ) : <EOL> subs_req = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } <EOL> self . stub_out ( "<STR_LIT>" , <EOL> self . _stub_volume_create ) <EOL> response = self . _do_post ( '<STR_LIT>' , '<STR_LIT>' , <EOL> subs_req ) <EOL> self . _verify_response ( '<STR_LIT>' , subs_req , response , <NUM_LIT:200> ) <EOL> def test_volumes_show ( self ) : <EOL> subs = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } <EOL> vol_id = self . _get_volume_id ( ) <EOL> response = self . _do_get ( '<STR_LIT>' % vol_id ) <EOL> self . _verify_response ( '<STR_LIT>' , subs , response , <NUM_LIT:200> ) <EOL> def test_volumes_index ( self ) : <EOL> subs = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . _verify_response ( '<STR_LIT>' , subs , response , <NUM_LIT:200> ) <EOL> def test_volumes_detail ( self ) : <EOL> subs = { <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> } <EOL> response = self . _do_get ( '<STR_LIT>' ) <EOL> self . _verify_response ( '<STR_LIT>' , subs , response , <NUM_LIT:200> ) <EOL> def test_volumes_create ( self ) : <EOL> self . _post_volume ( ) <EOL> def test_volumes_delete ( self ) : <EOL> self . _post_volume ( ) <EOL> vol_id = self . _get_volume_id ( ) <EOL> response = self . _do_delete ( '<STR_LIT>' % vol_id ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> self . assertEqual ( '<STR_LIT>' , response . content ) <EOL> class VolumeAttachmentsSample ( test_servers . ServersSampleBase ) : <EOL> extension_name = "<STR_LIT>" <EOL> def _stub_db_bdms_get_all_by_instance ( self , server_id ) : <EOL> def fake_bdms_get_all_by_instance ( context , instance_uuid , <EOL> use_slave = False ) : <EOL> bdms = [ <EOL> fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : server_id , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) , <EOL> fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : server_id , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> ] <EOL> return bdms <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fake_bdms_get_all_by_instance ) <EOL> def _stub_compute_api_get ( self ) : <EOL> def fake_compute_api_get ( self , context , instance_id , <EOL> want_objects = False , expected_attrs = None ) : <EOL> if want_objects : <EOL> return fake_instance . fake_instance_obj ( <EOL> context , ** { '<STR_LIT>' : instance_id } ) <EOL> else : <EOL> return { '<STR_LIT>' : instance_id } <EOL> self . stub_out ( '<STR_LIT>' , fake_compute_api_get ) <EOL> def _get_flags ( self ) : <EOL> f = super ( VolumeAttachmentsSample , self ) . _get_flags ( ) <EOL> f [ '<STR_LIT>' ] = CONF . osapi_compute_extension [ : ] <EOL> f [ '<STR_LIT>' ] . append ( <EOL> '<STR_LIT>' ) <EOL> f [ '<STR_LIT>' ] . append ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return f <EOL> def test_attach_volume_to_server ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , fakes . stub_volume_get ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> lambda * a , ** k : None ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> lambda * a , ** k : None ) <EOL> device_name = '<STR_LIT>' <EOL> bdm = objects . BlockDeviceMapping ( ) <EOL> bdm [ '<STR_LIT>' ] = device_name <EOL> self . stub_out ( <EOL> '<STR_LIT>' , <EOL> lambda * a , ** k : bdm ) <EOL> self . stub_out ( <EOL> '<STR_LIT>' , <EOL> lambda * a , ** k : None ) <EOL> self . stub_out ( <EOL> '<STR_LIT>' , <EOL> classmethod ( lambda * a , ** k : None ) ) <EOL> volume = fakes . stub_volume_get ( None , context . get_admin_context ( ) , <EOL> '<STR_LIT>' ) <EOL> subs = { <EOL> '<STR_LIT>' : volume [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : device_name <EOL> } <EOL> server_id = self . _post_server ( ) <EOL> response = self . _do_post ( '<STR_LIT>' <EOL> % server_id , <EOL> '<STR_LIT>' , subs ) <EOL> self . _verify_response ( '<STR_LIT>' , subs , <EOL> response , <NUM_LIT:200> ) <EOL> def test_list_volume_attachments ( self ) : <EOL> server_id = self . _post_server ( ) <EOL> self . _stub_db_bdms_get_all_by_instance ( server_id ) <EOL> response = self . _do_get ( '<STR_LIT>' <EOL> % server_id ) <EOL> self . _verify_response ( '<STR_LIT>' , { } , <EOL> response , <NUM_LIT:200> ) <EOL> def test_volume_attachment_detail ( self ) : <EOL> server_id = self . _post_server ( ) <EOL> attach_id = "<STR_LIT>" <EOL> self . _stub_db_bdms_get_all_by_instance ( server_id ) <EOL> self . _stub_compute_api_get ( ) <EOL> response = self . _do_get ( '<STR_LIT>' <EOL> % ( server_id , attach_id ) ) <EOL> self . _verify_response ( '<STR_LIT>' , { } , <EOL> response , <NUM_LIT:200> ) <EOL> def test_volume_attachment_delete ( self ) : <EOL> server_id = self . _post_server ( ) <EOL> attach_id = "<STR_LIT>" <EOL> self . _stub_db_bdms_get_all_by_instance ( server_id ) <EOL> self . _stub_compute_api_get ( ) <EOL> self . stub_out ( '<STR_LIT>' , fakes . stub_volume_get ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> lambda * a , ** k : None ) <EOL> response = self . _do_delete ( '<STR_LIT>' <EOL> % ( server_id , attach_id ) ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> self . assertEqual ( '<STR_LIT>' , response . content ) <EOL> def test_volume_attachment_update ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , fakes . stub_volume_get ) <EOL> subs = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> server_id = self . _post_server ( ) <EOL> attach_id = '<STR_LIT>' <EOL> self . _stub_db_bdms_get_all_by_instance ( server_id ) <EOL> self . _stub_compute_api_get ( ) <EOL> self . stub_out ( '<STR_LIT>' , fakes . stub_volume_get ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> lambda * a , ** k : None ) <EOL> response = self . _do_put ( '<STR_LIT>' <EOL> % ( server_id , attach_id ) , <EOL> '<STR_LIT>' , <EOL> subs ) <EOL> self . assertEqual ( <NUM_LIT> , response . status_code ) <EOL> self . assertEqual ( '<STR_LIT>' , response . content ) </s>
