Merge "Move set_admin_password tests from test_compute.py to api/mgr modules"
This commit is contained in:
Jenkins
@@ -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(
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user