openstack
/
nova
Code Issues Proposed changes

Remove mox in tests/unit/api/openstack/compute 

Replace mox with mock or stub_out in the following files
in nova/tests/unit/api/openstack/compute directory.

* admin_only_action_common.py
* test_admin_actions.py
* test_lock_server.py
* test_migrate_server.py
* test_pause_server.py
* test_suspend_server.py

Change-Id: I83b473e9ba557545b5c186f979e068e442de2424
Implements: blueprint mox-removal
This commit is contained in:
Takashi NATSUME 2018-05-07 17:46:06 +09:00
parent e2b0b469be
commit 399554a518
6 changed files with 180 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

201
nova/tests/unit/api/openstack/compute/admin_only_action_common.py
View File

@ -12,6 +12,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import fixtures
import mock
from oslo_utils import timeutils
from oslo_utils import uuidutils
import webob
@ -30,65 +32,64 @@ class CommonMixin(object):
self.req = fakes.HTTPRequest.blank('')
self.context = self.req.environ['nova.context']
self.mock_get = self.useFixture(
fixtures.MockPatch('nova.compute.api.API.get')).mock
def _stub_instance_get(self, uuid=None):
if uuid is None:
uuid = uuidutils.generate_uuid()
instance = fake_instance.fake_instance_obj(self.context,
id=1, uuid=uuid, vm_state=vm_states.ACTIVE,
task_state=None, launched_at=timeutils.utcnow())
self.compute_api.get(
self.context, uuid, expected_attrs=None).AndReturn(instance)
self.mock_get.return_value = instance
return instance
def _stub_instance_get_failure(self, exc_info, uuid=None):
if uuid is None:
uuid = uuidutils.generate_uuid()
self.compute_api.get(
self.context, uuid, expected_attrs=None).AndRaise(exc_info)
return uuid
def _test_non_existing_instance(self, action, body_map=None):
uuid = uuidutils.generate_uuid()
self._stub_instance_get_failure(
exception.InstanceNotFound(instance_id=uuid), uuid=uuid)
# Reset the mock.
self.mock_get.reset_mock()
uuid = uuidutils.generate_uuid()
self.mock_get.side_effect = exception.InstanceNotFound(
instance_id=uuid)
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPNotFound,
controller_function,
self.req, uuid, body=body_map)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
self.mock_get.assert_called_once_with(self.context, uuid,
expected_attrs=None)
def _test_action(self, action, body=None, method=None,
compute_api_args_map=None):
# Reset the mock.
self.mock_get.reset_mock()
if method is None:
method = action.replace('_', '')
compute_api_args_map = compute_api_args_map or {}
instance = self._stub_instance_get()
args, kwargs = compute_api_args_map.get(action, ((), {}))
getattr(self.compute_api, method)(self.context, instance, *args,
**kwargs)
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
res = controller_function(self.req, instance.uuid, body=body)
# NOTE: on v2.1, http status code is set as wsgi_code of API
# method instead of status_int in a response object.
if self._api_version == '2.1':
status_int = controller_function.wsgi_code
else:
status_int = res.status_int
self.assertEqual(202, status_int)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
with mock.patch.object(self.compute_api, method) as mock_method:
controller_function = getattr(self.controller, action)
res = controller_function(self.req, instance.uuid, body=body)
# NOTE: on v2.1, http status code is set as wsgi_code of API
# method instead of status_int in a response object.
if self._api_version == '2.1':
status_int = controller_function.wsgi_code
else:
status_int = res.status_int
self.assertEqual(202, status_int)
mock_method.assert_called_once_with(self.context, instance, *args,
**kwargs)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def _test_not_implemented_state(self, action, method=None):
# Reset the mock.
self.mock_get.reset_mock()
if method is None:
method = action.replace('_', '')
@ -96,23 +97,25 @@ class CommonMixin(object):
body = {}
compute_api_args_map = {}
args, kwargs = compute_api_args_map.get(action, ((), {}))
getattr(self.compute_api, method)(self.context, instance,
*args, **kwargs).AndRaise(
NotImplementedError())
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPNotImplemented,
controller_function,
self.req, instance.uuid, body=body)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
with mock.patch.object(
self.compute_api, method,
side_effect=NotImplementedError()) as mock_method:
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPNotImplemented,
controller_function,
self.req, instance.uuid, body=body)
mock_method.assert_called_once_with(self.context, instance,
*args, **kwargs)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def _test_invalid_state(self, action, method=None, body_map=None,
compute_api_args_map=None,
exception_arg=None):
# Reset the mock.
self.mock_get.reset_mock()
if method is None:
method = action.replace('_', '')
if body_map is None:
@ -124,28 +127,29 @@ class CommonMixin(object):
args, kwargs = compute_api_args_map.get(action, ((), {}))
getattr(self.compute_api, method)(self.context, instance,
*args, **kwargs).AndRaise(
exception.InstanceInvalidState(
with mock.patch.object(
self.compute_api, method,
side_effect=exception.InstanceInvalidState(
attr='vm_state', instance_uuid=instance.uuid,
state='foo', method=method))
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
ex = self.assertRaises(webob.exc.HTTPConflict,
controller_function,
self.req, instance.uuid,
body=body_map)
self.assertIn("Cannot \'%(action)s\' instance %(id)s"
% {'action': exception_arg or method,
'id': instance.uuid}, ex.explanation)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
state='foo', method=method)) as mock_method:
controller_function = getattr(self.controller, action)
ex = self.assertRaises(webob.exc.HTTPConflict,
controller_function,
self.req, instance.uuid,
body=body_map)
self.assertIn("Cannot \'%(action)s\' instance %(id)s"
% {'action': exception_arg or method,
'id': instance.uuid}, ex.explanation)
mock_method.assert_called_once_with(self.context, instance,
*args, **kwargs)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def _test_locked_instance(self, action, method=None, body=None,
compute_api_args_map=None):
# Reset the mock.
self.mock_get.reset_mock()
if method is None:
method = action.replace('_', '')
@ -153,23 +157,25 @@ class CommonMixin(object):
instance = self._stub_instance_get()
args, kwargs = compute_api_args_map.get(action, ((), {}))
getattr(self.compute_api, method)(self.context, instance, *args,
**kwargs).AndRaise(
exception.InstanceIsLocked(instance_uuid=instance.uuid))
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPConflict,
controller_function,
self.req, instance.uuid, body=body)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
with mock.patch.object(
self.compute_api, method,
side_effect=exception.InstanceIsLocked(
instance_uuid=instance.uuid)) as mock_method:
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPConflict,
controller_function,
self.req, instance.uuid, body=body)
mock_method.assert_called_once_with(self.context, instance,
*args, **kwargs)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def _test_instance_not_found_in_compute_api(self, action,
method=None, body=None, compute_api_args_map=None):
# Reset the mock.
self.mock_get.reset_mock()
if method is None:
method = action.replace('_', '')
compute_api_args_map = compute_api_args_map or {}
@ -177,20 +183,19 @@ class CommonMixin(object):
instance = self._stub_instance_get()
args, kwargs = compute_api_args_map.get(action, ((), {}))
getattr(self.compute_api, method)(self.context, instance, *args,
**kwargs).AndRaise(
exception.InstanceNotFound(instance_id=instance.uuid))
self.mox.ReplayAll()
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPNotFound,
controller_function,
self.req, instance.uuid, body=body)
# Do these here instead of tearDown because this method is called
# more than once for the same test case
self.mox.VerifyAll()
self.mox.UnsetStubs()
with mock.patch.object(
self.compute_api, method,
side_effect=exception.InstanceNotFound(
instance_id=instance.uuid)) as mock_method:
controller_function = getattr(self.controller, action)
self.assertRaises(webob.exc.HTTPNotFound,
controller_function,
self.req, instance.uuid, body=body)
mock_method.assert_called_once_with(self.context, instance,
*args, **kwargs)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
class CommonTests(CommonMixin, test.NoDBTestCase):
@ -202,12 +207,8 @@ class CommonTests(CommonMixin, test.NoDBTestCase):
for action in actions:
method = method_translations.get(action)
body = body_map.get(action)
self.mox.StubOutWithMock(self.compute_api,
method or action.replace('_', ''))
self._test_action(action, method=method, body=body,
compute_api_args_map=args_map)
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')
def _test_actions_instance_not_found_in_compute_api(self,
actions, method_translations=None, body_map=None,
@ -218,21 +219,15 @@ class CommonTests(CommonMixin, test.NoDBTestCase):
for action in actions:
method = method_translations.get(action)
body = body_map.get(action)
self.mox.StubOutWithMock(self.compute_api,
method or action.replace('_', ''))
self._test_instance_not_found_in_compute_api(
action, method=method, body=body,
compute_api_args_map=args_map)
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')
def _test_actions_with_non_existed_instance(self, actions, body_map=None):
body_map = body_map or {}
for action in actions:
self._test_non_existing_instance(action,
body_map=body_map.get(action))
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')
def _test_actions_raise_conflict_on_invalid_state(
self, actions, method_translations=None, body_map=None,
@ -244,14 +239,10 @@ class CommonTests(CommonMixin, test.NoDBTestCase):
for action in actions:
method = method_translations.get(action)
exception_arg = exception_args.get(action)
self.mox.StubOutWithMock(self.compute_api,
method or action.replace('_', ''))
self._test_invalid_state(action, method=method,
body_map=body_map.get(action),
compute_api_args_map=args_map,
exception_arg=exception_arg)
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')
def _test_actions_with_locked_instance(self, actions,
method_translations=None,
@ -262,9 +253,5 @@ class CommonTests(CommonMixin, test.NoDBTestCase):
for action in actions:
method = method_translations.get(action)
body = body_map.get(action)
self.mox.StubOutWithMock(self.compute_api,
method or action.replace('_', ''))
self._test_locked_instance(action, method=method, body=body,
compute_api_args_map=args_map)
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')

11
nova/tests/unit/api/openstack/compute/test_admin_actions.py
View File

@ -27,14 +27,9 @@ class AdminActionsTestV21(admin_only_action_common.CommonTests):
super(AdminActionsTestV21, self).setUp()
self.controller = self.admin_actions.AdminActionsController()
self.compute_api = self.controller.compute_api
def _fake_controller(*args, **kwargs):
return self.controller
self.stubs.Set(self.admin_actions, 'AdminActionsController',
_fake_controller)
self.mox.StubOutWithMock(self.compute_api, 'get')
self.stub_out('nova.api.openstack.compute.admin_actions.'
'AdminActionsController',
lambda *a, **k: self.controller)
def test_actions(self):
actions = ['_reset_network', '_inject_network_info']

29
nova/tests/unit/api/openstack/compute/test_lock_server.py
View File

@ -35,13 +35,9 @@ class LockServerTestsV21(admin_only_action_common.CommonTests):
super(LockServerTestsV21, self).setUp()
self.controller = getattr(self.lock_server, self.controller_name)()
self.compute_api = self.controller.compute_api
def _fake_controller(*args, **kwargs):
return self.controller
self.stubs.Set(self.lock_server, self.controller_name,
_fake_controller)
self.mox.StubOutWithMock(self.compute_api, 'get')
self.stub_out('nova.api.openstack.compute.lock_server.'
'LockServerController',
lambda *a, **kw: self.controller)
def test_lock_unlock(self):
self._test_actions(['_lock', '_unlock'])
@ -50,18 +46,19 @@ class LockServerTestsV21(admin_only_action_common.CommonTests):
self._test_actions_with_non_existed_instance(['_lock', '_unlock'])
def test_unlock_not_authorized(self):
self.mox.StubOutWithMock(self.compute_api, 'unlock')
instance = self._stub_instance_get()
self.compute_api.unlock(self.context, instance).AndRaise(
exception.PolicyNotAuthorized(action='unlock'))
self.mox.ReplayAll()
body = {}
self.assertRaises(self.authorization_error,
self.controller._unlock,
self.req, instance.uuid, body)
with mock.patch.object(
self.compute_api, 'unlock',
side_effect=exception.PolicyNotAuthorized(
action='unlock')) as mock_unlock:
self.assertRaises(self.authorization_error,
self.controller._unlock,
self.req, instance.uuid, body)
mock_unlock.assert_called_once_with(self.context, instance)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
@mock.patch.object(common, 'get_instance')
def test_unlock_override_not_authorized_with_non_admin_user(

123
nova/tests/unit/api/openstack/compute/test_migrate_server.py
View File

@ -43,13 +43,9 @@ class MigrateServerTestsV21(admin_only_action_common.CommonTests):
super(MigrateServerTestsV21, self).setUp()
self.controller = getattr(self.migrate_server, self.controller_name)()
self.compute_api = self.controller.compute_api
def _fake_controller(*args, **kwargs):
return self.controller
self.stubs.Set(self.migrate_server, self.controller_name,
_fake_controller)
self.mox.StubOutWithMock(self.compute_api, 'get')
self.stub_out('nova.api.openstack.compute.migrate_server.'
'MigrateServerController',
lambda *a, **kw: self.controller)
def _get_migration_body(self, **kwargs):
return {'os-migrateLive': self._get_params(**kwargs)}
@ -120,16 +116,18 @@ class MigrateServerTestsV21(admin_only_action_common.CommonTests):
args_map=args_map, method_translations=method_translations)
def _test_migrate_exception(self, exc_info, expected_result):
self.mox.StubOutWithMock(self.compute_api, 'resize')
instance = self._stub_instance_get()
self.compute_api.resize(
self.context, instance,
host_name=self.host_name).AndRaise(exc_info)
self.mox.ReplayAll()
self.assertRaises(expected_result,
self.controller._migrate,
self.req, instance['uuid'], body={'migrate': None})
with mock.patch.object(self.compute_api, 'resize',
side_effect=exc_info) as mock_resize:
self.assertRaises(expected_result,
self.controller._migrate,
self.req, instance['uuid'],
body={'migrate': None})
mock_resize.assert_called_once_with(
self.context, instance, host_name=self.host_name)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def test_migrate_too_many_instances(self):
exc_info = exception.TooManyInstances(overs='', req='', used=0,
@ -137,17 +135,21 @@ class MigrateServerTestsV21(admin_only_action_common.CommonTests):
self._test_migrate_exception(exc_info, webob.exc.HTTPForbidden)
def _test_migrate_live_succeeded(self, param):
self.mox.StubOutWithMock(self.compute_api, 'live_migrate')
instance = self._stub_instance_get()
self.compute_api.live_migrate(self.context, instance, False,
self.disk_over_commit, 'hostname',
self.force, self.async)
self.mox.ReplayAll()
live_migrate_method = self.controller._migrate_live
live_migrate_method(self.req, instance.uuid,
body={'os-migrateLive': param})
self.assertEqual(202, live_migrate_method.wsgi_code)
with mock.patch.object(self.compute_api,
'live_migrate') as mock_live_migrate:
live_migrate_method(self.req, instance.uuid,
body={'os-migrateLive': param})
self.assertEqual(202, live_migrate_method.wsgi_code)
mock_live_migrate.assert_called_once_with(
self.context, instance, False, self.disk_over_commit,
'hostname', self.force, self.async)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def test_migrate_live_enabled(self):
param = self._get_params(host='hostname')
@ -209,21 +211,22 @@ class MigrateServerTestsV21(admin_only_action_common.CommonTests):
uuid=None,
expected_exc=webob.exc.HTTPBadRequest,
check_response=True):
self.mox.StubOutWithMock(self.compute_api, 'live_migrate')
instance = self._stub_instance_get(uuid=uuid)
self.compute_api.live_migrate(self.context, instance, False,
self.disk_over_commit,
'hostname', self.force, self.async
).AndRaise(fake_exc)
self.mox.ReplayAll()
body = self._get_migration_body(host='hostname')
ex = self.assertRaises(expected_exc,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
if check_response:
self.assertIn(six.text_type(fake_exc), ex.explanation)
with mock.patch.object(
self.compute_api, 'live_migrate',
side_effect=fake_exc) as mock_live_migrate:
ex = self.assertRaises(expected_exc,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
if check_response:
self.assertIn(six.text_type(fake_exc), ex.explanation)
mock_live_migrate.assert_called_once_with(
self.context, instance, False, self.disk_over_commit,
'hostname', self.force, self.async)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def test_migrate_live_compute_service_unavailable(self):
self._test_migrate_live_failed_with_exception(
@ -428,39 +431,41 @@ class MigrateServerTestsV234(MigrateServerTestsV230):
pass
def test_migrate_live_compute_host_not_found(self):
body = {'os-migrateLive':
{'host': 'hostname', 'block_migration': 'auto'}}
exc = exception.ComputeHostNotFound(
reason="Compute host %(host)s could not be found.",
host='hostname')
self.mox.StubOutWithMock(self.compute_api, 'live_migrate')
instance = self._stub_instance_get()
self.compute_api.live_migrate(self.context, instance, None,
self.disk_over_commit, 'hostname',
self.force, self.async).AndRaise(exc)
self.mox.ReplayAll()
body = {'os-migrateLive':
{'host': 'hostname', 'block_migration': 'auto'}}
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
with mock.patch.object(self.compute_api, 'live_migrate',
side_effect=exc) as mock_live_migrate:
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
mock_live_migrate.assert_called_once_with(
self.context, instance, None, self.disk_over_commit,
'hostname', self.force, self.async)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
def test_migrate_live_unexpected_error(self):
exc = exception.InvalidHypervisorType(
reason="The supplied hypervisor type of is invalid.")
self.mox.StubOutWithMock(self.compute_api, 'live_migrate')
instance = self._stub_instance_get()
self.compute_api.live_migrate(self.context, instance, None,
self.disk_over_commit, 'hostname',
self.force, self.async).AndRaise(exc)
self.mox.ReplayAll()
body = {'os-migrateLive':
{'host': 'hostname', 'block_migration': 'auto'}}
exc = exception.InvalidHypervisorType(
reason="The supplied hypervisor type of is invalid.")
instance = self._stub_instance_get()
self.assertRaises(webob.exc.HTTPInternalServerError,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
with mock.patch.object(self.compute_api, 'live_migrate',
side_effect=exc) as mock_live_migrate:
self.assertRaises(webob.exc.HTTPInternalServerError,
self.controller._migrate_live,
self.req, instance.uuid, body=body)
mock_live_migrate.assert_called_once_with(
self.context, instance, None, self.disk_over_commit,
'hostname', self.force, self.async)
self.mock_get.assert_called_once_with(self.context, instance.uuid,
expected_attrs=None)
class MigrateServerTestsV256(MigrateServerTestsV234):

14
nova/tests/unit/api/openstack/compute/test_pause_server.py
View File

@ -33,24 +33,16 @@ class PauseServerTestsV21(admin_only_action_common.CommonTests):
super(PauseServerTestsV21, self).setUp()
self.controller = getattr(self.pause_server, self.controller_name)()
self.compute_api = self.controller.compute_api
def _fake_controller(*args, **kwargs):
return self.controller
self.stubs.Set(self.pause_server, self.controller_name,
_fake_controller)
self.mox.StubOutWithMock(self.compute_api, 'get')
self.stub_out('nova.api.openstack.compute.pause_server.'
'PauseServerController',
lambda *a, **kw: self.controller)
def test_pause_unpause(self):
self._test_actions(['_pause', '_unpause'])
def test_actions_raise_on_not_implemented(self):
for action in ['_pause', '_unpause']:
self.mox.StubOutWithMock(self.compute_api,
action.replace('_', ''))
self._test_not_implemented_state(action)
# Re-mock this.
self.mox.StubOutWithMock(self.compute_api, 'get')
def test_pause_unpause_with_non_existed_instance(self):
self._test_actions_with_non_existed_instance(['_pause', '_unpause'])

10
nova/tests/unit/api/openstack/compute/test_suspend_server.py
View File

@ -32,13 +32,9 @@ class SuspendServerTestsV21(admin_only_action_common.CommonTests):
super(SuspendServerTestsV21, self).setUp()
self.controller = getattr(self.suspend_server, self.controller_name)()
self.compute_api = self.controller.compute_api
def _fake_controller(*args, **kwargs):
return self.controller
self.stubs.Set(self.suspend_server, self.controller_name,
_fake_controller)
self.mox.StubOutWithMock(self.compute_api, 'get')
self.stub_out('nova.api.openstack.compute.suspend_server.'
'SuspendServerController',
lambda *a, **kw: self.controller)
def test_suspend_resume(self):
self._test_actions(['_suspend', '_resume'])