diff --git a/nova/tests/compute/test_compute.py b/nova/tests/compute/test_compute.py index 0d9e88d70a..a4136ec084 100644 --- a/nova/tests/compute/test_compute.py +++ b/nova/tests/compute/test_compute.py @@ -2688,122 +2688,6 @@ class ComputeTestCase(BaseTestCase): instance.uuid) self.assertEqual(expected_block_device_info, block_device_info) - def test_set_admin_password(self): - # Ensure instance can have its admin password set. - instance = jsonutils.to_primitive(self._create_fake_instance()) - self.compute.run_instance(self.context, instance, {}, {}, [], None, - None, True, None, False) - db.instance_update(self.context, instance['uuid'], - {'task_state': task_states.UPDATING_PASSWORD}) - - inst_ref = db.instance_get_by_uuid(self.context, instance['uuid']) - self.assertEqual(inst_ref['vm_state'], vm_states.ACTIVE) - self.assertEqual(inst_ref['task_state'], task_states.UPDATING_PASSWORD) - - self.compute.set_admin_password(self.context, instance, None) - - inst_ref = db.instance_get_by_uuid(self.context, instance['uuid']) - self.assertEqual(inst_ref['vm_state'], vm_states.ACTIVE) - self.assertIsNone(inst_ref['task_state']) - - self.compute.terminate_instance(self.context, - self._objectify(inst_ref), [], []) - - def test_set_admin_password_bad_state(self): - # Test setting password while instance is rebuilding. - instance = jsonutils.to_primitive(self._create_fake_instance()) - self.compute.run_instance(self.context, instance, {}, {}, [], None, - None, True, None, False) - db.instance_update(self.context, instance['uuid'], { - "power_state": power_state.NOSTATE, - }) - instance = jsonutils.to_primitive(db.instance_get_by_uuid( - self.context, instance['uuid'])) - - self.assertEqual(instance['power_state'], power_state.NOSTATE) - - def fake_driver_get_info(self2, _instance): - return {'state': power_state.NOSTATE, - 'max_mem': 0, - 'mem': 0, - 'num_cpu': 2, - 'cpu_time': 0} - - self.stubs.Set(nova.virt.fake.FakeDriver, 'get_info', - fake_driver_get_info) - - db.instance_update(self.context, instance['uuid'], - {"task_state": task_states.UPDATING_PASSWORD}) - self.assertRaises(exception.InstancePasswordSetFailed, - self.compute.set_admin_password, - self.context, - instance, None) - self.compute.terminate_instance(self.context, - self._objectify(instance), [], []) - - def _do_test_set_admin_password_driver_error(self, exc, expected_vm_state, - expected_task_state, - expected_exception): - """Ensure expected exception is raised if set_admin_password fails.""" - - def fake_sleep(_time): - pass - - self.stubs.Set(time, 'sleep', fake_sleep) - - def fake_driver_set_pass(self2, _instance, _pwd): - raise exc - - self.stubs.Set(nova.virt.fake.FakeDriver, 'set_admin_password', - fake_driver_set_pass) - - instance = jsonutils.to_primitive(self._create_fake_instance()) - self.compute.run_instance(self.context, instance, {}, {}, [], None, - None, True, None, False) - db.instance_update(self.context, instance['uuid'], - {'task_state': task_states.UPDATING_PASSWORD}) - - inst_ref = db.instance_get_by_uuid(self.context, instance['uuid']) - self.assertEqual(inst_ref['vm_state'], vm_states.ACTIVE) - self.assertEqual(inst_ref['task_state'], task_states.UPDATING_PASSWORD) - - #error raised from the driver should not reveal internal information - #so a new error is raised - self.assertRaises(expected_exception, - self.compute.set_admin_password, - self.context, - instance=jsonutils.to_primitive(inst_ref), - new_pass=None) - - inst_ref = db.instance_get_by_uuid(self.context, instance['uuid']) - self.assertEqual(inst_ref['vm_state'], expected_vm_state) - self.assertEqual(inst_ref['task_state'], expected_task_state) - - self.compute.terminate_instance(self.context, - self._objectify(inst_ref), [], []) - - def test_set_admin_password_driver_not_authorized(self): - """Ensure expected exception is raised if set_admin_password not - authorized. - """ - exc = exception.Forbidden(_('Internal error')) - expected_exception = exception.InstancePasswordSetFailed - self._do_test_set_admin_password_driver_error(exc, - vm_states.ERROR, - None, - expected_exception) - - def test_set_admin_password_driver_not_implemented(self): - """Ensure expected exception is raised if set_admin_password not - implemented by driver. - """ - exc = NotImplementedError() - expected_exception = NotImplementedError - self._do_test_set_admin_password_driver_error(exc, - vm_states.ACTIVE, - None, - expected_exception) - def test_inject_network_info(self): # Ensure we can inject network info. called = {'inject': False} @@ -7397,34 +7281,6 @@ class ComputeAPITestCase(BaseTestCase): self.assertEqual('test-host', instances[0]['hostname']) - def test_set_admin_password(self): - # Ensure instance can have its admin password set. - instance = jsonutils.to_primitive(self._create_fake_instance()) - instance_uuid = instance['uuid'] - self.compute.run_instance(self.context, instance, {}, {}, None, None, - None, True, None, False) - - inst_ref = db.instance_get_by_uuid(self.context, instance_uuid) - self.assertEqual(inst_ref['vm_state'], vm_states.ACTIVE) - self.assertIsNone(inst_ref['task_state']) - - def fake_set_admin_password(self, context, **kwargs): - pass - - self.stubs.Set(compute_rpcapi.ComputeAPI, - 'set_admin_password', - fake_set_admin_password) - - self.compute_api.set_admin_password(self.context, inst_ref) - - inst_ref = db.instance_get_by_uuid(self.context, instance_uuid) - self.assertEqual(inst_ref['vm_state'], vm_states.ACTIVE) - self.assertEqual(inst_ref['task_state'], - task_states.UPDATING_PASSWORD) - - self.compute.terminate_instance(self.context, - self._objectify(inst_ref), [], []) - def _fake_rescue_block_devices(self, instance, status="in-use"): fake_bdms = block_device_obj.block_device_make_list(self.context, [fake_block_device.FakeDbBlockDeviceDict( diff --git a/nova/tests/compute/test_compute_api.py b/nova/tests/compute/test_compute_api.py index c620ccb51c..67ed2cd9ec 100644 --- a/nova/tests/compute/test_compute_api.py +++ b/nova/tests/compute/test_compute_api.py @@ -2062,6 +2062,36 @@ class _ComputeAPIUnitTestMixIn(object): rpcapi_unrescue_instance.assert_called_once_with( self.context, instance=instance) + def test_set_admin_password_invalid_state(self): + # Tests that InstanceInvalidState is raised when not ACTIVE. + instance = self._create_instance_obj({'vm_state': vm_states.STOPPED}) + self.assertRaises(exception.InstanceInvalidState, + self.compute_api.set_admin_password, + self.context, instance) + + def test_set_admin_password(self): + # Ensure instance can have its admin password set. + instance = self._create_instance_obj() + + @mock.patch.object(self.compute_api, 'update') + @mock.patch.object(self.compute_api, '_record_action_start') + @mock.patch.object(self.compute_api.compute_rpcapi, + 'set_admin_password') + def do_test(compute_rpcapi_mock, record_mock, update_mock): + # call the API + self.compute_api.set_admin_password(self.context, instance) + # make our assertions + update_mock.assert_called_once_with( + self.context, instance, + task_state=task_states.UPDATING_PASSWORD, + expected_task_state=[None]) + record_mock.assert_called_once_with( + self.context, instance, instance_actions.CHANGE_PASSWORD) + compute_rpcapi_mock.assert_called_once_with( + self.context, instance=instance, new_pass=None) + + do_test() + class ComputeAPIUnitTestCase(_ComputeAPIUnitTestMixIn, test.NoDBTestCase): def setUp(self): diff --git a/nova/tests/compute/test_compute_mgr.py b/nova/tests/compute/test_compute_mgr.py index 0c01423f3e..8c6fe2ef09 100644 --- a/nova/tests/compute/test_compute_mgr.py +++ b/nova/tests/compute/test_compute_mgr.py @@ -1316,6 +1316,147 @@ class ComputeManagerUnitTestCase(test.NoDBTestCase): instance_save.assert_called_once_with( expected_task_state=task_states.UNRESCUING) + @mock.patch.object(objects.InstanceActionEvent, 'event_start') + @mock.patch.object(objects.InstanceActionEvent, + 'event_finish_with_failure') + @mock.patch('nova.compute.manager.ComputeManager._get_power_state', + return_value=power_state.RUNNING) + @mock.patch.object(objects.Instance, 'save') + @mock.patch('nova.utils.generate_password', return_value='fake-pass') + def test_set_admin_password(self, gen_password_mock, + instance_save_mock, power_state_mock, + event_finish_mock, event_start_mock): + # Ensure instance can have its admin password set. + instance = fake_instance.fake_instance_obj( + self.context, + vm_state=vm_states.ACTIVE, + task_state=task_states.UPDATING_PASSWORD) + + @mock.patch.object(self.context, 'elevated', return_value=self.context) + @mock.patch.object(self.compute.driver, 'set_admin_password') + def do_test(driver_mock, elevated_mock): + # call the manager method + self.compute.set_admin_password(self.context, instance, None) + # make our assertions + self.assertEqual(vm_states.ACTIVE, instance.vm_state) + self.assertIsNone(instance.task_state) + + power_state_mock.assert_called_once_with(self.context, instance) + driver_mock.assert_called_once_with(instance, 'fake-pass') + instance_save_mock.assert_called_once_with( + expected_task_state=task_states.UPDATING_PASSWORD) + + do_test() + + @mock.patch.object(objects.InstanceActionEvent, 'event_start') + @mock.patch.object(objects.InstanceActionEvent, + 'event_finish_with_failure') + @mock.patch('nova.compute.manager.ComputeManager._get_power_state', + return_value=power_state.NOSTATE) + @mock.patch.object(objects.Instance, 'save') + @mock.patch.object(compute_utils, 'add_instance_fault_from_exc') + def test_set_admin_password_bad_state(self, add_fault_mock, + instance_save_mock, power_state_mock, + event_finish_mock, event_start_mock): + # Test setting password while instance is rebuilding. + instance = fake_instance.fake_instance_obj(self.context) + with mock.patch.object(self.context, 'elevated', + return_value=self.context): + # call the manager method + self.assertRaises(exception.InstancePasswordSetFailed, + self.compute.set_admin_password, + self.context, instance, None) + + # make our assertions + power_state_mock.assert_called_once_with(self.context, instance) + instance_save_mock.assert_called_once_with( + expected_task_state=task_states.UPDATING_PASSWORD) + add_fault_mock.assert_called_once_with( + self.context, instance, mock.ANY, mock.ANY) + + @mock.patch.object(objects.InstanceActionEvent, 'event_start') + @mock.patch.object(objects.InstanceActionEvent, + 'event_finish_with_failure') + @mock.patch('nova.utils.generate_password', return_value='fake-pass') + @mock.patch('nova.compute.manager.ComputeManager._get_power_state', + return_value=power_state.RUNNING) + @mock.patch('nova.compute.manager.ComputeManager._instance_update') + @mock.patch.object(objects.Instance, 'save') + @mock.patch.object(compute_utils, 'add_instance_fault_from_exc') + def _do_test_set_admin_password_driver_error(self, exc, + expected_vm_state, + expected_task_state, + expected_exception, + add_fault_mock, + instance_save_mock, + update_mock, + power_state_mock, + gen_password_mock, + event_finish_mock, + event_start_mock): + # Ensure expected exception is raised if set_admin_password fails. + instance = fake_instance.fake_instance_obj( + self.context, + vm_state=vm_states.ACTIVE, + task_state=task_states.UPDATING_PASSWORD) + + @mock.patch.object(self.context, 'elevated', return_value=self.context) + @mock.patch.object(self.compute.driver, 'set_admin_password', + side_effect=exc) + def do_test(driver_mock, elevated_mock): + # error raised from the driver should not reveal internal + # information so a new error is raised + self.assertRaises(expected_exception, + self.compute.set_admin_password, + self.context, + instance=instance, + new_pass=None) + + # NOTE(mriedem): Currently instance.save is only called for + # NotImplementedError so if that's not the expected_exception then + # we have to check update_mock. + if expected_exception == NotImplementedError: + instance_save_mock.assert_called_once_with( + expected_task_state=task_states.UPDATING_PASSWORD) + self.assertEqual(expected_vm_state, instance.vm_state) + self.assertEqual(expected_task_state, instance.task_state) + # check revert_task_state decorator + update_mock.assert_called_once_with( + self.context, instance.uuid, + task_state=expected_task_state) + else: + self.assertFalse(instance_save_mock.called) + calls = [ + # _set_instance_error_state + mock.call(self.context, instance.uuid, + vm_state=expected_vm_state), + # revert_task_state + mock.call(self.context, instance.uuid, + task_state=expected_task_state) + ] + update_mock.assert_has_calls(calls) + + add_fault_mock.assert_called_once_with( + self.context, instance, mock.ANY, mock.ANY) + + do_test() + + def test_set_admin_password_driver_not_authorized(self): + # Ensure expected exception is raised if set_admin_password not + # authorized. + exc = exception.Forbidden('Internal error') + expected_exception = exception.InstancePasswordSetFailed + self._do_test_set_admin_password_driver_error( + exc, vm_states.ERROR, None, expected_exception) + + def test_set_admin_password_driver_not_implemented(self): + # Ensure expected exception is raised if set_admin_password not + # implemented by driver. + exc = NotImplementedError() + expected_exception = NotImplementedError + self._do_test_set_admin_password_driver_error( + exc, vm_states.ACTIVE, None, expected_exception) + class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase): def setUp(self):