<s> import contextlib <EOL> import mock <EOL> import unittest <EOL> from cloudify import mocks as cfy_mocks <EOL> from cloudify import exceptions as cfy_exc <EOL> from cloudify . state import current_ctx <EOL> from cinder_plugin import volume <EOL> from nova_plugin import server <EOL> from openstack_plugin_common import ( CinderClient , <EOL> NovaClient , <EOL> OPENSTACK_ID_PROPERTY , <EOL> OPENSTACK_TYPE_PROPERTY , <EOL> OPENSTACK_NAME_PROPERTY ) <EOL> class TestCinderVolume ( unittest . TestCase ) : <EOL> def _mock ( self , ** kwargs ) : <EOL> ctx = cfy_mocks . MockCloudifyContext ( ** kwargs ) <EOL> current_ctx . set ( ctx ) <EOL> return ctx <EOL> def tearDown ( self ) : <EOL> current_ctx . clear ( ) <EOL> def test_create_new ( self ) : <EOL> volume_name = '<STR_LIT>' <EOL> volume_description = '<STR_LIT>' <EOL> volume_id = '<STR_LIT>' <EOL> volume_size = <NUM_LIT:10> <EOL> volume_properties = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:size>' : volume_size , <EOL> '<STR_LIT:description>' : volume_description <EOL> } , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : volume_name , <EOL> } <EOL> creating_volume_m = mock . Mock ( ) <EOL> creating_volume_m . id = volume_id <EOL> creating_volume_m . status = volume . VOLUME_STATUS_CREATING <EOL> available_volume_m = mock . Mock ( ) <EOL> available_volume_m . id = volume_id <EOL> available_volume_m . status = volume . VOLUME_STATUS_AVAILABLE <EOL> cinder_client_m = mock . Mock ( ) <EOL> cinder_client_m . volumes = mock . Mock ( ) <EOL> cinder_client_m . volumes . create = mock . Mock ( <EOL> return_value = creating_volume_m ) <EOL> cinder_client_m . volumes . get = mock . Mock ( <EOL> return_value = available_volume_m ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , properties = volume_properties ) <EOL> volume . create ( cinder_client = cinder_client_m , args = { } , ctx = ctx_m ) <EOL> cinder_client_m . volumes . create . assert_called_once_with ( <EOL> size = volume_size , <EOL> display_name = volume_name , <EOL> description = volume_description ) <EOL> cinder_client_m . volumes . get . assert_called_once_with ( volume_id ) <EOL> self . assertEqual ( <EOL> volume_id , <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_ID_PROPERTY ] ) <EOL> self . assertEqual ( <EOL> volume . VOLUME_OPENSTACK_TYPE , <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_TYPE_PROPERTY ] ) <EOL> def test_create_use_existing ( self ) : <EOL> volume_id = '<STR_LIT>' <EOL> volume_properties = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : volume_id , <EOL> } <EOL> existing_volume_m = mock . Mock ( ) <EOL> existing_volume_m . id = volume_id <EOL> existing_volume_m . status = volume . VOLUME_STATUS_AVAILABLE <EOL> cinder_client_m = mock . Mock ( ) <EOL> cinder_client_m . volumes = mock . Mock ( ) <EOL> cinder_client_m . volumes . create = mock . Mock ( ) <EOL> cinder_client_m . cosmo_get_if_exists = mock . Mock ( <EOL> return_value = existing_volume_m ) <EOL> cinder_client_m . get_id_from_resource = mock . Mock ( <EOL> return_value = volume_id ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , properties = volume_properties ) <EOL> volume . create ( cinder_client = cinder_client_m , args = { } , ctx = ctx_m ) <EOL> self . assertFalse ( cinder_client_m . volumes . create . called ) <EOL> self . assertEqual ( <EOL> volume_id , <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_ID_PROPERTY ] ) <EOL> self . assertEqual ( <EOL> volume . VOLUME_OPENSTACK_TYPE , <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_TYPE_PROPERTY ] ) <EOL> def test_delete ( self ) : <EOL> volume_id = '<STR_LIT>' <EOL> volume_name = '<STR_LIT>' <EOL> volume_properties = { <EOL> '<STR_LIT>' : False , <EOL> } <EOL> cinder_client_m = mock . Mock ( ) <EOL> cinder_client_m . cosmo_delete_resource = mock . Mock ( ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , properties = volume_properties ) <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_ID_PROPERTY ] = volume_id <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_TYPE_PROPERTY ] = volume . VOLUME_OPENSTACK_TYPE <EOL> ctx_m . instance . runtime_properties [ OPENSTACK_NAME_PROPERTY ] = volume_name <EOL> volume . delete ( cinder_client = cinder_client_m , ctx = ctx_m ) <EOL> cinder_client_m . cosmo_delete_resource . assert_called_once_with ( <EOL> volume . VOLUME_OPENSTACK_TYPE , volume_id ) <EOL> self . assertTrue ( <EOL> OPENSTACK_ID_PROPERTY not in ctx_m . instance . runtime_properties ) <EOL> self . assertTrue ( OPENSTACK_TYPE_PROPERTY <EOL> not in ctx_m . instance . runtime_properties ) <EOL> self . assertTrue ( OPENSTACK_NAME_PROPERTY <EOL> not in ctx_m . instance . runtime_properties ) <EOL> def test_attach ( self ) : <EOL> volume_id = '<STR_LIT>' <EOL> server_id = '<STR_LIT>' <EOL> device_name = '<STR_LIT>' <EOL> volume_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { volume . DEVICE_NAME_PROPERTY : device_name } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> OPENSTACK_ID_PROPERTY : volume_id , <EOL> } <EOL> } ) <EOL> } ) <EOL> server_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> server . OPENSTACK_ID_PROPERTY : server_id <EOL> } <EOL> } ) <EOL> } ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , <EOL> target = server_ctx , <EOL> source = volume_ctx ) <EOL> cinderclient_m = mock . Mock ( ) <EOL> novaclient_m = mock . Mock ( ) <EOL> novaclient_m . volumes = mock . Mock ( ) <EOL> novaclient_m . volumes . create_server_volume = mock . Mock ( ) <EOL> with contextlib . nested ( <EOL> mock . patch . object ( NovaClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = novaclient_m ) ) , <EOL> mock . patch . object ( CinderClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = cinderclient_m ) ) , <EOL> mock . patch . object ( volume , <EOL> '<STR_LIT>' , <EOL> mock . Mock ( return_value = ( None , True ) ) ) ) : <EOL> server . attach_volume ( ctx = ctx_m ) <EOL> novaclient_m . volumes . create_server_volume . assert_called_once_with ( <EOL> server_id , volume_id , device_name ) <EOL> volume . wait_until_status . assert_called_once_with ( <EOL> cinder_client = cinderclient_m , <EOL> volume_id = volume_id , <EOL> status = volume . VOLUME_STATUS_IN_USE ) <EOL> def _test_cleanup__after_attach_fails ( <EOL> self , volume_ctx_mgr , expected_err_cls , expect_cleanup = True ) : <EOL> volume_id = '<STR_LIT>' <EOL> server_id = '<STR_LIT>' <EOL> attachment_id = '<STR_LIT>' <EOL> device_name = '<STR_LIT>' <EOL> attachment = { '<STR_LIT:id>' : attachment_id , <EOL> '<STR_LIT>' : server_id , <EOL> '<STR_LIT>' : volume_id } <EOL> volume_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { volume . DEVICE_NAME_PROPERTY : device_name } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> OPENSTACK_ID_PROPERTY : volume_id , <EOL> } <EOL> } ) <EOL> } ) <EOL> server_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> server . OPENSTACK_ID_PROPERTY : server_id <EOL> } <EOL> } ) <EOL> } ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , <EOL> target = server_ctx , <EOL> source = volume_ctx ) <EOL> attached_volume_m = mock . Mock ( ) <EOL> attached_volume_m . id = volume_id <EOL> attached_volume_m . status = volume . VOLUME_STATUS_IN_USE <EOL> attached_volume_m . attachments = [ attachment ] <EOL> cinderclient_m = mock . Mock ( ) <EOL> cinderclient_m . volumes = mock . Mock ( ) <EOL> cinderclient_m . volumes . get = mock . Mock ( <EOL> return_value = attached_volume_m ) <EOL> novaclient_m = mock . Mock ( ) <EOL> novacl_vols_m = novaclient_m . volumes = mock . Mock ( ) <EOL> novacl_vols_m . create_server_volume = mock . Mock ( ) <EOL> with contextlib . nested ( <EOL> mock . patch . object ( NovaClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = novaclient_m ) ) , <EOL> mock . patch . object ( CinderClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = cinderclient_m ) ) , <EOL> volume_ctx_mgr ) : <EOL> with self . assertRaises ( expected_err_cls ) : <EOL> server . attach_volume ( ctx = ctx_m ) <EOL> novacl_vols_m . create_server_volume . assert_called_once_with ( <EOL> server_id , volume_id , device_name ) <EOL> volume . wait_until_status . assert_any_call ( <EOL> cinder_client = cinderclient_m , <EOL> volume_id = volume_id , <EOL> status = volume . VOLUME_STATUS_IN_USE ) <EOL> if expect_cleanup : <EOL> novacl_vols_m . delete_server_volume . assert_called_once_with ( <EOL> server_id , attachment_id ) <EOL> self . assertEqual ( <NUM_LIT:2> , volume . wait_until_status . call_count ) <EOL> volume . wait_until_status . assert_called_with ( <EOL> cinder_client = cinderclient_m , <EOL> volume_id = volume_id , <EOL> status = volume . VOLUME_STATUS_AVAILABLE ) <EOL> def _test_cleanup_after_waituntilstatus_throws ( self , err , expect_cleanup ) : <EOL> def raise_once ( * args , ** kwargs ) : <EOL> if raise_once . first_call : <EOL> raise_once . first_call = False <EOL> raise err <EOL> else : <EOL> return None , True <EOL> raise_once . first_call = True <EOL> self . _test_cleanup__after_attach_fails ( <EOL> volume_ctx_mgr = mock . patch . object ( <EOL> volume , <EOL> '<STR_LIT>' , <EOL> mock . Mock ( side_effect = raise_once ) <EOL> ) , <EOL> expected_err_cls = type ( err ) , <EOL> expect_cleanup = expect_cleanup <EOL> ) <EOL> def test_cleanup_after_waituntilstatus_throws_recoverable_error ( self ) : <EOL> err = cfy_exc . RecoverableError ( '<STR_LIT>' ) <EOL> self . _test_cleanup_after_waituntilstatus_throws ( err , True ) <EOL> def test_cleanup_after_waituntilstatus_throws_any_not_nonrecov_error ( self ) : <EOL> class ArbitraryNonRecoverableException ( Exception ) : <EOL> pass <EOL> err = ArbitraryNonRecoverableException ( '<STR_LIT>' ) <EOL> self . _test_cleanup_after_waituntilstatus_throws ( err , True ) <EOL> def test_cleanup_after_waituntilstatus_lets_nonrecov_errors_pass ( self ) : <EOL> err = cfy_exc . NonRecoverableError ( '<STR_LIT>' ) <EOL> self . _test_cleanup_after_waituntilstatus_throws ( err , False ) <EOL> def test_cleanup_after_waituntilstatus_times_out ( self ) : <EOL> self . _test_cleanup__after_attach_fails ( <EOL> volume_ctx_mgr = mock . patch . object ( <EOL> volume , <EOL> '<STR_LIT>' , <EOL> mock . Mock ( return_value = ( None , False ) ) <EOL> ) , <EOL> expected_err_cls = cfy_exc . RecoverableError <EOL> ) <EOL> def test_detach ( self ) : <EOL> volume_id = '<STR_LIT>' <EOL> server_id = '<STR_LIT>' <EOL> attachment_id = '<STR_LIT>' <EOL> attachment = { '<STR_LIT:id>' : attachment_id , <EOL> '<STR_LIT>' : server_id , <EOL> '<STR_LIT>' : volume_id } <EOL> volume_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> OPENSTACK_ID_PROPERTY : volume_id , <EOL> } <EOL> } ) <EOL> } ) <EOL> server_ctx = cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { } <EOL> } ) , <EOL> '<STR_LIT>' : cfy_mocks . MockContext ( { <EOL> '<STR_LIT>' : { <EOL> server . OPENSTACK_ID_PROPERTY : server_id <EOL> } <EOL> } ) <EOL> } ) <EOL> ctx_m = self . _mock ( node_id = '<STR_LIT:a>' , <EOL> target = server_ctx , <EOL> source = volume_ctx ) <EOL> attached_volume_m = mock . Mock ( ) <EOL> attached_volume_m . id = volume_id <EOL> attached_volume_m . status = volume . VOLUME_STATUS_IN_USE <EOL> attached_volume_m . attachments = [ attachment ] <EOL> cinder_client_m = mock . Mock ( ) <EOL> cinder_client_m . volumes = mock . Mock ( ) <EOL> cinder_client_m . volumes . get = mock . Mock ( <EOL> return_value = attached_volume_m ) <EOL> novaclient_m = mock . Mock ( ) <EOL> novaclient_m . volumes = mock . Mock ( ) <EOL> novaclient_m . volumes . delete_server_volume = mock . Mock ( ) <EOL> with contextlib . nested ( <EOL> mock . patch . object ( NovaClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = novaclient_m ) ) , <EOL> mock . patch . object ( CinderClient , '<STR_LIT>' , <EOL> mock . Mock ( return_value = cinder_client_m ) ) , <EOL> mock . patch . object ( volume , '<STR_LIT>' , mock . Mock ( ) ) ) : <EOL> server . detach_volume ( ctx = ctx_m ) <EOL> novaclient_m . volumes . delete_server_volume . assert_called_once_with ( <EOL> server_id , attachment_id ) <EOL> volume . wait_until_status . assert_called_once_with ( <EOL> cinder_client = cinder_client_m , <EOL> volume_id = volume_id , <EOL> status = volume . VOLUME_STATUS_AVAILABLE ) </s>
