<s> import importlib <EOL> import itertools <EOL> import os <EOL> import unittest <EOL> try : <EOL> import unittest . mock as mock <EOL> except ImportError : <EOL> import mock <EOL> from oslo_config import cfg <EOL> from cloudbaseinit import exception <EOL> from cloudbaseinit . tests import testutils <EOL> CONF = cfg . CONF <EOL> OPEN = mock . mock_open ( ) <EOL> class TestWindowsConfigDriveManager ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> module_path = "<STR_LIT>" <EOL> mock_ctypes = mock . MagicMock ( ) <EOL> mock_ctypes . wintypes = mock . MagicMock ( ) <EOL> self . _module_patcher = mock . patch . dict ( <EOL> '<STR_LIT>' , <EOL> { '<STR_LIT>' : mock . Mock ( ) , <EOL> '<STR_LIT>' : mock_ctypes , <EOL> '<STR_LIT>' : mock . Mock ( ) } ) <EOL> self . _module_patcher . start ( ) <EOL> self . addCleanup ( self . _module_patcher . stop ) <EOL> self . conf_module = importlib . import_module ( module_path ) <EOL> self . conf_module . osutils_factory = mock . Mock ( ) <EOL> self . conf_module . disk . Disk = mock . MagicMock ( ) <EOL> self . conf_module . tempfile = mock . Mock ( ) <EOL> self . mock_gettempdir = self . conf_module . tempfile . gettempdir <EOL> self . mock_gettempdir . return_value = "<STR_LIT>" <EOL> self . conf_module . uuid = mock . Mock ( ) <EOL> self . mock_uuid4 = self . conf_module . uuid . uuid4 <EOL> self . mock_uuid4 . return_value = "<STR_LIT>" <EOL> self . _config_manager = self . conf_module . WindowsConfigDriveManager ( ) <EOL> self . addCleanup ( os . rmdir , self . _config_manager . target_path ) <EOL> self . osutils = mock . Mock ( ) <EOL> self . _config_manager . _osutils = self . osutils <EOL> self . snatcher = testutils . LogSnatcher ( module_path ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_check_for_config_drive ( self , mock_exists , exists = True , <EOL> label = "<STR_LIT>" , fail = False ) : <EOL> drive = "<STR_LIT>" <EOL> self . osutils . get_volume_label . return_value = label <EOL> mock_exists . return_value = exists <EOL> with self . snatcher : <EOL> response = self . _config_manager . _check_for_config_drive ( drive ) <EOL> self . osutils . get_volume_label . assert_called_once_with ( drive ) <EOL> if exists and not fail : <EOL> self . assertEqual ( [ "<STR_LIT>" ] , <EOL> self . snatcher . output ) <EOL> self . assertEqual ( not fail , response ) <EOL> def test_check_for_config_drive_exists ( self ) : <EOL> self . _test_check_for_config_drive ( ) <EOL> def test_check_for_config_drive_exists_upper_label ( self ) : <EOL> self . _test_check_for_config_drive ( label = "<STR_LIT>" ) <EOL> def test_check_for_config_drive_missing ( self ) : <EOL> self . _test_check_for_config_drive ( exists = False , fail = True ) <EOL> def test_check_for_config_drive_wrong_label ( self ) : <EOL> self . _test_check_for_config_drive ( label = "<STR_LIT>" , fail = True ) <EOL> def _test_get_iso_file_size ( self , fixed = True , small = False , <EOL> found_iso = True ) : <EOL> device = mock . Mock ( ) <EOL> device . fixed = fixed <EOL> device . size = ( self . conf_module . OFFSET_BLOCK_SIZE + <EOL> self . conf_module . PEEK_SIZE + int ( not small ) ) <EOL> iso_id = self . conf_module . ISO_ID <EOL> if not found_iso : <EOL> iso_id = b"<STR_LIT>" <EOL> iso_off = self . conf_module . OFFSET_ISO_ID - <NUM_LIT:1> <EOL> volume_off = self . conf_module . OFFSET_VOLUME_SIZE - <NUM_LIT:1> <EOL> block_off = self . conf_module . OFFSET_BLOCK_SIZE - <NUM_LIT:1> <EOL> volume_bytes = b'<STR_LIT>' <EOL> block_bytes = b'<STR_LIT>' <EOL> device . seek . side_effect = [ iso_off , volume_off , block_off ] <EOL> device . read . side_effect = [ iso_id , volume_bytes , block_bytes ] <EOL> response = self . _config_manager . _get_iso_file_size ( device ) <EOL> if not fixed or small or not found_iso : <EOL> self . assertIsNone ( response ) <EOL> return <EOL> seek_calls = [ <EOL> mock . call ( self . conf_module . OFFSET_ISO_ID ) , <EOL> mock . call ( self . conf_module . OFFSET_VOLUME_SIZE ) , <EOL> mock . call ( self . conf_module . OFFSET_BLOCK_SIZE ) ] <EOL> read_calls = [ <EOL> mock . call ( len ( iso_id ) , <EOL> skip = self . conf_module . OFFSET_ISO_ID - iso_off ) , <EOL> mock . call ( self . conf_module . PEEK_SIZE , <EOL> skip = self . conf_module . OFFSET_VOLUME_SIZE - volume_off ) , <EOL> mock . call ( self . conf_module . PEEK_SIZE , <EOL> skip = self . conf_module . OFFSET_BLOCK_SIZE - block_off ) ] <EOL> device . seek . assert_has_calls ( seek_calls ) <EOL> device . read . assert_has_calls ( read_calls ) <EOL> self . assertEqual ( <NUM_LIT:100> * <NUM_LIT> , response ) <EOL> def test_get_iso_file_size_not_fixed ( self ) : <EOL> self . _test_get_iso_file_size ( fixed = False ) <EOL> def test_get_iso_file_size_small ( self ) : <EOL> self . _test_get_iso_file_size ( small = True ) <EOL> def test_get_iso_file_size_not_found ( self ) : <EOL> self . _test_get_iso_file_size ( found_iso = False ) <EOL> def test_get_iso_file_size ( self ) : <EOL> self . _test_get_iso_file_size ( ) <EOL> @ mock . patch ( "<STR_LIT>" , new = OPEN ) <EOL> def test_write_iso_file ( self ) : <EOL> file_path = "<STR_LIT>" <EOL> file_size = <NUM_LIT:100> * <NUM_LIT> <EOL> sector_size = self . conf_module . MAX_SECTOR_SIZE <EOL> offsets = list ( range ( <NUM_LIT:0> , file_size , sector_size ) ) <EOL> remain = file_size % sector_size <EOL> reads = ( [ b"<STR_LIT:\x00>" * sector_size ] * <EOL> ( len ( offsets ) - int ( bool ( remain ) ) ) + <EOL> ( [ b"<STR_LIT:\x00>" * remain ] if remain else [ ] ) ) <EOL> device = mock . Mock ( ) <EOL> device_seek_calls = [ mock . call ( off ) for off in offsets ] <EOL> device_read_calls = [ <EOL> mock . call ( min ( sector_size , file_size - off ) , skip = <NUM_LIT:0> ) <EOL> for off in offsets ] <EOL> stream_write_calls = [ mock . call ( read ) for read in reads ] <EOL> device . seek . side_effect = offsets <EOL> device . read . side_effect = reads <EOL> self . _config_manager . _write_iso_file ( device , file_path , file_size ) <EOL> device . seek . assert_has_calls ( device_seek_calls ) <EOL> device . read . assert_has_calls ( device_read_calls ) <EOL> OPEN . return_value . write . assert_has_calls ( stream_write_calls ) <EOL> def _test_extract_files_from_iso ( self , exit_code ) : <EOL> fake_path = os . path . join ( '<STR_LIT>' , '<STR_LIT:path>' ) <EOL> fake_target_path = os . path . join ( fake_path , '<STR_LIT:target>' ) <EOL> self . _config_manager . target_path = fake_target_path <EOL> args = [ CONF . bsdtar_path , '<STR_LIT>' , fake_path , '<STR_LIT>' , fake_target_path ] <EOL> self . osutils . execute_process . return_value = ( '<STR_LIT>' , '<STR_LIT>' , <EOL> exit_code ) <EOL> if exit_code : <EOL> self . assertRaises ( exception . CloudbaseInitException , <EOL> self . _config_manager . _extract_files_from_iso , <EOL> fake_path ) <EOL> else : <EOL> self . _config_manager . _extract_files_from_iso ( fake_path ) <EOL> self . osutils . execute_process . assert_called_once_with ( args , False ) <EOL> def test_extract_files_from_iso ( self ) : <EOL> self . _test_extract_files_from_iso ( exit_code = <NUM_LIT:0> ) <EOL> def test_extract_files_from_iso_fail ( self ) : <EOL> self . _test_extract_files_from_iso ( exit_code = <NUM_LIT:1> ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_extract_iso_from_devices ( self , mock_get_iso_file_size , <EOL> mock_write_iso_file , <EOL> mock_extract_files_from_iso , <EOL> found = True ) : <EOL> size = <NUM_LIT:100> * <NUM_LIT> <EOL> devices = [ mock . MagicMock ( ) for _ in range ( <NUM_LIT:4> ) ] <EOL> devices [ <NUM_LIT:1> ] . __enter__ . side_effect = [ Exception ] <EOL> rest = [ size ] if found else [ None ] <EOL> mock_get_iso_file_size . side_effect = [ None ] + rest * <NUM_LIT:2> <EOL> file_path = os . path . join ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> with self . snatcher : <EOL> response = self . _config_manager . _extract_iso_from_devices ( devices ) <EOL> self . mock_gettempdir . assert_called_once_with ( ) <EOL> mock_get_iso_file_size . assert_has_calls ( [ <EOL> mock . call ( devices [ <NUM_LIT:0> ] ) , mock . call ( devices [ <NUM_LIT:2> ] ) ] ) <EOL> expected_log = [ <EOL> "<STR_LIT>" % <EOL> { "<STR_LIT>" : devices [ <NUM_LIT:1> ] , "<STR_LIT:error>" : Exception ( ) } ] <EOL> if found : <EOL> mock_write_iso_file . assert_called_once_with ( devices [ <NUM_LIT:2> ] , <EOL> file_path , size ) <EOL> mock_extract_files_from_iso . assert_called_once_with ( file_path ) <EOL> expected_log . append ( "<STR_LIT>" % devices [ <NUM_LIT:2> ] ) <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> self . assertEqual ( found , response ) <EOL> def test_extract_iso_from_devices_not_found ( self ) : <EOL> self . _test_extract_iso_from_devices ( found = False ) <EOL> def test_extract_iso_from_devices ( self ) : <EOL> self . _test_extract_iso_from_devices ( ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_get_config_drive_from_cdrom_drive ( self , mock_os_rmdir , <EOL> mock_copytree , <EOL> mock_check_for_config_drive , <EOL> found = True ) : <EOL> drives = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> self . osutils . get_cdrom_drives . return_value = drives <EOL> checks = [ False , False , True , False ] <EOL> if not found : <EOL> checks [ <NUM_LIT:2> ] = False <EOL> mock_check_for_config_drive . side_effect = checks <EOL> response = self . _config_manager . _get_config_drive_from_cdrom_drive ( ) <EOL> self . osutils . get_cdrom_drives . assert_called_once_with ( ) <EOL> idx = <NUM_LIT:3> if found else <NUM_LIT:4> <EOL> check_calls = [ mock . call ( drive ) for drive in drives [ : idx ] ] <EOL> mock_check_for_config_drive . assert_has_calls ( check_calls ) <EOL> if found : <EOL> mock_os_rmdir . assert_called_once_with ( <EOL> self . _config_manager . target_path ) <EOL> mock_copytree . assert_called_once_with ( <EOL> drives [ <NUM_LIT:2> ] , self . _config_manager . target_path ) <EOL> self . assertEqual ( found , response ) <EOL> def test_get_config_drive_from_cdrom_drive_not_found ( self ) : <EOL> self . _test_get_config_drive_from_cdrom_drive ( found = False ) <EOL> def test_get_config_drive_from_cdrom_drive ( self ) : <EOL> self . _test_get_config_drive_from_cdrom_drive ( ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_get_config_drive_from_raw_hdd ( self , mock_map , <EOL> mock_extract_iso_from_devices ) : <EOL> Disk = self . conf_module . disk . Disk <EOL> paths = [ mock . Mock ( ) for _ in range ( <NUM_LIT:3> ) ] <EOL> self . osutils . get_physical_disks . return_value = paths <EOL> mock_extract_iso_from_devices . return_value = True <EOL> response = self . _config_manager . _get_config_drive_from_raw_hdd ( ) <EOL> mock_map . assert_called_once_with ( Disk , paths ) <EOL> self . osutils . get_physical_disks . assert_called_once_with ( ) <EOL> mock_extract_iso_from_devices . assert_called_once_with ( <EOL> mock_map . return_value ) <EOL> self . assertTrue ( response ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_get_config_drive_from_vfat ( self , mock_is_vfat_drive , <EOL> mock_copy_from_vfat_drive ) : <EOL> self . osutils . get_physical_disks . return_value = ( <EOL> mock . sentinel . drive1 , <EOL> mock . sentinel . drive2 , <EOL> ) <EOL> mock_is_vfat_drive . side_effect = ( None , True ) <EOL> with testutils . LogSnatcher ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as snatcher : <EOL> response = self . _config_manager . _get_config_drive_from_vfat ( ) <EOL> self . assertTrue ( response ) <EOL> self . osutils . get_physical_disks . assert_called_once_with ( ) <EOL> expected_is_vfat_calls = [ <EOL> mock . call ( self . osutils , mock . sentinel . drive1 ) , <EOL> mock . call ( self . osutils , mock . sentinel . drive2 ) , <EOL> ] <EOL> self . assertEqual ( expected_is_vfat_calls , mock_is_vfat_drive . mock_calls ) <EOL> mock_copy_from_vfat_drive . assert_called_once_with ( <EOL> self . osutils , <EOL> mock . sentinel . drive2 , <EOL> self . _config_manager . target_path ) <EOL> expected_logging = [ <EOL> '<STR_LIT>' % mock . sentinel . drive2 , <EOL> ] <EOL> self . assertEqual ( expected_logging , snatcher . output ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_config_drive_from_partition ( self , <EOL> mock_extract_iso_from_devices , <EOL> found = True ) : <EOL> paths = [ mock . Mock ( ) for _ in range ( <NUM_LIT:3> ) ] <EOL> self . osutils . get_physical_disks . return_value = paths <EOL> disks = list ( map ( self . conf_module . disk . Disk , paths ) ) <EOL> mock_extract_iso_from_devices . side_effect = [ False , found , found ] <EOL> idx = <NUM_LIT:3> - int ( found ) <EOL> extract_calls = [ mock . call ( disk . partitions ( ) ) <EOL> for disk in disks [ : idx ] ] <EOL> response = self . _config_manager . _get_config_drive_from_partition ( ) <EOL> self . osutils . get_physical_disks . assert_called_once_with ( ) <EOL> mock_extract_iso_from_devices . assert_has_calls ( extract_calls ) <EOL> self . assertEqual ( found , response ) <EOL> def test_get_config_drive_from_partition_not_found ( self ) : <EOL> self . _test_get_config_drive_from_partition ( found = False ) <EOL> def test_get_config_drive_from_partition ( self ) : <EOL> self . _test_get_config_drive_from_partition ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_config_drive_from_volume ( self , mock_check_for_config_drive , <EOL> mock_copytree , mock_os_rmdir , <EOL> found = True ) : <EOL> volumes = [ mock . Mock ( ) for _ in range ( <NUM_LIT:3> ) ] <EOL> self . osutils . get_volumes . return_value = volumes <EOL> checks = [ False , found , found ] <EOL> mock_check_for_config_drive . side_effect = checks <EOL> idx = <NUM_LIT:3> - int ( found ) <EOL> check_calls = [ mock . call ( volume ) for volume in volumes [ : idx ] ] <EOL> response = self . _config_manager . _get_config_drive_from_volume ( ) <EOL> self . osutils . get_volumes . assert_called_once_with ( ) <EOL> mock_check_for_config_drive . assert_has_calls ( check_calls ) <EOL> if found : <EOL> mock_os_rmdir . assert_called_once_with ( <EOL> self . _config_manager . target_path ) <EOL> mock_copytree . assert_called_once_with ( <EOL> volumes [ <NUM_LIT:1> ] , self . _config_manager . target_path ) <EOL> self . assertEqual ( found , response ) <EOL> def test_get_config_drive_from_volume_not_found ( self ) : <EOL> self . _test_get_config_drive_from_volume ( found = False ) <EOL> def test_get_config_drive_from_volume ( self ) : <EOL> self . _test_get_config_drive_from_volume ( ) <EOL> def _test__get_config_drive_files ( self , cd_type , cd_location , <EOL> func , found = True ) : <EOL> response = self . _config_manager . _get_config_drive_files ( cd_type , <EOL> cd_location ) <EOL> if found : <EOL> if func : <EOL> func . assert_called_once_with ( ) <EOL> self . assertEqual ( func . return_value , response ) <EOL> else : <EOL> self . assertFalse ( response ) <EOL> def test__get_config_drive_files_not_found ( self ) : <EOL> self . _test__get_config_drive_files ( None , None , None , found = False ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test__get_config_drive_files_cdrom_iso ( self , func ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , func ) <EOL> def test__get_config_drive_files_cdrom_vfat ( self ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , None ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test__get_config_drive_files_hdd_iso ( self , func ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , func ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test__get_config_drive_files_hdd_vfat ( self , func ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , func ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test__get_config_drive_files_partition_iso ( self , func ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , func ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test__get_config_drive_files_partition_vfat ( self , func ) : <EOL> self . _test__get_config_drive_files ( <EOL> "<STR_LIT>" , "<STR_LIT>" , func ) <EOL> @ mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _test_get_config_drive_files ( self , mock_get_config_drive_files , <EOL> found = True ) : <EOL> check_types = [ "<STR_LIT>" , "<STR_LIT>" ] if found else [ ] <EOL> check_locations = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> product = list ( itertools . product ( check_types , check_locations ) ) <EOL> product_calls = [ mock . call ( cd_type , cd_location ) <EOL> for cd_type , cd_location in product ] <EOL> mock_get_config_drive_files . side_effect = [ False ] * ( len ( product_calls ) - <NUM_LIT:1> ) + [ True ] <EOL> expected_log = [ "<STR_LIT>" % <EOL> { "<STR_LIT:type>" : cd_type , "<STR_LIT:location>" : cd_location } <EOL> for cd_type , cd_location in product ] <EOL> with self . snatcher : <EOL> response = self . _config_manager . get_config_drive_files ( <EOL> check_types , check_locations ) <EOL> mock_get_config_drive_files . assert_has_calls ( product_calls ) <EOL> self . assertEqual ( expected_log , self . snatcher . output ) <EOL> self . assertEqual ( found , response ) <EOL> def test_get_config_drive_files_not_found ( self ) : <EOL> self . _test_get_config_drive_files ( found = False ) <EOL> def test_get_config_drive_files ( self ) : <EOL> self . _test_get_config_drive_files ( ) </s>
