<s> import mock <EOL> from rally import exceptions <EOL> from rally . plugins . openstack . scenarios . vm import vmtasks <EOL> from tests . unit import test <EOL> class VMTasksTestCase ( test . ScenarioTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( VMTasksTestCase , self ) . setUp ( ) <EOL> self . context . update ( { "<STR_LIT:user>" : { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : mock . MagicMock ( ) } } ) <EOL> self . scenario = vmtasks . VMTasks ( context = self . context ) <EOL> self . ip = { "<STR_LIT:id>" : "<STR_LIT>" , "<STR_LIT>" : "<STR_LIT>" , "<STR_LIT>" : True } <EOL> self . scenario . _boot_server_with_fip = mock . Mock ( <EOL> return_value = ( "<STR_LIT>" , self . ip ) ) <EOL> self . scenario . _wait_for_ping = mock . Mock ( ) <EOL> self . scenario . _delete_server_with_fip = mock . Mock ( ) <EOL> self . scenario . _create_volume = mock . Mock ( <EOL> return_value = mock . Mock ( id = "<STR_LIT>" ) ) <EOL> self . scenario . _run_command = mock . MagicMock ( <EOL> return_value = ( <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> def test_boot_runcommand_delete ( self ) : <EOL> self . scenario . boot_runcommand_delete ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> command = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> username = "<STR_LIT>" , <EOL> password = "<STR_LIT>" , <EOL> use_floating_ip = "<STR_LIT>" , <EOL> floating_network = "<STR_LIT>" , <EOL> force_delete = "<STR_LIT>" , <EOL> volume_args = { "<STR_LIT:size>" : <NUM_LIT:16> } , <EOL> foo_arg = "<STR_LIT>" ) <EOL> self . scenario . _create_volume . assert_called_once_with ( <EOL> <NUM_LIT:16> , imageRef = None ) <EOL> self . scenario . _boot_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , "<STR_LIT>" , key_name = "<STR_LIT>" , <EOL> use_floating_ip = "<STR_LIT>" , floating_network = "<STR_LIT>" , <EOL> block_device_mapping = { "<STR_LIT>" : "<STR_LIT>" } , <EOL> foo_arg = "<STR_LIT>" ) <EOL> self . scenario . _wait_for_ping . assert_called_once_with ( "<STR_LIT>" ) <EOL> self . scenario . _run_command . assert_called_once_with ( <EOL> "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , <EOL> command = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = "<STR_LIT>" ) <EOL> def test_boot_runcommand_delete_command ( self ) : <EOL> self . scenario . boot_runcommand_delete ( <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> command = { "<STR_LIT>" : "<STR_LIT:foo>" } , <EOL> username = "<STR_LIT>" , <EOL> password = "<STR_LIT>" , <EOL> use_floating_ip = "<STR_LIT>" , <EOL> floating_network = "<STR_LIT>" , <EOL> force_delete = "<STR_LIT>" , <EOL> volume_args = { "<STR_LIT:size>" : <NUM_LIT:16> } , <EOL> foo_arg = "<STR_LIT>" ) <EOL> self . scenario . _create_volume . assert_called_once_with ( <EOL> <NUM_LIT:16> , imageRef = None ) <EOL> self . scenario . _boot_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , "<STR_LIT>" , key_name = "<STR_LIT>" , <EOL> use_floating_ip = "<STR_LIT>" , floating_network = "<STR_LIT>" , <EOL> block_device_mapping = { "<STR_LIT>" : "<STR_LIT>" } , <EOL> foo_arg = "<STR_LIT>" ) <EOL> self . scenario . _run_command . assert_called_once_with ( <EOL> "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" , "<STR_LIT>" , <EOL> command = { "<STR_LIT>" : "<STR_LIT:foo>" } ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = "<STR_LIT>" ) <EOL> def test_boot_runcommand_delete_script_fails ( self ) : <EOL> self . scenario . _run_command = mock . MagicMock ( <EOL> return_value = ( <NUM_LIT:1> , "<STR_LIT>" , "<STR_LIT>" ) ) <EOL> self . assertRaises ( exceptions . ScriptError , <EOL> self . scenario . boot_runcommand_delete , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = False ) <EOL> def test_boot_runcommand_delete_command_timeouts ( self ) : <EOL> self . scenario . _run_command . side_effect = exceptions . SSHTimeout ( ) <EOL> self . assertRaises ( exceptions . SSHTimeout , <EOL> self . scenario . boot_runcommand_delete , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = False ) <EOL> def test_boot_runcommand_delete_ping_wait_timeouts ( self ) : <EOL> self . scenario . _wait_for_ping . side_effect = exceptions . TimeoutException ( <EOL> resource_type = "<STR_LIT>" , <EOL> resource_name = "<STR_LIT>" , <EOL> resource_id = "<STR_LIT>" , <EOL> desired_status = "<STR_LIT>" , <EOL> resource_status = "<STR_LIT>" ) <EOL> exc = self . assertRaises ( exceptions . TimeoutException , <EOL> self . scenario . boot_runcommand_delete , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> wait_for_ping = True ) <EOL> self . assertEqual ( exc . kwargs [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertEqual ( exc . kwargs [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertEqual ( exc . kwargs [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertEqual ( exc . kwargs [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . assertEqual ( exc . kwargs [ "<STR_LIT>" ] , "<STR_LIT>" ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = False ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_boot_runcommand_delete_json_fails ( self , mock_json ) : <EOL> mock_json . loads . side_effect = ValueError ( ) <EOL> self . assertRaises ( exceptions . ScriptError , <EOL> self . scenario . boot_runcommand_delete , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . scenario . _delete_server_with_fip . assert_called_once_with ( <EOL> "<STR_LIT>" , self . ip , force_delete = False ) <EOL> def test_boot_runcommand_delete_custom_image ( self ) : <EOL> context = { <EOL> "<STR_LIT:user>" : { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : mock . Mock ( ) <EOL> } , <EOL> "<STR_LIT>" : { <EOL> "<STR_LIT>" : { "<STR_LIT:id>" : "<STR_LIT>" } <EOL> } <EOL> } <EOL> scenario = vmtasks . VMTasks ( context ) <EOL> scenario . boot_runcommand_delete = mock . Mock ( ) <EOL> scenario . boot_runcommand_delete_custom_image ( <EOL> flavor = "<STR_LIT>" , <EOL> command = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> username = "<STR_LIT:username>" ) <EOL> scenario . boot_runcommand_delete . assert_called_once_with ( <EOL> image = "<STR_LIT>" , flavor = "<STR_LIT>" , username = "<STR_LIT:username>" , <EOL> command = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_runcommand_heat ( self , mock_sshutils , mock_heat ) : <EOL> fake_ssh = mock . Mock ( ) <EOL> fake_ssh . execute . return_value = [ <NUM_LIT:0> , "<STR_LIT>" , "<STR_LIT>" ] <EOL> mock_sshutils . SSH . return_value = fake_ssh <EOL> fake_stack = mock . Mock ( ) <EOL> fake_stack . stack . outputs = [ { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : "<STR_LIT>" } ] <EOL> mock_heat . main . Stack . return_value = fake_stack <EOL> context = { <EOL> "<STR_LIT:user>" : { "<STR_LIT>" : { "<STR_LIT:name>" : "<STR_LIT:name>" , "<STR_LIT>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : { "<STR_LIT>" : [ { "<STR_LIT>" : "<STR_LIT:1>" } ] } <EOL> } <EOL> scenario = vmtasks . VMTasks ( context ) <EOL> scenario . generate_random_name = mock . Mock ( return_value = "<STR_LIT:name>" ) <EOL> scenario . add_output = mock . Mock ( ) <EOL> workload = { "<STR_LIT:username>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT:foo>" , "<STR_LIT:bar>" ] } <EOL> scenario . runcommand_heat ( workload , "<STR_LIT>" , <EOL> { "<STR_LIT>" : "<STR_LIT>" } , <EOL> { "<STR_LIT>" : "<STR_LIT>" } ) <EOL> expected = { "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT:data>" : { "<STR_LIT>" : [ [ "<STR_LIT:key>" , "<STR_LIT>" ] ] , <EOL> "<STR_LIT>" : [ "<STR_LIT:key>" , "<STR_LIT:value>" ] } , <EOL> "<STR_LIT:description>" : "<STR_LIT>" , <EOL> "<STR_LIT:title>" : "<STR_LIT>" } <EOL> scenario . add_output . assert_called_once_with ( complete = expected ) </s>
