<s> """<STR_LIT>""" <EOL> import os <EOL> import tempfile <EOL> from ironic_lib import disk_utils <EOL> from ironic_lib import utils as ironic_utils <EOL> import mock <EOL> from oslo_config import cfg <EOL> from oslo_utils import fileutils <EOL> from oslo_utils import uuidutils <EOL> from ironic . common import dhcp_factory <EOL> from ironic . common import driver_factory <EOL> from ironic . common import exception <EOL> from ironic . common import keystone <EOL> from ironic . common import pxe_utils <EOL> from ironic . common import states <EOL> from ironic . common import utils <EOL> from ironic . conductor import task_manager <EOL> from ironic . conductor import utils as manager_utils <EOL> from ironic . drivers . modules import agent_base_vendor <EOL> from ironic . drivers . modules import agent_client <EOL> from ironic . drivers . modules import deploy_utils <EOL> from ironic . drivers . modules import fake <EOL> from ironic . drivers . modules import iscsi_deploy <EOL> from ironic . drivers . modules import pxe <EOL> from ironic . tests . unit . conductor import mgr_utils <EOL> from ironic . tests . unit . db import base as db_base <EOL> from ironic . tests . unit . db import utils as db_utils <EOL> from ironic . tests . unit . objects import utils as obj_utils <EOL> CONF = cfg . CONF <EOL> INST_INFO_DICT = db_utils . get_test_pxe_instance_info ( ) <EOL> DRV_INFO_DICT = db_utils . get_test_pxe_driver_info ( ) <EOL> DRV_INTERNAL_INFO_DICT = db_utils . get_test_pxe_driver_internal_info ( ) <EOL> class IscsiDeployPrivateMethodsTestCase ( db_base . DbTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( IscsiDeployPrivateMethodsTestCase , self ) . setUp ( ) <EOL> n = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : INST_INFO_DICT , <EOL> '<STR_LIT>' : DRV_INFO_DICT , <EOL> '<STR_LIT>' : DRV_INTERNAL_INFO_DICT , <EOL> } <EOL> mgr_utils . mock_the_extension_manager ( driver = "<STR_LIT>" ) <EOL> self . node = obj_utils . create_test_node ( self . context , ** n ) <EOL> def test__save_disk_layout ( self ) : <EOL> info = dict ( INST_INFO_DICT ) <EOL> info [ '<STR_LIT>' ] = <NUM_LIT:10> <EOL> info [ '<STR_LIT>' ] = <NUM_LIT:0> <EOL> info [ '<STR_LIT>' ] = <NUM_LIT:10> <EOL> info [ '<STR_LIT>' ] = False <EOL> self . node . instance_info = info <EOL> iscsi_deploy . _save_disk_layout ( self . node , info ) <EOL> self . node . refresh ( ) <EOL> for param in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . assertEqual ( <EOL> info [ param ] , self . node . driver_internal_info [ '<STR_LIT>' ] [ param ] <EOL> ) <EOL> def test__get_image_dir_path ( self ) : <EOL> self . assertEqual ( os . path . join ( CONF . pxe . images_path , <EOL> self . node . uuid ) , <EOL> iscsi_deploy . _get_image_dir_path ( self . node . uuid ) ) <EOL> def test__get_image_file_path ( self ) : <EOL> self . assertEqual ( os . path . join ( CONF . pxe . images_path , <EOL> self . node . uuid , <EOL> '<STR_LIT>' ) , <EOL> iscsi_deploy . _get_image_file_path ( self . node . uuid ) ) <EOL> class IscsiDeployMethodsTestCase ( db_base . DbTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( IscsiDeployMethodsTestCase , self ) . setUp ( ) <EOL> instance_info = dict ( INST_INFO_DICT ) <EOL> instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> n = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_info , <EOL> '<STR_LIT>' : DRV_INFO_DICT , <EOL> '<STR_LIT>' : DRV_INTERNAL_INFO_DICT , <EOL> } <EOL> mgr_utils . mock_the_extension_manager ( driver = "<STR_LIT>" ) <EOL> self . node = obj_utils . create_test_node ( self . context , ** n ) <EOL> @ mock . patch . object ( disk_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_check_image_size ( self , get_image_mb_mock ) : <EOL> get_image_mb_mock . return_value = <NUM_LIT:1000> <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> iscsi_deploy . check_image_size ( task ) <EOL> get_image_mb_mock . assert_called_once_with ( <EOL> iscsi_deploy . _get_image_file_path ( task . node . uuid ) ) <EOL> @ mock . patch . object ( disk_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_check_image_size_fails ( self , get_image_mb_mock ) : <EOL> get_image_mb_mock . return_value = <NUM_LIT> <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . check_image_size , <EOL> task ) <EOL> get_image_mb_mock . assert_called_once_with ( <EOL> iscsi_deploy . _get_image_file_path ( task . node . uuid ) ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_cache_instance_images_master_path ( self , mock_fetch_image ) : <EOL> temp_dir = tempfile . mkdtemp ( ) <EOL> self . config ( images_path = temp_dir , group = '<STR_LIT>' ) <EOL> self . config ( instance_master_path = os . path . join ( temp_dir , <EOL> '<STR_LIT>' ) , <EOL> group = '<STR_LIT>' ) <EOL> fileutils . ensure_tree ( CONF . pxe . instance_master_path ) <EOL> ( uuid , image_path ) = iscsi_deploy . cache_instance_image ( None , self . node ) <EOL> mock_fetch_image . assert_called_once_with ( None , <EOL> mock . ANY , <EOL> [ ( uuid , image_path ) ] , True ) <EOL> self . assertEqual ( '<STR_LIT>' , uuid ) <EOL> self . assertEqual ( os . path . join ( temp_dir , <EOL> self . node . uuid , <EOL> '<STR_LIT>' ) , <EOL> image_path ) <EOL> @ mock . patch . object ( ironic_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_destroy_images ( self , mock_cache , mock_rmtree , mock_unlink ) : <EOL> self . config ( images_path = '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> iscsi_deploy . destroy_images ( '<STR_LIT>' ) <EOL> mock_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> mock_unlink . assert_called_once_with ( '<STR_LIT>' ) <EOL> mock_rmtree . assert_called_once_with ( '<STR_LIT>' ) <EOL> def _test_build_deploy_ramdisk_options ( self , mock_alnum , api_url , <EOL> expected_root_device = None , <EOL> expected_boot_option = '<STR_LIT>' , <EOL> expected_boot_mode = '<STR_LIT>' ) : <EOL> fake_key = '<STR_LIT>' <EOL> fake_disk = '<STR_LIT>' <EOL> self . config ( disk_devices = fake_disk , group = '<STR_LIT>' ) <EOL> mock_alnum . return_value = fake_key <EOL> expected_iqn = '<STR_LIT>' % self . node . uuid <EOL> expected_opts = { <EOL> '<STR_LIT>' : expected_iqn , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : self . node . uuid , <EOL> '<STR_LIT>' : fake_key , <EOL> '<STR_LIT>' : fake_disk , <EOL> '<STR_LIT>' : api_url , <EOL> '<STR_LIT>' : expected_boot_option , <EOL> '<STR_LIT>' : expected_boot_mode , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> if expected_root_device : <EOL> expected_opts [ '<STR_LIT>' ] = expected_root_device <EOL> opts = iscsi_deploy . build_deploy_ramdisk_options ( self . node ) <EOL> self . assertEqual ( expected_opts , opts ) <EOL> mock_alnum . assert_called_once_with ( <NUM_LIT:32> ) <EOL> self . assertIn ( '<STR_LIT>' , self . node . instance_info ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> def test_build_deploy_ramdisk_options ( self , mock_alnum , mock_get_url ) : <EOL> fake_api_url = '<STR_LIT>' <EOL> self . config ( api_url = fake_api_url , group = '<STR_LIT>' ) <EOL> self . _test_build_deploy_ramdisk_options ( mock_alnum , fake_api_url ) <EOL> self . assertFalse ( mock_get_url . called ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> def test_build_deploy_ramdisk_options_keystone ( self , mock_alnum , <EOL> mock_get_url ) : <EOL> fake_api_url = '<STR_LIT>' <EOL> mock_get_url . return_value = fake_api_url <EOL> self . _test_build_deploy_ramdisk_options ( mock_alnum , fake_api_url ) <EOL> mock_get_url . assert_called_once_with ( ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> def test_build_deploy_ramdisk_options_root_device ( self , mock_alnum , <EOL> mock_get_url ) : <EOL> self . node . properties [ '<STR_LIT>' ] = { '<STR_LIT>' : <NUM_LIT> } <EOL> expected = '<STR_LIT>' <EOL> fake_api_url = '<STR_LIT>' <EOL> self . config ( api_url = fake_api_url , group = '<STR_LIT>' ) <EOL> self . _test_build_deploy_ramdisk_options ( mock_alnum , fake_api_url , <EOL> expected_root_device = expected ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> def test_build_deploy_ramdisk_options_boot_option ( self , mock_alnum , <EOL> mock_get_url ) : <EOL> self . node . instance_info = { '<STR_LIT>' : '<STR_LIT>' } <EOL> expected = '<STR_LIT>' <EOL> fake_api_url = '<STR_LIT>' <EOL> self . config ( api_url = fake_api_url , group = '<STR_LIT>' ) <EOL> self . _test_build_deploy_ramdisk_options ( mock_alnum , fake_api_url , <EOL> expected_boot_option = expected ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' , autospec = True ) <EOL> def test_build_deploy_ramdisk_options_whole_disk_image ( self , mock_alnum , <EOL> mock_get_url ) : <EOL> """<STR_LIT>""" <EOL> self . node . instance_info = { '<STR_LIT>' : '<STR_LIT>' } <EOL> dii = self . node . driver_internal_info <EOL> dii [ '<STR_LIT>' ] = True <EOL> self . node . driver_internal_info = dii <EOL> self . node . save ( ) <EOL> expected = '<STR_LIT>' <EOL> fake_api_url = '<STR_LIT>' <EOL> self . config ( api_url = fake_api_url , group = '<STR_LIT>' ) <EOL> self . _test_build_deploy_ramdisk_options ( mock_alnum , fake_api_url , <EOL> expected_boot_option = expected ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_fail ( self , deploy_mock , power_mock , <EOL> mock_image_cache , mock_disk_layout ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> deploy_mock . side_effect = iter ( [ <EOL> exception . InstanceDeployFailure ( "<STR_LIT>" ) ] ) <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> params = iscsi_deploy . get_deploy_info ( task . node , ** kwargs ) <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . continue_deploy , <EOL> task , ** kwargs ) <EOL> self . assertEqual ( states . DEPLOYFAIL , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> self . assertIsNotNone ( task . node . last_error ) <EOL> deploy_mock . assert_called_once_with ( ** params ) <EOL> power_mock . assert_called_once_with ( task , states . POWER_OFF ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertFalse ( mock_disk_layout . called ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_ramdisk_fails ( self , deploy_mock , power_mock , <EOL> mock_image_cache , mock_disk_layout ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT:error>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . continue_deploy , <EOL> task , ** kwargs ) <EOL> self . assertIsNotNone ( task . node . last_error ) <EOL> self . assertEqual ( states . DEPLOYFAIL , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> power_mock . assert_called_once_with ( task , states . POWER_OFF ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertFalse ( deploy_mock . called ) <EOL> self . assertFalse ( mock_disk_layout . called ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_fail_no_root_uuid_or_disk_id ( <EOL> self , deploy_mock , power_mock , mock_image_cache , mock_disk_layout ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> deploy_mock . return_value = { } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> params = iscsi_deploy . get_deploy_info ( task . node , ** kwargs ) <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . continue_deploy , <EOL> task , ** kwargs ) <EOL> self . assertEqual ( states . DEPLOYFAIL , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> self . assertIsNotNone ( task . node . last_error ) <EOL> deploy_mock . assert_called_once_with ( ** params ) <EOL> power_mock . assert_called_once_with ( task , states . POWER_OFF ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertFalse ( mock_disk_layout . called ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_fail_empty_root_uuid ( <EOL> self , deploy_mock , power_mock , mock_image_cache , mock_disk_layout ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> deploy_mock . return_value = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> params = iscsi_deploy . get_deploy_info ( task . node , ** kwargs ) <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . continue_deploy , <EOL> task , ** kwargs ) <EOL> self . assertEqual ( states . DEPLOYFAIL , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> self . assertIsNotNone ( task . node . last_error ) <EOL> deploy_mock . assert_called_once_with ( ** params ) <EOL> power_mock . assert_called_once_with ( task , states . POWER_OFF ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertFalse ( mock_disk_layout . called ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy ( self , deploy_mock , power_mock , mock_image_cache , <EOL> mock_deploy_info , mock_log , mock_disk_layout ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> mock_deploy_info . return_value = { <EOL> '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : ( u'<STR_LIT>' <EOL> u'<STR_LIT>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> log_params = mock_deploy_info . return_value . copy ( ) <EOL> log_params [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> expected_dict = { <EOL> '<STR_LIT>' : self . node . uuid , <EOL> '<STR_LIT>' : log_params , <EOL> } <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> deploy_mock . return_value = uuid_dict_returned <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> mock_log . isEnabledFor . return_value = True <EOL> retval = iscsi_deploy . continue_deploy ( task , ** kwargs ) <EOL> mock_log . debug . assert_called_once_with ( <EOL> mock . ANY , expected_dict ) <EOL> self . assertEqual ( states . DEPLOYWAIT , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> self . assertIsNone ( task . node . last_error ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertEqual ( uuid_dict_returned , retval ) <EOL> mock_disk_layout . assert_called_once_with ( task . node , mock . ANY ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_whole_disk_image ( <EOL> self , deploy_mock , power_mock , mock_image_cache , mock_deploy_info , <EOL> mock_log ) : <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> mock_deploy_info . return_value = { <EOL> '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( u'<STR_LIT>' <EOL> u'<STR_LIT>' ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:1>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> } <EOL> log_params = mock_deploy_info . return_value . copy ( ) <EOL> expected_dict = { <EOL> '<STR_LIT>' : self . node . uuid , <EOL> '<STR_LIT>' : log_params , <EOL> } <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> deploy_mock . return_value = uuid_dict_returned <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . driver_internal_info [ '<STR_LIT>' ] = True <EOL> mock_log . isEnabledFor . return_value = True <EOL> retval = iscsi_deploy . continue_deploy ( task , ** kwargs ) <EOL> mock_log . debug . assert_called_once_with ( <EOL> mock . ANY , expected_dict ) <EOL> self . assertEqual ( states . DEPLOYWAIT , task . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , task . node . target_provision_state ) <EOL> self . assertIsNone ( task . node . last_error ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> self . assertEqual ( uuid_dict_returned , retval ) <EOL> def _test_get_deploy_info ( self , extra_instance_info = None ) : <EOL> if extra_instance_info is None : <EOL> extra_instance_info = { } <EOL> instance_info = self . node . instance_info <EOL> instance_info [ '<STR_LIT>' ] = '<STR_LIT:key>' <EOL> instance_info . update ( extra_instance_info ) <EOL> self . node . instance_info = instance_info <EOL> kwargs = { '<STR_LIT:address>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT:key>' } <EOL> ret_val = iscsi_deploy . get_deploy_info ( self . node , ** kwargs ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT:address>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT>' ] ) <EOL> return ret_val <EOL> def test_get_deploy_info_boot_option_default ( self ) : <EOL> ret_val = self . _test_get_deploy_info ( ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT>' ] ) <EOL> def test_get_deploy_info_netboot_specified ( self ) : <EOL> capabilities = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> ret_val = self . _test_get_deploy_info ( extra_instance_info = capabilities ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT>' ] ) <EOL> def test_get_deploy_info_localboot ( self ) : <EOL> capabilities = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> ret_val = self . _test_get_deploy_info ( extra_instance_info = capabilities ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT>' ] ) <EOL> def test_get_deploy_info_disk_label ( self ) : <EOL> capabilities = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> ret_val = self . _test_get_deploy_info ( extra_instance_info = capabilities ) <EOL> self . assertEqual ( '<STR_LIT>' , ret_val [ '<STR_LIT>' ] ) <EOL> def test_get_deploy_info_not_specified ( self ) : <EOL> ret_val = self . _test_get_deploy_info ( ) <EOL> self . assertNotIn ( '<STR_LIT>' , ret_val ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_do_agent_iscsi_deploy_okay ( self , build_options_mock , <EOL> continue_deploy_mock ) : <EOL> build_options_mock . return_value = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> agent_client_mock = mock . MagicMock ( spec_set = agent_client . AgentClient ) <EOL> agent_client_mock . start_iscsi_target . return_value = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : None } <EOL> driver_internal_info = { '<STR_LIT>' : '<STR_LIT>' } <EOL> self . node . driver_internal_info = driver_internal_info <EOL> self . node . save ( ) <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> continue_deploy_mock . return_value = uuid_dict_returned <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> ret_val = iscsi_deploy . do_agent_iscsi_deploy ( <EOL> task , agent_client_mock ) <EOL> build_options_mock . assert_called_once_with ( task . node ) <EOL> agent_client_mock . start_iscsi_target . assert_called_once_with ( <EOL> task . node , '<STR_LIT>' , <NUM_LIT> , wipe_disk_metadata = True ) <EOL> continue_deploy_mock . assert_called_once_with ( <EOL> task , error = None , iqn = '<STR_LIT>' , key = '<STR_LIT>' , <EOL> address = '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> '<STR_LIT>' , <EOL> task . node . driver_internal_info [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( ret_val , uuid_dict_returned ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_do_agent_iscsi_deploy_preserve_ephemeral ( self , build_options_mock , <EOL> continue_deploy_mock ) : <EOL> """<STR_LIT>""" <EOL> build_options_mock . return_value = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> agent_client_mock = mock . MagicMock ( spec_set = agent_client . AgentClient ) <EOL> agent_client_mock . start_iscsi_target . return_value = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : None } <EOL> driver_internal_info = { <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . node . driver_internal_info = driver_internal_info <EOL> self . node . save ( ) <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> continue_deploy_mock . return_value = uuid_dict_returned <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = True <EOL> iscsi_deploy . do_agent_iscsi_deploy ( <EOL> task , agent_client_mock ) <EOL> build_options_mock . assert_called_once_with ( task . node ) <EOL> agent_client_mock . start_iscsi_target . assert_called_once_with ( <EOL> task . node , '<STR_LIT>' , <NUM_LIT> , wipe_disk_metadata = False ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_do_agent_iscsi_deploy_start_iscsi_failure ( self , <EOL> build_options_mock ) : <EOL> build_options_mock . return_value = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } <EOL> agent_client_mock = mock . MagicMock ( spec_set = agent_client . AgentClient ) <EOL> agent_client_mock . start_iscsi_target . return_value = { <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYING <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . do_agent_iscsi_deploy , <EOL> task , agent_client_mock ) <EOL> build_options_mock . assert_called_once_with ( task . node ) <EOL> agent_client_mock . start_iscsi_target . assert_called_once_with ( <EOL> task . node , '<STR_LIT>' , <NUM_LIT> , wipe_disk_metadata = True ) <EOL> self . node . refresh ( ) <EOL> self . assertEqual ( states . DEPLOYFAIL , self . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , self . node . target_provision_state ) <EOL> self . assertIsNotNone ( self . node . last_error ) <EOL> def test_validate_pass_bootloader_info_input ( self ) : <EOL> params = { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT:error>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> iscsi_deploy . validate_pass_bootloader_info_input ( task , params ) <EOL> def test_validate_pass_bootloader_info_missing_status ( self ) : <EOL> params = { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:address>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> self . assertRaises ( exception . MissingParameterValue , <EOL> iscsi_deploy . validate_pass_bootloader_info_input , <EOL> task , params ) <EOL> def test_validate_pass_bootloader_info_missing_key ( self ) : <EOL> params = { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT:address>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> self . assertRaises ( exception . MissingParameterValue , <EOL> iscsi_deploy . validate_pass_bootloader_info_input , <EOL> task , params ) <EOL> def test_validate_pass_bootloader_info_missing_address ( self ) : <EOL> params = { '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT:key>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> self . assertRaises ( exception . MissingParameterValue , <EOL> iscsi_deploy . validate_pass_bootloader_info_input , <EOL> task , params ) <EOL> def test_validate_pass_bootloader_info_input_invalid_key ( self ) : <EOL> params = { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertRaises ( exception . InvalidParameterValue , <EOL> iscsi_deploy . validate_pass_bootloader_info_input , <EOL> task , params ) <EOL> def test_validate_bootloader_install_status ( self ) : <EOL> kwargs = { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT:error>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> iscsi_deploy . validate_bootloader_install_status ( task , kwargs ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_validate_bootloader_install_status_install_failed ( <EOL> self , set_fail_state_mock ) : <EOL> kwargs = { '<STR_LIT:key>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' , '<STR_LIT:error>' : '<STR_LIT>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . node . provision_state = states . DEPLOYING <EOL> task . node . target_provision_state = states . ACTIVE <EOL> task . node . instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . validate_bootloader_install_status , <EOL> task , kwargs ) <EOL> set_fail_state_mock . assert_called_once_with ( task , mock . ANY ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_finish_deploy ( self , notify_mock ) : <EOL> self . node . provision_state = states . DEPLOYING <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> iscsi_deploy . finish_deploy ( task , '<STR_LIT>' ) <EOL> notify_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> self . assertEqual ( states . ACTIVE , task . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , task . node . target_provision_state ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_finish_deploy_notify_fails ( self , notify_mock , <EOL> set_fail_state_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> notify_mock . side_effect = RuntimeError ( ) <EOL> self . assertRaises ( exception . InstanceDeployFailure , <EOL> iscsi_deploy . finish_deploy , task , '<STR_LIT>' ) <EOL> set_fail_state_mock . assert_called_once_with ( task , mock . ANY ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_finish_deploy_ssh_with_local_boot ( self , notify_mock , <EOL> node_power_mock ) : <EOL> instance_info = dict ( INST_INFO_DICT ) <EOL> instance_info [ '<STR_LIT>' ] = { '<STR_LIT>' : '<STR_LIT>' } <EOL> n = { <EOL> '<STR_LIT>' : uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : instance_info , <EOL> '<STR_LIT>' : states . DEPLOYING , <EOL> '<STR_LIT>' : states . ACTIVE , <EOL> } <EOL> mgr_utils . mock_the_extension_manager ( driver = "<STR_LIT>" ) <EOL> node = obj_utils . create_test_node ( self . context , ** n ) <EOL> with task_manager . acquire ( self . context , node . uuid , <EOL> shared = False ) as task : <EOL> iscsi_deploy . finish_deploy ( task , '<STR_LIT>' ) <EOL> notify_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> self . assertEqual ( states . ACTIVE , task . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , task . node . target_provision_state ) <EOL> node_power_mock . assert_called_once_with ( task , states . REBOOT ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> def test_validate_good_api_url_from_config_file ( self , mock_ks ) : <EOL> mock_ks . side_effect = exception . KeystoneFailure <EOL> self . config ( group = '<STR_LIT>' , api_url = '<STR_LIT>' ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> iscsi_deploy . validate ( task ) <EOL> self . assertFalse ( mock_ks . called ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> def test_validate_good_api_url_from_keystone ( self , mock_ks ) : <EOL> mock_ks . return_value = '<STR_LIT>' <EOL> self . config ( group = '<STR_LIT>' , api_url = None ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> iscsi_deploy . validate ( task ) <EOL> mock_ks . assert_called_once_with ( ) <EOL> @ mock . patch . object ( keystone , '<STR_LIT>' , autospec = True ) <EOL> def test_validate_fail_no_api_url ( self , mock_ks ) : <EOL> mock_ks . side_effect = exception . KeystoneFailure <EOL> self . config ( group = '<STR_LIT>' , api_url = None ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> self . assertRaises ( exception . InvalidParameterValue , <EOL> iscsi_deploy . validate , task ) <EOL> mock_ks . assert_called_once_with ( ) <EOL> def test_validate_invalid_root_device_hints ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . node . properties [ '<STR_LIT>' ] = { '<STR_LIT:size>' : '<STR_LIT>' } <EOL> self . assertRaises ( exception . InvalidParameterValue , <EOL> iscsi_deploy . validate , task ) <EOL> class ISCSIDeployTestCase ( db_base . DbTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( ISCSIDeployTestCase , self ) . setUp ( ) <EOL> mgr_utils . mock_the_extension_manager ( driver = "<STR_LIT>" ) <EOL> self . driver = driver_factory . get_driver ( "<STR_LIT>" ) <EOL> self . driver . vendor = iscsi_deploy . VendorPassthru ( ) <EOL> self . node = obj_utils . create_test_node ( <EOL> self . context , driver = '<STR_LIT>' , <EOL> instance_info = INST_INFO_DICT , <EOL> driver_info = DRV_INFO_DICT , <EOL> driver_internal_info = DRV_INTERNAL_INFO_DICT , <EOL> ) <EOL> self . node . driver_internal_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . task = mock . MagicMock ( spec = task_manager . TaskManager ) <EOL> self . task . shared = False <EOL> self . task . node = self . node <EOL> self . task . driver = self . driver <EOL> self . task . context = self . context <EOL> dhcp_factory . DHCPFactory . _dhcp_provider = None <EOL> def test_get_properties ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> self . assertEqual ( { } , task . driver . deploy . get_properties ( ) ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( pxe . PXEBoot , '<STR_LIT>' , autospec = True ) <EOL> def test_validate ( self , pxe_validate_mock , <EOL> validate_capabilities_mock , validate_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . driver . deploy . validate ( task ) <EOL> pxe_validate_mock . assert_called_once_with ( task . driver . boot , task ) <EOL> validate_capabilities_mock . assert_called_once_with ( task . node ) <EOL> validate_mock . assert_called_once_with ( task ) <EOL> @ mock . patch . object ( pxe . PXEBoot , '<STR_LIT>' , autospec = True ) <EOL> def test_prepare_node_active ( self , prepare_instance_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . node . provision_state = states . ACTIVE <EOL> task . driver . deploy . prepare ( task ) <EOL> prepare_instance_mock . assert_called_once_with ( <EOL> task . driver . boot , task ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( pxe . PXEBoot , '<STR_LIT>' , autospec = True ) <EOL> def test_prepare_node_deploying ( self , mock_prepare_ramdisk , <EOL> mock_iscsi_options , mock_agent_options ) : <EOL> mock_iscsi_options . return_value = { '<STR_LIT:a>' : '<STR_LIT:b>' } <EOL> mock_agent_options . return_value = { '<STR_LIT:c>' : '<STR_LIT:d>' } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . node . provision_state = states . DEPLOYWAIT <EOL> task . driver . deploy . prepare ( task ) <EOL> mock_iscsi_options . assert_called_once_with ( task . node ) <EOL> mock_agent_options . assert_called_once_with ( task . node ) <EOL> mock_prepare_ramdisk . assert_called_once_with ( <EOL> task . driver . boot , task , { '<STR_LIT:a>' : '<STR_LIT:b>' , '<STR_LIT:c>' : '<STR_LIT:d>' } ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_deploy ( self , mock_cache_instance_image , <EOL> mock_check_image_size , mock_node_power_action ) : <EOL> with task_manager . acquire ( self . context , <EOL> self . node . uuid , shared = False ) as task : <EOL> state = task . driver . deploy . deploy ( task ) <EOL> self . assertEqual ( state , states . DEPLOYWAIT ) <EOL> mock_cache_instance_image . assert_called_once_with ( <EOL> self . context , task . node ) <EOL> mock_check_image_size . assert_called_once_with ( task ) <EOL> mock_node_power_action . assert_called_once_with ( task , states . REBOOT ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_tear_down ( self , node_power_action_mock ) : <EOL> with task_manager . acquire ( self . context , <EOL> self . node . uuid , shared = False ) as task : <EOL> state = task . driver . deploy . tear_down ( task ) <EOL> self . assertEqual ( state , states . DELETED ) <EOL> node_power_action_mock . assert_called_once_with ( task , <EOL> states . POWER_OFF ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( pxe . PXEBoot , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( pxe . PXEBoot , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_clean_up ( self , destroy_images_mock , clean_up_ramdisk_mock , <EOL> clean_up_instance_mock , clean_dhcp_mock , <EOL> set_dhcp_provider_mock ) : <EOL> with task_manager . acquire ( self . context , <EOL> self . node . uuid , shared = False ) as task : <EOL> task . driver . deploy . clean_up ( task ) <EOL> destroy_images_mock . assert_called_once_with ( task . node . uuid ) <EOL> clean_up_ramdisk_mock . assert_called_once_with ( <EOL> task . driver . boot , task ) <EOL> clean_up_instance_mock . assert_called_once_with ( <EOL> task . driver . boot , task ) <EOL> set_dhcp_provider_mock . assert_called_once_with ( ) <EOL> clean_dhcp_mock . assert_called_once_with ( task ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_prepare_cleaning ( self , prepare_inband_cleaning_mock ) : <EOL> prepare_inband_cleaning_mock . return_value = states . CLEANWAIT <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> self . assertEqual ( <EOL> states . CLEANWAIT , task . driver . deploy . prepare_cleaning ( task ) ) <EOL> prepare_inband_cleaning_mock . assert_called_once_with ( <EOL> task , manage_boot = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_tear_down_cleaning ( self , tear_down_cleaning_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . driver . deploy . tear_down_cleaning ( task ) <EOL> tear_down_cleaning_mock . assert_called_once_with ( <EOL> task , manage_boot = True ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_get_clean_steps ( self , mock_get_clean_steps ) : <EOL> self . config ( group = '<STR_LIT>' , erase_devices_priority = <NUM_LIT:10> ) <EOL> mock_steps = [ { '<STR_LIT>' : <NUM_LIT:10> , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . node . driver_internal_info = { '<STR_LIT>' : '<STR_LIT:foo>' } <EOL> self . node . save ( ) <EOL> mock_get_clean_steps . return_value = mock_steps <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> steps = task . driver . deploy . get_clean_steps ( task ) <EOL> mock_get_clean_steps . assert_called_once_with ( <EOL> task , interface = '<STR_LIT>' , <EOL> override_priorities = { <EOL> '<STR_LIT>' : <NUM_LIT:10> } ) <EOL> self . assertEqual ( mock_steps , steps ) <EOL> @ mock . patch ( '<STR_LIT>' , <EOL> autospec = True ) <EOL> def test_get_clean_steps_no_agent_url ( self , mock_get_clean_steps ) : <EOL> self . node . driver_internal_info = { } <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> steps = task . driver . deploy . get_clean_steps ( task ) <EOL> self . assertEqual ( [ ] , steps ) <EOL> self . assertFalse ( mock_get_clean_steps . called ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def test_execute_clean_step ( self , agent_execute_clean_step_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . driver . deploy . execute_clean_step ( <EOL> task , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> agent_execute_clean_step_mock . assert_called_once_with ( <EOL> task , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> class TestVendorPassthru ( db_base . DbTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestVendorPassthru , self ) . setUp ( ) <EOL> mgr_utils . mock_the_extension_manager ( ) <EOL> self . driver = driver_factory . get_driver ( "<STR_LIT>" ) <EOL> self . driver . vendor = iscsi_deploy . VendorPassthru ( ) <EOL> self . node = obj_utils . create_test_node ( <EOL> self . context , driver = '<STR_LIT>' , <EOL> instance_info = INST_INFO_DICT , <EOL> driver_info = DRV_INFO_DICT , <EOL> driver_internal_info = DRV_INTERNAL_INFO_DICT , <EOL> ) <EOL> self . node . driver_internal_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . task = mock . MagicMock ( spec = task_manager . TaskManager ) <EOL> self . task . shared = False <EOL> self . task . node = self . node <EOL> self . task . driver = self . driver <EOL> self . task . context = self . context <EOL> def test_validate_good ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . node . instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> task . driver . vendor . validate ( task , method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , iqn = '<STR_LIT>' , <EOL> key = '<STR_LIT>' ) <EOL> def test_validate_pass_deploy_info_during_cleaning ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . node . provision_state = states . CLEANWAIT <EOL> self . assertIsNone ( <EOL> task . driver . vendor . validate ( task , method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , iqn = '<STR_LIT>' , <EOL> key = '<STR_LIT>' ) ) <EOL> def test_validate_fail ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> self . assertRaises ( exception . InvalidParameterValue , <EOL> task . driver . vendor . validate , <EOL> task , method = '<STR_LIT>' , <EOL> key = '<STR_LIT>' ) <EOL> def test_validate_key_notmatch ( self ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> self . assertRaises ( exception . InvalidParameterValue , <EOL> task . driver . vendor . validate , <EOL> task , method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , iqn = '<STR_LIT>' , <EOL> key = '<STR_LIT>' ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , spec = [ '<STR_LIT>' ] ) <EOL> def test__initiate_cleaning ( self , log_mock , set_node_cleaning_steps_mock , <EOL> notify_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . driver . vendor . _initiate_cleaning ( task ) <EOL> log_mock . warning . assert_called_once_with ( mock . ANY , mock . ANY ) <EOL> set_node_cleaning_steps_mock . assert_called_once_with ( task ) <EOL> notify_mock . assert_called_once_with ( self . driver . vendor , task ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( manager_utils , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , spec = [ '<STR_LIT>' ] ) <EOL> def test__initiate_cleaning_exception ( <EOL> self , log_mock , set_node_cleaning_steps_mock , <EOL> cleaning_error_handler_mock , notify_mock ) : <EOL> set_node_cleaning_steps_mock . side_effect = RuntimeError ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . driver . vendor . _initiate_cleaning ( task ) <EOL> log_mock . warning . assert_called_once_with ( mock . ANY , mock . ANY ) <EOL> set_node_cleaning_steps_mock . assert_called_once_with ( task ) <EOL> cleaning_error_handler_mock . assert_called_once_with ( task , mock . ANY ) <EOL> self . assertFalse ( notify_mock . called ) <EOL> @ mock . patch . object ( fake . FakeBoot , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def _test_pass_deploy_info_deploy ( self , is_localboot , mock_deploy , <EOL> mock_image_cache , <EOL> notify_mock , <EOL> fakeboot_prepare_instance_mock ) : <EOL> i_info = self . node . instance_info <EOL> if is_localboot : <EOL> i_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> i_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . node . instance_info = i_info <EOL> self . node . power_state = states . POWER_ON <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> root_uuid = "<STR_LIT>" <EOL> mock_deploy . return_value = { '<STR_LIT>' : root_uuid } <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . driver . vendor . pass_deploy_info ( <EOL> task , address = '<STR_LIT>' , iqn = '<STR_LIT>' , key = '<STR_LIT>' ) <EOL> self . node . refresh ( ) <EOL> self . assertEqual ( states . POWER_ON , self . node . power_state ) <EOL> self . assertIn ( '<STR_LIT>' , self . node . driver_internal_info ) <EOL> self . assertIsNone ( self . node . last_error ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> notify_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> fakeboot_prepare_instance_mock . assert_called_once_with ( mock . ANY , task ) <EOL> @ mock . patch . object ( fake . FakeBoot , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( deploy_utils , '<STR_LIT>' , autospec = True ) <EOL> def _test_pass_deploy_info_whole_disk_image ( self , is_localboot , <EOL> mock_deploy , <EOL> mock_image_cache , <EOL> notify_mock , <EOL> fakeboot_prep_inst_mock ) : <EOL> i_info = self . node . instance_info <EOL> if is_localboot : <EOL> i_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> i_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . node . instance_info = i_info <EOL> self . node . power_state = states . POWER_ON <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> disk_id = '<STR_LIT>' <EOL> mock_deploy . return_value = { '<STR_LIT>' : disk_id } <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . node . driver_internal_info [ '<STR_LIT>' ] = True <EOL> task . driver . vendor . pass_deploy_info ( task , address = '<STR_LIT>' , <EOL> iqn = '<STR_LIT>' , <EOL> key = '<STR_LIT>' ) <EOL> self . node . refresh ( ) <EOL> self . assertEqual ( states . POWER_ON , self . node . power_state ) <EOL> self . assertIsNone ( self . node . last_error ) <EOL> mock_image_cache . assert_called_once_with ( ) <EOL> mock_image_cache . return_value . clean_up . assert_called_once_with ( ) <EOL> notify_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> fakeboot_prep_inst_mock . assert_called_once_with ( mock . ANY , task ) <EOL> def test_pass_deploy_info_deploy ( self ) : <EOL> self . _test_pass_deploy_info_deploy ( False ) <EOL> self . assertEqual ( states . ACTIVE , self . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , self . node . target_provision_state ) <EOL> def test_pass_deploy_info_localboot ( self ) : <EOL> self . _test_pass_deploy_info_deploy ( True ) <EOL> self . assertEqual ( states . DEPLOYWAIT , self . node . provision_state ) <EOL> self . assertEqual ( states . ACTIVE , self . node . target_provision_state ) <EOL> def test_pass_deploy_info_whole_disk_image ( self ) : <EOL> self . _test_pass_deploy_info_whole_disk_image ( False ) <EOL> self . assertEqual ( states . ACTIVE , self . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , self . node . target_provision_state ) <EOL> def test_pass_deploy_info_whole_disk_image_localboot ( self ) : <EOL> self . _test_pass_deploy_info_whole_disk_image ( True ) <EOL> self . assertEqual ( states . ACTIVE , self . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , self . node . target_provision_state ) <EOL> def test_pass_deploy_info_invalid ( self ) : <EOL> self . node . power_state = states . POWER_ON <EOL> self . node . provision_state = states . AVAILABLE <EOL> self . node . target_provision_state = states . NOSTATE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> self . assertRaises ( exception . InvalidState , <EOL> task . driver . vendor . pass_deploy_info , <EOL> task , address = '<STR_LIT>' , iqn = '<STR_LIT>' , <EOL> key = '<STR_LIT>' , error = '<STR_LIT>' ) <EOL> self . node . refresh ( ) <EOL> self . assertEqual ( states . AVAILABLE , self . node . provision_state ) <EOL> self . assertEqual ( states . NOSTATE , self . node . target_provision_state ) <EOL> self . assertEqual ( states . POWER_ON , self . node . power_state ) <EOL> @ mock . patch . object ( iscsi_deploy . VendorPassthru , '<STR_LIT>' ) <EOL> def test_pass_deploy_info_lock_elevated ( self , mock_deploy_info ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . driver . vendor . pass_deploy_info ( <EOL> task , address = '<STR_LIT>' , iqn = '<STR_LIT>' , key = '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_deploy_info . call_count , <EOL> "<STR_LIT>" ) <EOL> @ mock . patch . object ( iscsi_deploy . VendorPassthru , <EOL> '<STR_LIT>' , autospec = True ) <EOL> def test_pass_deploy_info_cleaning ( self , initiate_cleaning_mock ) : <EOL> with task_manager . acquire ( self . context , self . node . uuid ) as task : <EOL> task . node . provision_state = states . CLEANWAIT <EOL> task . driver . vendor . pass_deploy_info ( <EOL> task , address = '<STR_LIT>' , iqn = '<STR_LIT>' , key = '<STR_LIT>' ) <EOL> initiate_cleaning_mock . assert_called_once_with ( <EOL> task . driver . vendor , task ) <EOL> self . assertEqual ( states . CLEANWAIT , task . node . provision_state ) <EOL> def test_vendor_routes ( self ) : <EOL> expected = [ '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> vendor_routes = task . driver . vendor . vendor_routes <EOL> self . assertIsInstance ( vendor_routes , dict ) <EOL> self . assertEqual ( sorted ( expected ) , sorted ( list ( vendor_routes ) ) ) <EOL> def test_driver_routes ( self ) : <EOL> expected = [ '<STR_LIT>' ] <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> driver_routes = task . driver . vendor . driver_routes <EOL> self . assertIsInstance ( driver_routes , dict ) <EOL> self . assertEqual ( sorted ( expected ) , sorted ( list ( driver_routes ) ) ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , <EOL> autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_pass_bootloader_install_info ( self , finish_deploy_mock , <EOL> validate_input_mock ) : <EOL> kwargs = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:address>' : '<STR_LIT>' } <EOL> self . node . provision_state = states . DEPLOYWAIT <EOL> self . node . target_provision_state = states . ACTIVE <EOL> self . node . save ( ) <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = False ) as task : <EOL> task . driver . vendor . pass_bootloader_install_info ( task , ** kwargs ) <EOL> finish_deploy_mock . assert_called_once_with ( task , '<STR_LIT>' ) <EOL> validate_input_mock . assert_called_once_with ( task , kwargs ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_netboot ( self , do_agent_iscsi_deploy_mock , <EOL> reboot_and_finish_deploy_mock ) : <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> do_agent_iscsi_deploy_mock . return_value = uuid_dict_returned <EOL> self . driver . vendor . continue_deploy ( self . task ) <EOL> do_agent_iscsi_deploy_mock . assert_called_once_with ( <EOL> self . task , self . driver . vendor . _client ) <EOL> reboot_and_finish_deploy_mock . assert_called_once_with ( <EOL> mock . ANY , self . task ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_localboot ( self , do_agent_iscsi_deploy_mock , <EOL> configure_local_boot_mock , <EOL> reboot_and_finish_deploy_mock ) : <EOL> self . node . instance_info = { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> self . node . save ( ) <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' } <EOL> do_agent_iscsi_deploy_mock . return_value = uuid_dict_returned <EOL> self . driver . vendor . continue_deploy ( self . task ) <EOL> do_agent_iscsi_deploy_mock . assert_called_once_with ( <EOL> self . task , self . driver . vendor . _client ) <EOL> configure_local_boot_mock . assert_called_once_with ( <EOL> self . task . driver . vendor , self . task , root_uuid = '<STR_LIT>' , <EOL> efi_system_part_uuid = None ) <EOL> reboot_and_finish_deploy_mock . assert_called_once_with ( <EOL> self . task . driver . vendor , self . task ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( agent_base_vendor . BaseAgentVendor , <EOL> '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( iscsi_deploy , '<STR_LIT>' , autospec = True ) <EOL> def test_continue_deploy_localboot_uefi ( self , do_agent_iscsi_deploy_mock , <EOL> configure_local_boot_mock , <EOL> reboot_and_finish_deploy_mock ) : <EOL> self . node . instance_info = { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> self . node . save ( ) <EOL> uuid_dict_returned = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> do_agent_iscsi_deploy_mock . return_value = uuid_dict_returned <EOL> self . driver . vendor . continue_deploy ( self . task ) <EOL> do_agent_iscsi_deploy_mock . assert_called_once_with ( <EOL> self . task , self . driver . vendor . _client ) <EOL> configure_local_boot_mock . assert_called_once_with ( <EOL> self . task . driver . vendor , self . task , root_uuid = '<STR_LIT>' , <EOL> efi_system_part_uuid = '<STR_LIT>' ) <EOL> reboot_and_finish_deploy_mock . assert_called_once_with ( <EOL> self . task . driver . vendor , self . task ) <EOL> class CleanUpFullFlowTestCase ( db_base . DbTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( CleanUpFullFlowTestCase , self ) . setUp ( ) <EOL> self . config ( image_cache_size = <NUM_LIT:0> , group = '<STR_LIT>' ) <EOL> mgr_utils . mock_the_extension_manager ( driver = "<STR_LIT>" ) <EOL> instance_info = INST_INFO_DICT <EOL> instance_info [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . node = obj_utils . create_test_node ( <EOL> self . context , driver = '<STR_LIT>' , <EOL> instance_info = instance_info , <EOL> driver_info = DRV_INFO_DICT , <EOL> driver_internal_info = DRV_INTERNAL_INFO_DICT , <EOL> ) <EOL> self . port = obj_utils . create_test_port ( self . context , <EOL> node_id = self . node . id ) <EOL> pxe_temp_dir = tempfile . mkdtemp ( ) <EOL> self . config ( tftp_root = pxe_temp_dir , group = '<STR_LIT>' ) <EOL> tftp_master_dir = os . path . join ( CONF . pxe . tftp_root , <EOL> '<STR_LIT>' ) <EOL> self . config ( tftp_master_path = tftp_master_dir , group = '<STR_LIT>' ) <EOL> os . makedirs ( tftp_master_dir ) <EOL> instance_temp_dir = tempfile . mkdtemp ( ) <EOL> self . config ( images_path = instance_temp_dir , <EOL> group = '<STR_LIT>' ) <EOL> instance_master_dir = os . path . join ( CONF . pxe . images_path , <EOL> '<STR_LIT>' ) <EOL> self . config ( instance_master_path = instance_master_dir , <EOL> group = '<STR_LIT>' ) <EOL> os . makedirs ( instance_master_dir ) <EOL> self . pxe_config_dir = os . path . join ( CONF . pxe . tftp_root , '<STR_LIT>' ) <EOL> os . makedirs ( self . pxe_config_dir ) <EOL> self . master_kernel_path = os . path . join ( CONF . pxe . tftp_master_path , <EOL> '<STR_LIT>' ) <EOL> self . master_instance_path = os . path . join ( CONF . pxe . instance_master_path , <EOL> '<STR_LIT>' ) <EOL> self . node_tftp_dir = os . path . join ( CONF . pxe . tftp_root , <EOL> self . node . uuid ) <EOL> os . makedirs ( self . node_tftp_dir ) <EOL> self . kernel_path = os . path . join ( self . node_tftp_dir , <EOL> '<STR_LIT>' ) <EOL> self . node_image_dir = iscsi_deploy . _get_image_dir_path ( self . node . uuid ) <EOL> os . makedirs ( self . node_image_dir ) <EOL> self . image_path = iscsi_deploy . _get_image_file_path ( self . node . uuid ) <EOL> self . config_path = pxe_utils . get_pxe_config_file_path ( self . node . uuid ) <EOL> self . mac_path = pxe_utils . _get_pxe_mac_path ( self . port . address ) <EOL> self . files = [ self . config_path , self . master_kernel_path , <EOL> self . master_instance_path ] <EOL> for fname in self . files : <EOL> with open ( fname , '<STR_LIT:w>' ) as fp : <EOL> fp . write ( '<STR_LIT:test>' ) <EOL> os . link ( self . config_path , self . mac_path ) <EOL> os . link ( self . master_kernel_path , self . kernel_path ) <EOL> os . link ( self . master_instance_path , self . image_path ) <EOL> dhcp_factory . DHCPFactory . _dhcp_provider = None <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( pxe , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( pxe , '<STR_LIT>' , autospec = True ) <EOL> def test_clean_up_with_master ( self , mock_get_deploy_image_info , <EOL> mock_get_instance_image_info , <EOL> clean_dhcp_mock , set_dhcp_provider_mock ) : <EOL> image_info = { '<STR_LIT>' : ( '<STR_LIT>' , <EOL> self . kernel_path ) } <EOL> mock_get_instance_image_info . return_value = image_info <EOL> mock_get_deploy_image_info . return_value = { } <EOL> with task_manager . acquire ( self . context , self . node . uuid , <EOL> shared = True ) as task : <EOL> task . driver . deploy . clean_up ( task ) <EOL> mock_get_instance_image_info . assert_called_with ( task . node , <EOL> task . context ) <EOL> mock_get_deploy_image_info . assert_called_with ( task . node ) <EOL> set_dhcp_provider_mock . assert_called_once_with ( ) <EOL> clean_dhcp_mock . assert_called_once_with ( task ) <EOL> for path in ( [ self . kernel_path , self . image_path , self . config_path ] <EOL> + self . files ) : <EOL> self . assertFalse ( os . path . exists ( path ) , <EOL> '<STR_LIT>' % path ) </s>
