<s> import base64 <EOL> import uuid <EOL> import mock <EOL> from mox3 import mox <EOL> from oslo_policy import policy as oslo_policy <EOL> from oslo_utils import uuidutils <EOL> import webob <EOL> from nova . api . openstack . compute import extension_info <EOL> from nova . api . openstack . compute . legacy_v2 import servers as servers_v2 <EOL> from nova . api . openstack . compute import servers as servers_v21 <EOL> from nova . compute import api as compute_api <EOL> from nova . compute import task_states <EOL> from nova . compute import vm_states <EOL> import nova . conf <EOL> from nova import exception <EOL> from nova . image import glance <EOL> from nova import objects <EOL> from nova import policy <EOL> from nova import test <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> from nova . tests . unit . image import fake <EOL> CONF = nova . conf . CONF <EOL> FAKE_UUID = fakes . FAKE_UUID <EOL> INSTANCE_IDS = { FAKE_UUID : <NUM_LIT:1> } <EOL> def return_server_not_found ( * arg , ** kwarg ) : <EOL> raise exception . InstanceNotFound ( instance_id = FAKE_UUID ) <EOL> def instance_update_and_get_original ( context , instance_uuid , values , <EOL> columns_to_join = None , <EOL> ) : <EOL> inst = fakes . stub_instance ( INSTANCE_IDS [ instance_uuid ] , host = '<STR_LIT>' ) <EOL> inst = dict ( inst , ** values ) <EOL> return ( inst , inst ) <EOL> def instance_update ( context , instance_uuid , kwargs ) : <EOL> inst = fakes . stub_instance ( INSTANCE_IDS [ instance_uuid ] , host = '<STR_LIT>' ) <EOL> return inst <EOL> class MockSetAdminPassword ( object ) : <EOL> def __init__ ( self ) : <EOL> self . instance_id = None <EOL> self . password = None <EOL> def __call__ ( self , context , instance , password ) : <EOL> self . instance_id = instance [ '<STR_LIT>' ] <EOL> self . password = password <EOL> class ServerActionsControllerTestV21 ( test . TestCase ) : <EOL> image_uuid = '<STR_LIT>' <EOL> image_base_url = '<STR_LIT>' <EOL> image_href = image_base_url + '<STR_LIT:/>' + image_uuid <EOL> servers = servers_v21 <EOL> validation_error = exception . ValidationError <EOL> request_too_large_error = exception . ValidationError <EOL> image_url = None <EOL> def setUp ( self ) : <EOL> super ( ServerActionsControllerTestV21 , self ) . setUp ( ) <EOL> self . flags ( group = '<STR_LIT>' , api_servers = [ '<STR_LIT>' ] ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE , <EOL> host = '<STR_LIT>' ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> instance_update_and_get_original ) <EOL> fakes . stub_out_nw_api ( self ) <EOL> fakes . stub_out_compute_api_snapshot ( self . stubs ) <EOL> fake . stub_out_image_service ( self ) <EOL> self . flags ( allow_instance_snapshots = True , <EOL> enable_instance_password = True ) <EOL> self . _image_href = '<STR_LIT>' <EOL> self . controller = self . _get_controller ( ) <EOL> self . compute_api = self . controller . compute_api <EOL> self . req = fakes . HTTPRequest . blank ( '<STR_LIT>' ) <EOL> self . context = self . req . environ [ '<STR_LIT>' ] <EOL> def _get_controller ( self ) : <EOL> ext_info = extension_info . LoadedExtensionInfo ( ) <EOL> return self . servers . ServersController ( extension_info = ext_info ) <EOL> def _set_fake_extension ( self ) : <EOL> pass <EOL> def _rebuild ( self , context , image_ref , value = None ) : <EOL> if value is not None : <EOL> compute_api . API . rebuild ( context , mox . IgnoreArg ( ) , image_ref , <EOL> mox . IgnoreArg ( ) , preserve_ephemeral = value ) <EOL> else : <EOL> compute_api . API . rebuild ( context , mox . IgnoreArg ( ) , image_ref , <EOL> mox . IgnoreArg ( ) ) <EOL> def _stub_instance_get ( self , uuid = None ) : <EOL> self . mox . StubOutWithMock ( compute_api . API , '<STR_LIT>' ) <EOL> if uuid is None : <EOL> uuid = uuidutils . generate_uuid ( ) <EOL> instance = fake_instance . fake_db_instance ( <EOL> id = <NUM_LIT:1> , uuid = uuid , vm_state = vm_states . ACTIVE , task_state = None ) <EOL> instance = objects . Instance . _from_db_object ( <EOL> self . context , objects . Instance ( ) , instance ) <EOL> self . compute_api . get ( self . context , uuid , <EOL> expected_attrs = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> want_objects = True ) . AndReturn ( instance ) <EOL> return instance <EOL> def _test_locked_instance ( self , action , method = None , body_map = None , <EOL> compute_api_args_map = None ) : <EOL> if body_map is None : <EOL> body_map = { } <EOL> if compute_api_args_map is None : <EOL> compute_api_args_map = { } <EOL> instance = self . _stub_instance_get ( ) <EOL> args , kwargs = compute_api_args_map . get ( action , ( ( ) , { } ) ) <EOL> getattr ( compute_api . API , method ) ( self . context , instance , <EOL> * args , ** kwargs ) . AndRaise ( <EOL> exception . InstanceIsLocked ( instance_uuid = instance [ '<STR_LIT>' ] ) ) <EOL> self . mox . ReplayAll ( ) <EOL> controller_function = '<STR_LIT>' + action <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> eval ( controller_function ) , <EOL> self . req , instance [ '<STR_LIT>' ] , <EOL> body = body_map . get ( action ) ) <EOL> self . mox . VerifyAll ( ) <EOL> self . mox . UnsetStubs ( ) <EOL> def test_actions_with_locked_instance ( self ) : <EOL> actions = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> method_translations = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> body_map = { '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:2>' } } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : { '<STR_LIT:type>' : '<STR_LIT>' } } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . image_uuid , <EOL> '<STR_LIT>' : '<STR_LIT>' } } } <EOL> args_map = { '<STR_LIT>' : ( ( '<STR_LIT:2>' ) , { } ) , <EOL> '<STR_LIT>' : ( ( ) , { } ) , <EOL> '<STR_LIT>' : ( ( '<STR_LIT>' , ) , { } ) , <EOL> '<STR_LIT>' : ( ( self . image_uuid , <EOL> '<STR_LIT>' ) , { } ) } <EOL> if self . servers == servers_v2 : <EOL> args_map [ '<STR_LIT>' ] = ( ( self . image_uuid , '<STR_LIT>' ) , <EOL> { '<STR_LIT>' : None } ) <EOL> for action in actions : <EOL> method = method_translations . get ( action ) <EOL> self . mox . StubOutWithMock ( compute_api . API , method ) <EOL> self . _test_locked_instance ( action , method = method , <EOL> body_map = body_map , <EOL> compute_api_args_map = args_map ) <EOL> def test_reboot_hard ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . controller . _action_reboot ( self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_soft ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . controller . _action_reboot ( self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_incorrect_type ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_missing_type ( self ) : <EOL> body = dict ( reboot = dict ( ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_none ( self ) : <EOL> body = dict ( reboot = dict ( type = None ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_not_found ( self ) : <EOL> self . stub_out ( '<STR_LIT>' , <EOL> return_server_not_found ) <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . controller . _action_reboot , <EOL> self . req , str ( uuid . uuid4 ( ) ) , body = body ) <EOL> def test_reboot_raises_conflict_on_invalid_state ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> def fake_reboot ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , fake_reboot ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_soft_with_soft_in_progress_raises_conflict ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE , <EOL> task_state = task_states . REBOOTING ) ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_hard_with_soft_in_progress_does_not_raise ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE , <EOL> task_state = task_states . REBOOTING ) ) <EOL> self . controller . _action_reboot ( self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_hard_with_hard_in_progress ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE , <EOL> task_state = task_states . REBOOTING_HARD ) ) <EOL> self . controller . _action_reboot ( self . req , FAKE_UUID , body = body ) <EOL> def test_reboot_soft_with_hard_in_progress_raises_conflict ( self ) : <EOL> body = dict ( reboot = dict ( type = "<STR_LIT>" ) ) <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE , <EOL> task_state = task_states . REBOOTING_HARD ) ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_reboot , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def _test_rebuild_preserve_ephemeral ( self , value = None ) : <EOL> self . _set_fake_extension ( ) <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , <EOL> host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> if value is not None : <EOL> body [ '<STR_LIT>' ] [ '<STR_LIT>' ] = value <EOL> self . mox . StubOutWithMock ( compute_api . API , '<STR_LIT>' ) <EOL> self . _rebuild ( self . context , self . _image_href , value ) <EOL> self . mox . ReplayAll ( ) <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_preserve_ephemeral_true ( self ) : <EOL> self . _test_rebuild_preserve_ephemeral ( True ) <EOL> def test_rebuild_preserve_ephemeral_false ( self ) : <EOL> self . _test_rebuild_preserve_ephemeral ( False ) <EOL> def test_rebuild_preserve_ephemeral_default ( self ) : <EOL> self . _test_rebuild_preserve_ephemeral ( ) <EOL> def test_rebuild_accepted_minimum ( self ) : <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self_href = '<STR_LIT>' % FAKE_UUID <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> robj = self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> body = robj . obj <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] , '<STR_LIT:2>' ) <EOL> self . assertEqual ( len ( body [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) , <EOL> CONF . password_length ) <EOL> self . assertEqual ( robj [ '<STR_LIT:location>' ] , self_href ) <EOL> def test_rebuild_instance_with_image_uuid ( self ) : <EOL> info = dict ( image_href_in_call = None ) <EOL> def rebuild ( self2 , context , instance , image_href , * args , ** kwargs ) : <EOL> info [ '<STR_LIT>' ] = image_href <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE ) ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , rebuild ) <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . image_uuid , <EOL> } , <EOL> } <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , self . image_uuid ) <EOL> def test_rebuild_instance_with_image_href_uses_uuid ( self ) : <EOL> info = dict ( image_href_in_call = None ) <EOL> def rebuild ( self2 , context , instance , image_href , * args , ** kwargs ) : <EOL> info [ '<STR_LIT>' ] = image_href <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fakes . fake_instance_get ( vm_state = vm_states . ACTIVE ) ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , rebuild ) <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : self . image_href , <EOL> } , <EOL> } <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> self . assertEqual ( info [ '<STR_LIT>' ] , self . image_uuid ) <EOL> def test_rebuild_accepted_minimum_pass_disabled ( self ) : <EOL> self . flags ( enable_instance_password = False ) <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> self_href = '<STR_LIT>' % FAKE_UUID <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> robj = self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> body = robj . obj <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] , '<STR_LIT:2>' ) <EOL> self . assertNotIn ( "<STR_LIT>" , body [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( robj [ '<STR_LIT:location>' ] , self_href ) <EOL> def test_rebuild_raises_conflict_on_invalid_state ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> def fake_rebuild ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , fake_rebuild ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_accepted_with_metadata ( self ) : <EOL> metadata = { '<STR_LIT>' : '<STR_LIT>' } <EOL> return_server = fakes . fake_instance_get ( metadata = metadata , <EOL> vm_state = vm_states . ACTIVE , host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : metadata , <EOL> } , <EOL> } <EOL> body = self . controller . _action_rebuild ( self . req , FAKE_UUID , <EOL> body = body ) . obj <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT>' ] , metadata ) <EOL> def test_rebuild_accepted_with_bad_metadata ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_with_too_large_metadata ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : { <EOL> <NUM_LIT> * "<STR_LIT:k>" : "<STR_LIT:value>" <EOL> } <EOL> } <EOL> } <EOL> self . assertRaises ( self . request_too_large_error , <EOL> self . controller . _action_rebuild , self . req , <EOL> FAKE_UUID , body = body ) <EOL> def test_rebuild_bad_entity ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_admin_pass ( self ) : <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> body = self . controller . _action_rebuild ( self . req , FAKE_UUID , <EOL> body = body ) . obj <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] , '<STR_LIT:2>' ) <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> def test_rebuild_admin_pass_pass_disabled ( self ) : <EOL> self . flags ( enable_instance_password = False ) <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> body = self . controller . _action_rebuild ( self . req , FAKE_UUID , <EOL> body = body ) . obj <EOL> self . assertEqual ( body [ '<STR_LIT>' ] [ '<STR_LIT:image>' ] [ '<STR_LIT:id>' ] , '<STR_LIT:2>' ) <EOL> self . assertNotIn ( '<STR_LIT>' , body [ '<STR_LIT>' ] ) <EOL> def test_rebuild_server_not_found ( self ) : <EOL> def server_not_found ( self , instance_id , <EOL> columns_to_join = None , use_slave = False ) : <EOL> raise exception . InstanceNotFound ( instance_id = instance_id ) <EOL> self . stub_out ( '<STR_LIT>' , server_not_found ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_with_bad_image ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT:foo>" , <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_accessIP ( self ) : <EOL> attributes = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> data = { '<STR_LIT>' : { } } <EOL> orig_get = compute_api . API . get <EOL> def wrap_get ( * args , ** kwargs ) : <EOL> data [ '<STR_LIT>' ] = orig_get ( * args , ** kwargs ) <EOL> return data [ '<STR_LIT>' ] <EOL> def fake_save ( context , ** kwargs ) : <EOL> data [ '<STR_LIT>' ] . update ( data [ '<STR_LIT>' ] . obj_get_changes ( ) ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , wrap_get ) <EOL> self . stubs . Set ( objects . Instance , '<STR_LIT>' , fake_save ) <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) <EOL> self . assertEqual ( self . _image_href , data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( "<STR_LIT>" , data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( task_states . REBUILDING , data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , data [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> for attr , value in attributes . items ( ) : <EOL> self . assertEqual ( value , str ( data [ '<STR_LIT>' ] [ attr ] ) ) <EOL> def test_rebuild_when_kernel_not_exists ( self ) : <EOL> def return_image_meta ( * args , ** kwargs ) : <EOL> image_meta_table = { <EOL> '<STR_LIT:2>' : { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : <EOL> { '<STR_LIT:id>' : <NUM_LIT:3> , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> } } , <EOL> } <EOL> image_id = args [ <NUM_LIT:2> ] <EOL> try : <EOL> image_meta = image_meta_table [ str ( image_id ) ] <EOL> except KeyError : <EOL> raise exception . ImageNotFound ( image_id = image_id ) <EOL> return image_meta <EOL> self . stubs . Set ( fake . _FakeImageService , '<STR_LIT>' , return_image_meta ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_proper_kernel_ram ( self ) : <EOL> instance_meta = { '<STR_LIT>' : None , '<STR_LIT>' : None } <EOL> orig_get = compute_api . API . get <EOL> def wrap_get ( * args , ** kwargs ) : <EOL> inst = orig_get ( * args , ** kwargs ) <EOL> instance_meta [ '<STR_LIT>' ] = inst <EOL> return inst <EOL> def fake_save ( context , ** kwargs ) : <EOL> instance = instance_meta [ '<STR_LIT>' ] <EOL> for key in instance_meta . keys ( ) : <EOL> if key in instance . obj_what_changed ( ) : <EOL> instance_meta [ key ] = instance [ key ] <EOL> def return_image_meta ( * args , ** kwargs ) : <EOL> image_meta_table = { <EOL> '<STR_LIT:1>' : { '<STR_LIT:id>' : <NUM_LIT:1> , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT:2>' : { '<STR_LIT:id>' : <NUM_LIT:2> , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } , <EOL> '<STR_LIT>' : <EOL> { '<STR_LIT:id>' : <NUM_LIT:3> , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> } } , <EOL> } <EOL> image_id = args [ <NUM_LIT:2> ] <EOL> try : <EOL> image_meta = image_meta_table [ str ( image_id ) ] <EOL> except KeyError : <EOL> raise exception . ImageNotFound ( image_id = image_id ) <EOL> return image_meta <EOL> self . stubs . Set ( fake . _FakeImageService , '<STR_LIT>' , return_image_meta ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , wrap_get ) <EOL> self . stubs . Set ( objects . Instance , '<STR_LIT>' , fake_save ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body = body ) . obj <EOL> self . assertEqual ( instance_meta [ '<STR_LIT>' ] , '<STR_LIT:1>' ) <EOL> self . assertEqual ( instance_meta [ '<STR_LIT>' ] , '<STR_LIT:2>' ) <EOL> @ mock . patch . object ( compute_api . API , '<STR_LIT>' ) <EOL> def test_rebuild_instance_raise_auto_disk_config_exc ( self , mock_rebuild ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> } , <EOL> } <EOL> mock_rebuild . side_effect = exception . AutoDiskConfigDisabledByImage ( <EOL> image = '<STR_LIT>' ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_server ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . resize_called = False <EOL> def resize_mock ( * args ) : <EOL> self . resize_called = True <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , resize_mock ) <EOL> self . controller . _action_resize ( self . req , FAKE_UUID , body = body ) <EOL> self . assertTrue ( self . resize_called ) <EOL> def test_resize_server_no_flavor ( self ) : <EOL> body = dict ( resize = dict ( ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_server_no_flavor_ref ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = None ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_server_with_extra_arg ( self ) : <EOL> body = dict ( resize = dict ( favorRef = "<STR_LIT>" , <EOL> extra_arg = "<STR_LIT>" ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_server_invalid_flavor_ref ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = <NUM_LIT> ) ) <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_with_server_not_found ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , return_server_not_found ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_with_image_exceptions ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . resize_called = <NUM_LIT:0> <EOL> image_id = '<STR_LIT>' <EOL> exceptions = [ <EOL> ( exception . ImageNotAuthorized ( image_id = image_id ) , <EOL> webob . exc . HTTPUnauthorized ) , <EOL> ( exception . ImageNotFound ( image_id = image_id ) , <EOL> webob . exc . HTTPBadRequest ) , <EOL> ( exception . Invalid , webob . exc . HTTPBadRequest ) , <EOL> ( exception . NoValidHost ( reason = '<STR_LIT>' ) , <EOL> webob . exc . HTTPBadRequest ) , <EOL> ( exception . AutoDiskConfigDisabledByImage ( image = image_id ) , <EOL> webob . exc . HTTPBadRequest ) , <EOL> ] <EOL> raised , expected = map ( iter , zip ( * exceptions ) ) <EOL> def _fake_resize ( obj , context , instance , flavor_id ) : <EOL> self . resize_called += <NUM_LIT:1> <EOL> raise next ( raised ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , _fake_resize ) <EOL> for call_no in range ( len ( exceptions ) ) : <EOL> next_exception = next ( expected ) <EOL> actual = self . assertRaises ( next_exception , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> if ( isinstance ( exceptions [ call_no ] [ <NUM_LIT:0> ] , <EOL> exception . NoValidHost ) ) : <EOL> self . assertEqual ( actual . explanation , <EOL> '<STR_LIT>' ) <EOL> elif ( isinstance ( exceptions [ call_no ] [ <NUM_LIT:0> ] , <EOL> exception . AutoDiskConfigDisabledByImage ) ) : <EOL> self . assertEqual ( actual . explanation , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( self . resize_called , call_no + <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> side_effect = exception . CannotResizeDisk ( reason = '<STR_LIT>' ) ) <EOL> def test_resize_raises_cannot_resize_disk ( self , mock_resize ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> side_effect = exception . FlavorNotFound ( reason = '<STR_LIT>' , <EOL> flavor_id = '<STR_LIT>' ) ) <EOL> def test_resize_raises_flavor_not_found ( self , mock_resize ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_with_too_many_instances ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> def fake_resize ( * args , ** kwargs ) : <EOL> raise exception . TooManyInstances ( message = "<STR_LIT>" ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , fake_resize ) <EOL> self . assertRaises ( webob . exc . HTTPForbidden , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_resize_raises_conflict_on_invalid_state ( self ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> def fake_resize ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , fake_resize ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> side_effect = exception . NoValidHost ( reason = '<STR_LIT>' ) ) <EOL> def test_resize_raises_no_valid_host ( self , mock_resize ) : <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> @ mock . patch . object ( compute_api . API , '<STR_LIT>' ) <EOL> def test_resize_instance_raise_auto_disk_config_exc ( self , mock_resize ) : <EOL> mock_resize . side_effect = exception . AutoDiskConfigDisabledByImage ( <EOL> image = '<STR_LIT>' ) <EOL> body = dict ( resize = dict ( flavorRef = "<STR_LIT>" ) ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_confirm_resize_server ( self ) : <EOL> body = dict ( confirmResize = None ) <EOL> self . confirm_resize_called = False <EOL> def cr_mock ( * args ) : <EOL> self . confirm_resize_called = True <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , cr_mock ) <EOL> self . controller . _action_confirm_resize ( self . req , FAKE_UUID , body = body ) <EOL> self . assertTrue ( self . confirm_resize_called ) <EOL> def test_confirm_resize_migration_not_found ( self ) : <EOL> body = dict ( confirmResize = None ) <EOL> def confirm_resize_mock ( * args ) : <EOL> raise exception . MigrationNotFoundByStatus ( instance_id = <NUM_LIT:1> , <EOL> status = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , <EOL> '<STR_LIT>' , <EOL> confirm_resize_mock ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_confirm_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_confirm_resize_raises_conflict_on_invalid_state ( self ) : <EOL> body = dict ( confirmResize = None ) <EOL> def fake_confirm_resize ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , <EOL> fake_confirm_resize ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_confirm_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_revert_resize_migration_not_found ( self ) : <EOL> body = dict ( revertResize = None ) <EOL> def revert_resize_mock ( * args ) : <EOL> raise exception . MigrationNotFoundByStatus ( instance_id = <NUM_LIT:1> , <EOL> status = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , <EOL> '<STR_LIT>' , <EOL> revert_resize_mock ) <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_revert_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_revert_resize_server_not_found ( self ) : <EOL> body = dict ( revertResize = None ) <EOL> self . assertRaises ( webob . exc . HTTPNotFound , <EOL> self . controller . _action_revert_resize , <EOL> self . req , "<STR_LIT>" , body = body ) <EOL> def test_revert_resize_server ( self ) : <EOL> body = dict ( revertResize = None ) <EOL> self . revert_resize_called = False <EOL> def revert_mock ( * args ) : <EOL> self . revert_resize_called = True <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , revert_mock ) <EOL> body = self . controller . _action_revert_resize ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> self . assertTrue ( self . revert_resize_called ) <EOL> def test_revert_resize_raises_conflict_on_invalid_state ( self ) : <EOL> body = dict ( revertResize = None ) <EOL> def fake_revert_resize ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , <EOL> fake_revert_resize ) <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_revert_resize , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> response = self . controller . _action_create_image ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> location = response . headers [ '<STR_LIT>' ] <EOL> self . assertEqual ( self . image_url + '<STR_LIT>' if self . image_url else <EOL> glance . generate_image_url ( '<STR_LIT>' ) , <EOL> location ) <EOL> def test_create_image_name_too_long ( self ) : <EOL> long_name = '<STR_LIT:a>' * <NUM_LIT> <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : long_name , <EOL> } , <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_create_image , self . req , <EOL> FAKE_UUID , body = body ) <EOL> def _do_test_create_volume_backed_image ( self , extra_properties ) : <EOL> def _fake_id ( x ) : <EOL> return '<STR_LIT>' % ( x * <NUM_LIT:8> , x * <NUM_LIT:4> , x * <NUM_LIT:4> , x * <NUM_LIT:12> ) <EOL> body = dict ( createImage = dict ( name = '<STR_LIT>' ) ) <EOL> if extra_properties : <EOL> body [ '<STR_LIT>' ] [ '<STR_LIT>' ] = extra_properties <EOL> image_service = glance . get_default_image_service ( ) <EOL> bdm = [ dict ( volume_id = _fake_id ( '<STR_LIT:a>' ) , <EOL> volume_size = <NUM_LIT:1> , <EOL> device_name = '<STR_LIT>' , <EOL> delete_on_termination = False ) ] <EOL> def fake_block_device_mapping_get_all_by_instance ( context , inst_id , <EOL> use_slave = False ) : <EOL> return [ fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT>' : _fake_id ( '<STR_LIT:a>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } ) ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fake_block_device_mapping_get_all_by_instance ) <EOL> system_metadata = dict ( image_kernel_id = _fake_id ( '<STR_LIT:b>' ) , <EOL> image_ramdisk_id = _fake_id ( '<STR_LIT:c>' ) , <EOL> image_root_device_name = '<STR_LIT>' , <EOL> image_block_device_mapping = str ( bdm ) , <EOL> image_container_format = '<STR_LIT>' ) <EOL> instance = fakes . fake_instance_get ( image_ref = str ( uuid . uuid4 ( ) ) , <EOL> vm_state = vm_states . ACTIVE , <EOL> root_device_name = '<STR_LIT>' , <EOL> system_metadata = system_metadata ) <EOL> self . stub_out ( '<STR_LIT>' , instance ) <EOL> self . mox . StubOutWithMock ( self . controller . compute_api . compute_rpcapi , <EOL> '<STR_LIT>' ) <EOL> self . controller . compute_api . compute_rpcapi . quiesce_instance ( <EOL> mox . IgnoreArg ( ) , mox . IgnoreArg ( ) ) . AndRaise ( <EOL> exception . InstanceQuiesceNotSupported ( instance_id = '<STR_LIT>' , <EOL> reason = '<STR_LIT:test>' ) ) <EOL> volume = dict ( id = _fake_id ( '<STR_LIT:a>' ) , <EOL> size = <NUM_LIT:1> , <EOL> host = '<STR_LIT>' , <EOL> display_description = '<STR_LIT>' ) <EOL> snapshot = dict ( id = _fake_id ( '<STR_LIT:d>' ) ) <EOL> self . mox . StubOutWithMock ( self . controller . compute_api , '<STR_LIT>' ) <EOL> volume_api = self . controller . compute_api . volume_api <EOL> volume_api . get ( mox . IgnoreArg ( ) , volume [ '<STR_LIT:id>' ] ) . AndReturn ( volume ) <EOL> volume_api . create_snapshot_force ( mox . IgnoreArg ( ) , volume [ '<STR_LIT:id>' ] , <EOL> mox . IgnoreArg ( ) , mox . IgnoreArg ( ) ) . AndReturn ( snapshot ) <EOL> self . mox . ReplayAll ( ) <EOL> response = self . controller . _action_create_image ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> location = response . headers [ '<STR_LIT>' ] <EOL> image_id = location . replace ( self . image_url or <EOL> glance . generate_image_url ( '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> image = image_service . show ( None , image_id ) <EOL> self . assertEqual ( image [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> properties = image [ '<STR_LIT>' ] <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , _fake_id ( '<STR_LIT:b>' ) ) <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , _fake_id ( '<STR_LIT:c>' ) ) <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertTrue ( properties [ '<STR_LIT>' ] ) <EOL> bdms = properties [ '<STR_LIT>' ] <EOL> self . assertEqual ( len ( bdms ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( bdms [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <NUM_LIT:0> ) <EOL> self . assertEqual ( bdms [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( bdms [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( bdms [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , snapshot [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , bdms [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> for fld in ( '<STR_LIT>' , '<STR_LIT:id>' , '<STR_LIT>' ) : <EOL> self . assertNotIn ( fld , bdms [ <NUM_LIT:0> ] ) <EOL> for k in extra_properties . keys ( ) : <EOL> self . assertEqual ( properties [ k ] , extra_properties [ k ] ) <EOL> def test_create_volume_backed_image_no_metadata ( self ) : <EOL> self . _do_test_create_volume_backed_image ( { } ) <EOL> def test_create_volume_backed_image_with_metadata ( self ) : <EOL> self . _do_test_create_volume_backed_image ( dict ( ImageType = '<STR_LIT>' , <EOL> ImageVersion = '<STR_LIT>' ) ) <EOL> def _test_create_volume_backed_image_with_metadata_from_volume ( <EOL> self , extra_metadata = None ) : <EOL> def _fake_id ( x ) : <EOL> return '<STR_LIT>' % ( x * <NUM_LIT:8> , x * <NUM_LIT:4> , x * <NUM_LIT:4> , x * <NUM_LIT:12> ) <EOL> body = dict ( createImage = dict ( name = '<STR_LIT>' ) ) <EOL> if extra_metadata : <EOL> body [ '<STR_LIT>' ] [ '<STR_LIT>' ] = extra_metadata <EOL> image_service = glance . get_default_image_service ( ) <EOL> def fake_block_device_mapping_get_all_by_instance ( context , inst_id , <EOL> use_slave = False ) : <EOL> return [ fake_block_device . FakeDbBlockDeviceDict ( <EOL> { '<STR_LIT>' : _fake_id ( '<STR_LIT:a>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None } ) ] <EOL> self . stub_out ( '<STR_LIT>' , <EOL> fake_block_device_mapping_get_all_by_instance ) <EOL> instance = fakes . fake_instance_get ( <EOL> image_ref = '<STR_LIT>' , <EOL> vm_state = vm_states . ACTIVE , <EOL> root_device_name = '<STR_LIT>' , <EOL> system_metadata = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . stub_out ( '<STR_LIT>' , instance ) <EOL> self . mox . StubOutWithMock ( self . controller . compute_api . compute_rpcapi , <EOL> '<STR_LIT>' ) <EOL> self . controller . compute_api . compute_rpcapi . quiesce_instance ( <EOL> mox . IgnoreArg ( ) , mox . IgnoreArg ( ) ) . AndRaise ( <EOL> exception . InstanceQuiesceNotSupported ( instance_id = '<STR_LIT>' , <EOL> reason = '<STR_LIT:test>' ) ) <EOL> volume = dict ( id = _fake_id ( '<STR_LIT:a>' ) , <EOL> size = <NUM_LIT:1> , <EOL> host = '<STR_LIT>' , <EOL> display_description = '<STR_LIT>' ) <EOL> snapshot = dict ( id = _fake_id ( '<STR_LIT:d>' ) ) <EOL> self . mox . StubOutWithMock ( self . controller . compute_api , '<STR_LIT>' ) <EOL> volume_api = self . controller . compute_api . volume_api <EOL> volume_api . get ( mox . IgnoreArg ( ) , volume [ '<STR_LIT:id>' ] ) . AndReturn ( volume ) <EOL> volume_api . create_snapshot_force ( mox . IgnoreArg ( ) , volume [ '<STR_LIT:id>' ] , <EOL> mox . IgnoreArg ( ) , mox . IgnoreArg ( ) ) . AndReturn ( snapshot ) <EOL> self . mox . ReplayAll ( ) <EOL> response = self . controller . _action_create_image ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> location = response . headers [ '<STR_LIT>' ] <EOL> image_id = location . replace ( self . image_base_url , '<STR_LIT>' ) <EOL> image = image_service . show ( None , image_id ) <EOL> properties = image [ '<STR_LIT>' ] <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertEqual ( properties [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> if extra_metadata : <EOL> for key , val in extra_metadata . items ( ) : <EOL> self . assertEqual ( properties [ key ] , val ) <EOL> def test_create_vol_backed_img_with_meta_from_vol_without_extra_meta ( self ) : <EOL> self . _test_create_volume_backed_image_with_metadata_from_volume ( ) <EOL> def test_create_vol_backed_img_with_meta_from_vol_with_extra_meta ( self ) : <EOL> self . _test_create_volume_backed_image_with_metadata_from_volume ( <EOL> extra_metadata = { '<STR_LIT:a>' : '<STR_LIT:b>' } ) <EOL> def test_create_image_snapshots_disabled ( self ) : <EOL> """<STR_LIT>""" <EOL> self . flags ( allow_instance_snapshots = False ) <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image_with_metadata ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { '<STR_LIT:key>' : '<STR_LIT>' } , <EOL> } , <EOL> } <EOL> response = self . controller . _action_create_image ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> location = response . headers [ '<STR_LIT>' ] <EOL> self . assertEqual ( self . image_url + '<STR_LIT>' if self . image_url else <EOL> glance . generate_image_url ( '<STR_LIT>' ) , location ) <EOL> def test_create_image_with_too_much_metadata ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { } , <EOL> } , <EOL> } <EOL> for num in range ( CONF . quota_metadata_items + <NUM_LIT:1> ) : <EOL> body [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ '<STR_LIT>' % num ] = "<STR_LIT:bar>" <EOL> self . assertRaises ( webob . exc . HTTPForbidden , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image_no_name ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { } , <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image_blank_name ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image_bad_metadata ( self ) : <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> self . assertRaises ( self . validation_error , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_create_image_raises_conflict_on_invalid_state ( self ) : <EOL> def snapshot ( * args , ** kwargs ) : <EOL> raise exception . InstanceInvalidState ( attr = '<STR_LIT>' , <EOL> state = '<STR_LIT>' , method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' ) <EOL> self . stubs . Set ( compute_api . API , '<STR_LIT>' , snapshot ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPConflict , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> class ServerActionsControllerTestV2 ( ServerActionsControllerTestV21 ) : <EOL> image_base_url = '<STR_LIT>' <EOL> servers = servers_v2 <EOL> validation_error = webob . exc . HTTPBadRequest <EOL> request_too_large_error = webob . exc . HTTPRequestEntityTooLarge <EOL> image_url = '<STR_LIT>' <EOL> def _get_controller ( self ) : <EOL> class FakeExtManager ( object ) : <EOL> def is_loaded ( self , ext ) : <EOL> return False <EOL> return self . servers . Controller ( ext_mgr = FakeExtManager ( ) ) <EOL> def _set_fake_extension ( self ) : <EOL> def fake_is_loaded ( ext ) : <EOL> return ext == '<STR_LIT>' <EOL> self . stubs . Set ( self . controller . ext_mgr , '<STR_LIT>' , fake_is_loaded ) <EOL> def _rebuild ( self , context , image_ref , value = None ) : <EOL> if value is not None : <EOL> compute_api . API . rebuild ( context , mox . IgnoreArg ( ) , image_ref , <EOL> mox . IgnoreArg ( ) , preserve_ephemeral = value , <EOL> files_to_inject = None ) <EOL> else : <EOL> compute_api . API . rebuild ( context , mox . IgnoreArg ( ) , image_ref , <EOL> mox . IgnoreArg ( ) , files_to_inject = None ) <EOL> def _stub_instance_get ( self , uuid = None ) : <EOL> self . mox . StubOutWithMock ( compute_api . API , '<STR_LIT>' ) <EOL> if uuid is None : <EOL> uuid = uuidutils . generate_uuid ( ) <EOL> instance = fake_instance . fake_db_instance ( <EOL> id = <NUM_LIT:1> , uuid = uuid , vm_state = vm_states . ACTIVE , task_state = None ) <EOL> instance = objects . Instance . _from_db_object ( <EOL> self . context , objects . Instance ( ) , instance ) <EOL> self . compute_api . get ( self . context , uuid , <EOL> expected_attrs = [ '<STR_LIT>' ] , <EOL> want_objects = True ) . AndReturn ( instance ) <EOL> return instance <EOL> def test_rebuild_bad_personality ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT:path>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } ] <EOL> } , <EOL> } <EOL> self . assertRaises ( webob . exc . HTTPBadRequest , <EOL> self . controller . _action_rebuild , <EOL> self . req , FAKE_UUID , body = body ) <EOL> def test_rebuild_personality ( self ) : <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : [ { <EOL> "<STR_LIT:path>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : base64 . b64encode ( "<STR_LIT>" ) , <EOL> } ] <EOL> } , <EOL> } <EOL> body = self . controller . _action_rebuild ( self . req , FAKE_UUID , <EOL> body = body ) . obj <EOL> self . assertNotIn ( '<STR_LIT>' , body [ '<STR_LIT>' ] ) <EOL> def test_resize_server_with_extra_arg ( self ) : <EOL> pass <EOL> def test_resize_server_invalid_flavor_ref ( self ) : <EOL> pass <EOL> def test_create_image_glance_link_prefix ( self ) : <EOL> self . flags ( osapi_glance_link_prefix = '<STR_LIT>' ) <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> response = self . controller . _action_create_image ( self . req , FAKE_UUID , <EOL> body = body ) <EOL> location = response . headers [ '<STR_LIT>' ] <EOL> self . assertEqual ( '<STR_LIT>' , location ) <EOL> def test_rebuild_preserve_ephemeral_is_ignored_when_ext_not_loaded ( self ) : <EOL> return_server = fakes . fake_instance_get ( image_ref = '<STR_LIT:2>' , <EOL> vm_state = vm_states . ACTIVE , <EOL> host = '<STR_LIT>' ) <EOL> self . stub_out ( '<STR_LIT>' , return_server ) <EOL> body = { <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : self . _image_href , <EOL> "<STR_LIT>" : False , <EOL> } , <EOL> } <EOL> self . mox . StubOutWithMock ( compute_api . API , '<STR_LIT>' ) <EOL> compute_api . API . rebuild ( self . context , mox . IgnoreArg ( ) , <EOL> self . _image_href , <EOL> mox . IgnoreArg ( ) , files_to_inject = None ) <EOL> self . mox . ReplayAll ( ) <EOL> self . controller . _action_rebuild ( self . req , FAKE_UUID , body ) <EOL> def test_create_vol_backed_img_snapshotting_policy_blocks_project ( self ) : <EOL> """<STR_LIT>""" <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> rule_name = "<STR_LIT>" <EOL> rules = { <EOL> rule_name : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> policy . set_rules ( oslo_policy . Rules . from_dict ( rules ) ) <EOL> with mock . patch . object ( compute_api . API , '<STR_LIT>' , <EOL> return_value = True ) : <EOL> exc = self . assertRaises ( exception . PolicyNotAuthorized , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % rule_name , <EOL> exc . format_message ( ) ) <EOL> def test_create_vol_backed_img_snapshotting_policy_blocks_role ( self ) : <EOL> """<STR_LIT>""" <EOL> body = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> } , <EOL> } <EOL> rule_name = "<STR_LIT>" <EOL> rules = { <EOL> rule_name : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" <EOL> } <EOL> policy . set_rules ( oslo_policy . Rules . from_dict ( rules ) ) <EOL> with mock . patch . object ( compute_api . API , '<STR_LIT>' , <EOL> return_value = True ) : <EOL> exc = self . assertRaises ( exception . PolicyNotAuthorized , <EOL> self . controller . _action_create_image , <EOL> self . req , FAKE_UUID , body = body ) <EOL> self . assertEqual ( <EOL> "<STR_LIT>" % rule_name , <EOL> exc . format_message ( ) ) </s>
