<s> """<STR_LIT>""" <EOL> import os <EOL> import tarfile <EOL> import mock <EOL> from oslo_utils import units <EOL> from oslo_vmware import rw_handles <EOL> from nova import exception <EOL> from nova import objects <EOL> from nova import test <EOL> import nova . tests . unit . image . fake <EOL> from nova . tests import uuidsentinel <EOL> from nova . virt . vmwareapi import constants <EOL> from nova . virt . vmwareapi import images <EOL> from nova . virt . vmwareapi import vm_util <EOL> class VMwareImagesTestCase ( test . NoDBTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_fetch_image ( self ) : <EOL> """<STR_LIT>""" <EOL> dc_name = '<STR_LIT>' <EOL> file_path = '<STR_LIT>' <EOL> ds_name = '<STR_LIT>' <EOL> host = mock . MagicMock ( ) <EOL> port = <NUM_LIT> <EOL> context = mock . MagicMock ( ) <EOL> image_data = { <EOL> '<STR_LIT:id>' : nova . tests . unit . image . fake . get_valid_image_id ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:size>' : <NUM_LIT> , <EOL> } <EOL> read_file_handle = mock . MagicMock ( ) <EOL> write_file_handle = mock . MagicMock ( ) <EOL> read_iter = mock . MagicMock ( ) <EOL> instance = objects . Instance ( id = <NUM_LIT:1> , <EOL> uuid = uuidsentinel . foo , <EOL> image_ref = image_data [ '<STR_LIT:id>' ] ) <EOL> def fake_read_handle ( read_iter ) : <EOL> return read_file_handle <EOL> def fake_write_handle ( host , port , dc_name , ds_name , cookies , <EOL> file_path , file_size ) : <EOL> return write_file_handle <EOL> with test . nested ( <EOL> mock . patch . object ( rw_handles , '<STR_LIT>' , <EOL> side_effect = fake_read_handle ) , <EOL> mock . patch . object ( rw_handles , '<STR_LIT>' , <EOL> side_effect = fake_write_handle ) , <EOL> mock . patch . object ( images , '<STR_LIT>' ) , <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' , <EOL> return_value = image_data ) , <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' , <EOL> return_value = read_iter ) , <EOL> ) as ( glance_read , http_write , start_transfer , image_show , <EOL> image_download ) : <EOL> images . fetch_image ( context , instance , <EOL> host , port , dc_name , <EOL> ds_name , file_path ) <EOL> glance_read . assert_called_once_with ( read_iter ) <EOL> http_write . assert_called_once_with ( host , port , dc_name , ds_name , None , <EOL> file_path , image_data [ '<STR_LIT:size>' ] ) <EOL> start_transfer . assert_called_once_with ( <EOL> context , read_file_handle , <EOL> image_data [ '<STR_LIT:size>' ] , <EOL> write_file_handle = write_file_handle ) <EOL> image_download . assert_called_once_with ( context , instance [ '<STR_LIT>' ] ) <EOL> image_show . assert_called_once_with ( context , instance [ '<STR_LIT>' ] ) <EOL> def _setup_mock_get_remote_image_service ( self , <EOL> mock_get_remote_image_service , <EOL> metadata ) : <EOL> mock_image_service = mock . MagicMock ( ) <EOL> mock_image_service . show . return_value = metadata <EOL> mock_get_remote_image_service . return_value = [ mock_image_service , '<STR_LIT:i>' ] <EOL> def test_get_vmdk_name_from_ovf ( self ) : <EOL> ovf_path = os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) <EOL> with open ( ovf_path ) as f : <EOL> ovf_descriptor = f . read ( ) <EOL> vmdk_name = images . get_vmdk_name_from_ovf ( ovf_descriptor ) <EOL> self . assertEqual ( "<STR_LIT>" , vmdk_name ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( tarfile , '<STR_LIT>' ) <EOL> def test_fetch_image_ova ( self , mock_tar_open , mock_write_class , <EOL> mock_read_class ) : <EOL> session = mock . MagicMock ( ) <EOL> ovf_descriptor = None <EOL> ovf_path = os . path . join ( os . path . dirname ( __file__ ) , '<STR_LIT>' ) <EOL> with open ( ovf_path ) as f : <EOL> ovf_descriptor = f . read ( ) <EOL> with test . nested ( <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' ) , <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' ) , <EOL> mock . patch . object ( images , '<STR_LIT>' ) , <EOL> mock . patch . object ( images , '<STR_LIT>' ) , <EOL> mock . patch . object ( session , '<STR_LIT>' ) , <EOL> mock . patch . object ( vm_util , '<STR_LIT>' ) <EOL> ) as ( mock_image_api_get , <EOL> mock_image_api_download , <EOL> mock_start_transfer , <EOL> mock_build_shadow_vm_config_spec , <EOL> mock_call_method , <EOL> mock_get_vmdk_info ) : <EOL> image_data = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:size>' : <NUM_LIT> } <EOL> instance = mock . MagicMock ( ) <EOL> instance . image_ref = image_data [ '<STR_LIT:id>' ] <EOL> mock_image_api_get . return_value = image_data <EOL> vm_folder_ref = mock . MagicMock ( ) <EOL> res_pool_ref = mock . MagicMock ( ) <EOL> context = mock . MagicMock ( ) <EOL> mock_read_handle = mock . MagicMock ( ) <EOL> mock_read_class . return_value = mock_read_handle <EOL> mock_write_handle = mock . MagicMock ( ) <EOL> mock_write_class . return_value = mock_write_handle <EOL> mock_write_handle . get_imported_vm . return_value = mock . sentinel . vm_ref <EOL> mock_ovf = mock . MagicMock ( ) <EOL> mock_ovf . name = '<STR_LIT>' <EOL> mock_vmdk = mock . MagicMock ( ) <EOL> mock_vmdk . name = "<STR_LIT>" <EOL> def fake_extract ( name ) : <EOL> if name == mock_ovf : <EOL> m = mock . MagicMock ( ) <EOL> m . read . return_value = ovf_descriptor <EOL> return m <EOL> elif name == mock_vmdk : <EOL> return mock_read_handle <EOL> mock_tar = mock . MagicMock ( ) <EOL> mock_tar . __iter__ = mock . Mock ( return_value = iter ( [ mock_ovf , <EOL> mock_vmdk ] ) ) <EOL> mock_tar . extractfile = fake_extract <EOL> mock_tar_open . return_value . __enter__ . return_value = mock_tar <EOL> images . fetch_image_ova ( <EOL> context , instance , session , '<STR_LIT>' , '<STR_LIT>' , <EOL> vm_folder_ref , res_pool_ref ) <EOL> mock_tar_open . assert_called_once_with ( mode = '<STR_LIT>' , <EOL> fileobj = mock_read_handle ) <EOL> mock_start_transfer . assert_called_once_with ( context , <EOL> mock_read_handle , <NUM_LIT> , write_file_handle = mock_write_handle ) <EOL> mock_get_vmdk_info . assert_called_once_with ( <EOL> session , mock . sentinel . vm_ref , '<STR_LIT>' ) <EOL> mock_call_method . assert_called_once_with ( <EOL> session . vim , "<STR_LIT>" , mock . sentinel . vm_ref ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_fetch_image_stream_optimized ( self , <EOL> mock_write_class , <EOL> mock_read_class ) : <EOL> """<STR_LIT>""" <EOL> session = mock . MagicMock ( ) <EOL> with test . nested ( <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' ) , <EOL> mock . patch . object ( images . IMAGE_API , '<STR_LIT>' ) , <EOL> mock . patch . object ( images , '<STR_LIT>' ) , <EOL> mock . patch . object ( images , '<STR_LIT>' ) , <EOL> mock . patch . object ( session , '<STR_LIT>' ) , <EOL> mock . patch . object ( vm_util , '<STR_LIT>' ) <EOL> ) as ( mock_image_api_get , <EOL> mock_image_api_download , <EOL> mock_start_transfer , <EOL> mock_build_shadow_vm_config_spec , <EOL> mock_call_method , <EOL> mock_get_vmdk_info ) : <EOL> image_data = { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:size>' : <NUM_LIT> } <EOL> instance = mock . MagicMock ( ) <EOL> instance . image_ref = image_data [ '<STR_LIT:id>' ] <EOL> mock_image_api_get . return_value = image_data <EOL> vm_folder_ref = mock . MagicMock ( ) <EOL> res_pool_ref = mock . MagicMock ( ) <EOL> context = mock . MagicMock ( ) <EOL> mock_read_handle = mock . MagicMock ( ) <EOL> mock_read_class . return_value = mock_read_handle <EOL> mock_write_handle = mock . MagicMock ( ) <EOL> mock_write_class . return_value = mock_write_handle <EOL> mock_write_handle . get_imported_vm . return_value = mock . sentinel . vm_ref <EOL> images . fetch_image_stream_optimized ( <EOL> context , instance , session , '<STR_LIT>' , '<STR_LIT>' , <EOL> vm_folder_ref , res_pool_ref ) <EOL> mock_start_transfer . assert_called_once_with ( context , <EOL> mock_read_handle , <NUM_LIT> , write_file_handle = mock_write_handle ) <EOL> mock_call_method . assert_called_once_with ( <EOL> session . vim , "<STR_LIT>" , mock . sentinel . vm_ref ) <EOL> mock_get_vmdk_info . assert_called_once_with ( <EOL> session , mock . sentinel . vm_ref , '<STR_LIT>' ) <EOL> def test_from_image_with_image_ref ( self ) : <EOL> raw_disk_size_in_gb = <NUM_LIT> <EOL> raw_disk_size_in_bytes = raw_disk_size_in_gb * units . Gi <EOL> image_id = nova . tests . unit . image . fake . get_valid_image_id ( ) <EOL> mdata = { '<STR_LIT:size>' : raw_disk_size_in_bytes , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : constants . DEFAULT_OS_TYPE , <EOL> "<STR_LIT>" : constants . DEFAULT_ADAPTER_TYPE , <EOL> "<STR_LIT>" : constants . DEFAULT_DISK_TYPE , <EOL> "<STR_LIT>" : constants . DEFAULT_VIF_MODEL , <EOL> "<STR_LIT>" : True } } <EOL> mdata = objects . ImageMeta . from_dict ( mdata ) <EOL> img_props = images . VMwareImage . from_image ( image_id , mdata ) <EOL> image_size_in_kb = raw_disk_size_in_bytes / units . Ki <EOL> self . assertEqual ( constants . DEFAULT_OS_TYPE , img_props . os_type ) <EOL> self . assertEqual ( constants . DEFAULT_ADAPTER_TYPE , <EOL> img_props . adapter_type ) <EOL> self . assertEqual ( constants . DEFAULT_DISK_TYPE , img_props . disk_type ) <EOL> self . assertEqual ( constants . DEFAULT_VIF_MODEL , img_props . vif_model ) <EOL> self . assertTrue ( img_props . linked_clone ) <EOL> self . assertEqual ( image_size_in_kb , img_props . file_size_in_kb ) <EOL> def _image_build ( self , image_lc_setting , global_lc_setting , <EOL> disk_format = constants . DEFAULT_DISK_FORMAT , <EOL> os_type = constants . DEFAULT_OS_TYPE , <EOL> adapter_type = constants . DEFAULT_ADAPTER_TYPE , <EOL> disk_type = constants . DEFAULT_DISK_TYPE , <EOL> vif_model = constants . DEFAULT_VIF_MODEL ) : <EOL> self . flags ( use_linked_clone = global_lc_setting , group = '<STR_LIT>' ) <EOL> raw_disk_size_in_gb = <NUM_LIT> <EOL> raw_disk_size_in_btyes = raw_disk_size_in_gb * units . Gi <EOL> image_id = nova . tests . unit . image . fake . get_valid_image_id ( ) <EOL> mdata = { '<STR_LIT:size>' : raw_disk_size_in_btyes , <EOL> '<STR_LIT>' : disk_format , <EOL> '<STR_LIT>' : { <EOL> "<STR_LIT>" : os_type , <EOL> "<STR_LIT>" : adapter_type , <EOL> "<STR_LIT>" : disk_type , <EOL> "<STR_LIT>" : vif_model } } <EOL> if image_lc_setting is not None : <EOL> mdata [ '<STR_LIT>' ] [ "<STR_LIT>" ] = image_lc_setting <EOL> mdata = objects . ImageMeta . from_dict ( mdata ) <EOL> return images . VMwareImage . from_image ( image_id , mdata ) <EOL> def test_use_linked_clone_override_nf ( self ) : <EOL> image_props = self . _image_build ( None , False ) <EOL> self . assertFalse ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_linked_clone_override_nt ( self ) : <EOL> image_props = self . _image_build ( None , True ) <EOL> self . assertTrue ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_linked_clone_override_ny ( self ) : <EOL> image_props = self . _image_build ( None , "<STR_LIT:yes>" ) <EOL> self . assertTrue ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_linked_clone_override_ft ( self ) : <EOL> image_props = self . _image_build ( False , True ) <EOL> self . assertFalse ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_linked_clone_override_string_nt ( self ) : <EOL> image_props = self . _image_build ( "<STR_LIT>" , True ) <EOL> self . assertFalse ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_linked_clone_override_string_yf ( self ) : <EOL> image_props = self . _image_build ( "<STR_LIT:yes>" , False ) <EOL> self . assertTrue ( image_props . linked_clone , <EOL> "<STR_LIT>" ) <EOL> def test_use_disk_format_iso ( self ) : <EOL> image = self . _image_build ( None , True , disk_format = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , image . file_type ) <EOL> self . assertTrue ( image . is_iso ) <EOL> def test_use_bad_disk_format ( self ) : <EOL> self . assertRaises ( exception . InvalidDiskFormat , <EOL> self . _image_build , <EOL> None , <EOL> True , <EOL> disk_format = '<STR_LIT>' ) <EOL> def test_image_no_defaults ( self ) : <EOL> image = self . _image_build ( False , False , <EOL> disk_format = '<STR_LIT>' , <EOL> os_type = '<STR_LIT>' , <EOL> adapter_type = '<STR_LIT>' , <EOL> disk_type = '<STR_LIT>' , <EOL> vif_model = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , image . file_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . os_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . adapter_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . disk_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . vif_model ) <EOL> self . assertFalse ( image . linked_clone ) <EOL> def test_image_defaults ( self ) : <EOL> image = images . VMwareImage ( image_id = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , image . os_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . adapter_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . disk_type ) <EOL> self . assertEqual ( '<STR_LIT>' , image . vif_model ) </s>
