Improve mocking in REST API tests (test_cinder)

The current REST API tests have a lot of rooms to be improved like:
- mocking the whole module of the API wrapper
- Use test_data.cinder_data
- Use response.json attribute to decode responses as much as possible

APIResourceWrapper.to_dict() method of openstack_dashboard.api.base
is changed to refer attributes of an object itself instead of
self._apiresource. There is a case where APIResourceWrapper class
defines its own attributes and such attributes should be used in
to_dict() method. If self._apiresouce is referred these attributes
are not considered. We need to consume attributes of a class object
itself. It helps us write unit tests using sample test data.

Change-Id: I5b0c3bb52d97f377c6e5fe2b9a4ace2931114ab0
Partial-Bug: #1752917
This commit is contained in:
Akihiro Motoki 2018-03-13 04:58:56 +09:00
parent 8aebb03e9f
commit 64d5ba8515
2 changed files with 249 additions and 244 deletions

View File

@ -133,7 +133,7 @@ class APIResourceWrapper(object):
def to_dict(self): def to_dict(self):
obj = {} obj = {}
for key in self._attrs: for key in self._attrs:
obj[key] = getattr(self._apiresource, key, None) obj[key] = getattr(self, key, None)
return obj return obj

View File

@ -11,7 +11,6 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json
from django.conf import settings from django.conf import settings
import mock import mock
@ -21,6 +20,7 @@ from openstack_dashboard.api.base import Quota
from openstack_dashboard.api.cinder import VolTypeExtraSpec from openstack_dashboard.api.cinder import VolTypeExtraSpec
from openstack_dashboard.api.rest import cinder from openstack_dashboard.api.rest import cinder
from openstack_dashboard.test import helpers as test from openstack_dashboard.test import helpers as test
from openstack_dashboard.usage import quotas
class CinderRestTestCase(test.TestCase): class CinderRestTestCase(test.TestCase):
@ -38,43 +38,45 @@ class CinderRestTestCase(test.TestCase):
filters = {'status': 'available'} filters = {'status': 'available'}
self._test_volumes_get(False, filters) self._test_volumes_get(False, filters)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_list_paged']})
def _test_volumes_get(self, all, filters, cc): def _test_volumes_get(self, all_, filters):
if all: if all_:
request = self.mock_rest_request(GET={'all_projects': 'true'}) request = self.mock_rest_request(GET={'all_projects': 'true'})
else: else:
request = self.mock_rest_request(**{'GET': filters}) request = self.mock_rest_request(**{'GET': filters})
cc.volume_list_paged.return_value = [ volumes = [self.cinder_volumes.first()]
mock.Mock(**{'to_dict.return_value': {'id': 'test123'}}), self.maxDiff = None
], False, False self.mock_volume_list_paged.return_value = volumes, False, False
cc.Volume.return_value = mock.Mock(
**{'to_dict.return_value': {"id": "test123"}})
response = cinder.Volumes().get(request) response = cinder.Volumes().get(request)
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.json, self.assertEqual({'items': [v.to_dict() for v in volumes],
{"items": [{"id": "test123"}], 'has_more_data': False,
"has_more_data": False, 'has_prev_data': False},
"has_prev_data": False}) response.json)
if all: if all_:
cc.volume_list_paged.assert_called_once_with(request, self.mock_volume_list_paged.assert_called_once_with(
{'all_tenants': 1}) request, {'all_tenants': 1})
else: else:
cc.volume_list_paged.assert_called_once_with(request, self.mock_volume_list_paged.assert_called_once_with(
search_opts=filters) request, search_opts=filters)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_get']})
def test_volume_get(self, cc): def test_volume_get(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_get.return_value = mock.Mock( vol = self.cinder_volumes.first()
**{'to_dict.return_value': {'id': 'one'}}) self.mock_volume_get.return_value = vol
response = cinder.Volume().get(request, '1')
self.assertStatusCode(response, 200)
self.assertEqual(response.json, {"id": "one"})
cc.volume_get.assert_called_once_with(request, '1')
@mock.patch.object(cinder.api, 'cinder') response = cinder.Volume().get(request, '1')
def test_volume_create(self, cc):
self.assertStatusCode(response, 200)
self.assertEqual(vol.to_dict(), response.json)
self.mock_volume_get.assert_called_once_with(request, '1')
@test.create_mocks({api.cinder: ['volume_create']})
def test_volume_create(self):
mock_body = '''{ mock_body = '''{
"size": "", "size": "",
"name": "", "name": "",
@ -87,34 +89,31 @@ class CinderRestTestCase(test.TestCase):
"source_volid": "" "source_volid": ""
}''' }'''
mock_volume_create_response = {
"size": ""
}
mock_post_response = '{"size": ""}'
request = self.mock_rest_request(POST={}, body=mock_body) request = self.mock_rest_request(POST={}, body=mock_body)
cc.volume_create.return_value = \ vol = self.cinder_volumes.first()
mock.Mock(**{'to_dict.return_value': mock_volume_create_response}) self.mock_volume_create.return_value = vol
response = cinder.Volumes().post(request) response = cinder.Volumes().post(request)
self.assertStatusCode(response, 201) self.assertStatusCode(response, 201)
self.assertEqual(response.content.decode("utf-8"), mock_post_response) self.assertEqual(vol.to_dict(), response.json)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_get']})
def test_volume_get_metadata(self, cc): def test_volume_get_metadata(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_get.return_value = mock.Mock( vol = self.cinder_volumes.first()
**{'to_dict.return_value': {'id': 'one', vol._apiresource.metadata = {'foo': 'bar'}
'metadata': {'foo': 'bar'}}}) self.mock_volume_get.return_value = vol
response = cinder.VolumeMetadata().get(request, '1')
self.assertStatusCode(response, 200)
self.assertEqual(response.json, {'foo': 'bar'})
cc.volume_get.assert_called_once_with(request, '1')
@mock.patch.object(cinder.api, 'cinder') response = cinder.VolumeMetadata().get(request, '1')
def test_volume_update_metadata(self, cc):
self.assertStatusCode(response, 200)
self.assertEqual(vol.to_dict()['metadata'], response.json)
self.mock_volume_get.assert_called_once_with(request, '1')
@test.create_mocks({api.cinder: ['volume_set_metadata',
'volume_delete_metadata']})
def test_volume_update_metadata(self):
request = self.mock_rest_request( request = self.mock_rest_request(
body='{"updated": {"a": "1", "b": "2"}, ' body='{"updated": {"a": "1", "b": "2"}, '
'"removed": ["c", "d"]}' '"removed": ["c", "d"]}'
@ -122,67 +121,68 @@ class CinderRestTestCase(test.TestCase):
response = cinder.VolumeMetadata().patch(request, '1') response = cinder.VolumeMetadata().patch(request, '1')
self.assertStatusCode(response, 204) self.assertStatusCode(response, 204)
self.assertEqual(b'', response.content) self.assertEqual(b'', response.content)
cc.volume_set_metadata.assert_called_once_with( self.mock_volume_set_metadata.assert_called_once_with(
request, '1', {'a': '1', 'b': '2'} request, '1', {'a': '1', 'b': '2'}
) )
cc.volume_delete_metadata.assert_called_once_with( self.mock_volume_delete_metadata.assert_called_once_with(
request, '1', ['c', 'd'] request, '1', ['c', 'd']
) )
# #
# Volume Types # Volume Types
# #
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_type_list']})
def test_volume_types_get(self, cc): def test_volume_types_get(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.VolumeType.return_value = mock.Mock( types = self.cinder_volume_types.list()
**{'to_dict.return_value': {'id': 'one'}}) self.mock_volume_type_list.return_value = types
cc.volume_type_list.return_value = [{'id': 'one'}]
response = cinder.VolumeTypes().get(request) response = cinder.VolumeTypes().get(request)
self.assertStatusCode(response, 200)
self.assertEqual(response.json, {"items": [{"id": "one"}]})
cc.volume_type_list.assert_called_once_with(request)
cc.VolumeType.assert_called_once_with({'id': 'one'})
@mock.patch.object(cinder.api, 'cinder') self.assertStatusCode(response, 200)
def test_volume_type_get(self, cc): self.assertEqual([api.cinder.VolumeType(t).to_dict() for t in types],
response.json['items'])
self.mock_volume_type_list.assert_called_once_with(request)
@test.create_mocks({api.cinder: ['volume_type_get']})
def test_volume_type_get(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_type_get.return_value = {'name': 'one'} type_ = self.cinder_volume_types.first()
cc.VolumeType.return_value = mock.Mock( self.mock_volume_type_get.return_value = type_
**{'to_dict.return_value': {'id': 'one'}})
response = cinder.VolumeType().get(request, '1') response = cinder.VolumeType().get(request, '1')
self.assertStatusCode(response, 200)
self.assertEqual(response.json, {"id": "one"})
cc.volume_type_get.assert_called_once_with(request, '1')
cc.VolumeType.assert_called_once_with({'name': 'one'})
@mock.patch.object(cinder.api, 'cinder') self.assertStatusCode(response, 200)
def test_volume_type_get_default(self, cc): self.assertEqual(api.cinder.VolumeType(type_).to_dict(),
response.json)
self.mock_volume_type_get.assert_called_once_with(request, '1')
@test.create_mocks({api.cinder: ['volume_type_default']})
def test_volume_type_get_default(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_type_default.return_value = {'name': 'one'} type_ = self.cinder_volume_types.first()
cc.VolumeType.return_value = mock.Mock( self.mock_volume_type_default.return_value = type_
**{'to_dict.return_value': {'id': 'one'}})
response = cinder.VolumeType().get(request, 'default') response = cinder.VolumeType().get(request, 'default')
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.json, {"id": "one"}) self.assertEqual(api.cinder.VolumeType(type_).to_dict(),
cc.volume_type_default.assert_called_once_with(request) response.json)
cc.VolumeType.assert_called_once_with({'name': 'one'}) self.mock_volume_type_default.assert_called_once_with(request)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_type_extra_get']})
def test_volume_type_get_metadata(self, cc): def test_volume_type_get_metadata(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_type_extra_get = mock.Mock() self.mock_volume_type_extra_get.return_value = \
cc.volume_type_extra_get.return_value = \
[VolTypeExtraSpec(1, 'foo', 'bar')] [VolTypeExtraSpec(1, 'foo', 'bar')]
# cc.volume_type_extra_get.side_effect = [{'foo': 'bar'}]
response = cinder.VolumeTypeMetadata().get(request, '1') response = cinder.VolumeTypeMetadata().get(request, '1')
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.json, {'foo': 'bar'}) self.assertEqual(response.json, {'foo': 'bar'})
cc.volume_type_extra_get.assert_called_once_with(request, '1') self.mock_volume_type_extra_get.assert_called_once_with(request, '1')
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_type_extra_set',
def test_volume_type_update_metadata(self, cc): 'volume_type_extra_delete']})
def test_volume_type_update_metadata(self):
request = self.mock_rest_request( request = self.mock_rest_request(
body='{"updated": {"a": "1", "b": "2"}, ' body='{"updated": {"a": "1", "b": "2"}, '
'"removed": ["c", "d"]}' '"removed": ["c", "d"]}'
@ -190,81 +190,91 @@ class CinderRestTestCase(test.TestCase):
response = cinder.VolumeTypeMetadata().patch(request, '1') response = cinder.VolumeTypeMetadata().patch(request, '1')
self.assertStatusCode(response, 204) self.assertStatusCode(response, 204)
self.assertEqual(b'', response.content) self.assertEqual(b'', response.content)
cc.volume_type_extra_set.assert_called_once_with( self.mock_volume_type_extra_set.assert_called_once_with(
request, '1', {'a': '1', 'b': '2'} request, '1', {'a': '1', 'b': '2'}
) )
cc.volume_type_extra_delete.assert_called_once_with( self.mock_volume_type_extra_delete.assert_called_once_with(
request, '1', ['c', 'd'] request, '1', ['c', 'd']
) )
# #
# Volume Snapshots # Volume Snapshots
# #
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_snapshot_list']})
def test_volume_snaps_get(self, cc): def test_volume_snaps_get(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_snapshot_list.return_value = [ snapshots = self.cinder_volume_snapshots.list()
mock.Mock(**{'to_dict.return_value': {'id': 'one'}}), self.mock_volume_snapshot_list.return_value = snapshots
mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),
]
response = cinder.VolumeSnapshots().get(request) response = cinder.VolumeSnapshots().get(request)
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.json, self.assertEqual([s.to_dict() for s in snapshots],
{"items": [{"id": "one"}, {"id": "two"}]}) response.json['items'])
cc.volume_snapshot_list.assert_called_once_with(request, self.mock_volume_snapshot_list.assert_called_once_with(request,
search_opts={}) search_opts={})
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['volume_snapshot_list']})
def test_volume_snaps_get_with_filters(self, cc): def test_volume_snaps_get_with_filters(self):
filters = {'status': 'available'} filters = {'status': 'available'}
request = self.mock_rest_request(**{'GET': dict(filters)}) request = self.mock_rest_request(**{'GET': dict(filters)})
cc.volume_snapshot_list.return_value = [ snapshots = self.cinder_volume_snapshots.list()
mock.Mock(**{'to_dict.return_value': {'id': 'one'}}), self.mock_volume_snapshot_list.return_value = snapshots
mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),
]
response = cinder.VolumeSnapshots().get(request) response = cinder.VolumeSnapshots().get(request)
self.assertStatusCode(response, 200)
self.assertEqual(response.json,
{"items": [{"id": "one"}, {"id": "two"}]})
cc.volume_snapshot_list.assert_called_once_with(request,
search_opts=filters)
@mock.patch.object(cinder.api, 'cinder') self.assertStatusCode(response, 200)
def test_volume_snapshot_get_metadata(self, cc): self.assertEqual([s.to_dict() for s in snapshots],
response.json['items'])
self.mock_volume_snapshot_list.assert_called_once_with(
request, search_opts=filters)
@test.create_mocks({api.cinder: ['volume_snapshot_get']})
def test_volume_snapshot_get_metadata(self):
request = self.mock_rest_request(**{'GET': {}}) request = self.mock_rest_request(**{'GET': {}})
cc.volume_snapshot_get.return_value = mock.Mock( # 4th item contains metadata
**{'to_dict.return_value': {'id': 'one', snapshot = self.cinder_volume_snapshots.list()[3]
'metadata': {'foo': 'bar'}}}) # Ensure metadata is contained in test data
response = cinder.VolumeSnapshotMetadata().get(request, '1') assert 'metadata' in snapshot.to_dict()
self.assertStatusCode(response, 200) self.mock_volume_snapshot_get.return_value = snapshot
self.assertEqual(response.json, {'foo': 'bar'})
cc.volume_snapshot_get.assert_called_once_with(request, '1')
@mock.patch.object(cinder.api, 'cinder') response = cinder.VolumeSnapshotMetadata().get(request, '1')
def test_volume_snapshot_update_metadata(self, cc):
self.assertStatusCode(response, 200)
self.assertEqual(snapshot.metadata, response.json)
self.mock_volume_snapshot_get.assert_called_once_with(request, '1')
@test.create_mocks({api.cinder: ['volume_snapshot_set_metadata',
'volume_snapshot_delete_metadata']})
def test_volume_snapshot_update_metadata(self):
request = self.mock_rest_request( request = self.mock_rest_request(
body='{"updated": {"a": "1", "b": "2"}, ' body='{"updated": {"a": "1", "b": "2"}, '
'"removed": ["c", "d"]}' '"removed": ["c", "d"]}'
) )
self.mock_volume_snapshot_set_metadata.return_value = None
self.mock_volume_snapshot_delete_metadata.return_value = None
response = cinder.VolumeSnapshotMetadata().patch(request, '1') response = cinder.VolumeSnapshotMetadata().patch(request, '1')
self.assertStatusCode(response, 204) self.assertStatusCode(response, 204)
self.assertEqual(b'', response.content) self.assertEqual(b'', response.content)
cc.volume_snapshot_set_metadata.assert_called_once_with( self.mock_volume_snapshot_set_metadata.assert_called_once_with(
request, '1', {'a': '1', 'b': '2'} request, '1', {'a': '1', 'b': '2'}
) )
cc.volume_snapshot_delete_metadata.assert_called_once_with( self.mock_volume_snapshot_delete_metadata.assert_called_once_with(
request, '1', ['c', 'd'] request, '1', ['c', 'd']
) )
# #
# Extensions # Extensions
# #
@mock.patch.object(cinder.api, 'cinder') # TODO(amotoki): This is actually not tested. Make this work.
# TODO(amotoki): There is no such setting named
# OPENSTACK_CINDER_EXTENSIONS_BLACKLIST.
@test.create_mocks({api.cinder: ['list_extensions']})
@mock.patch.object(settings, @mock.patch.object(settings,
'OPENSTACK_CINDER_EXTENSIONS_BLACKLIST', ['baz']) 'OPENSTACK_CINDER_EXTENSIONS_BLACKLIST', ['baz'])
def _test_extension_list(self, cc): def _test_extension_list(self):
request = self.mock_rest_request() request = self.mock_rest_request()
cc.list_extensions.return_value = [ self.mock_list_extensions.return_value = [
mock.Mock(**{'to_dict.return_value': {'name': 'foo'}}), mock.Mock(**{'to_dict.return_value': {'name': 'foo'}}),
mock.Mock(**{'to_dict.return_value': {'name': 'bar'}}), mock.Mock(**{'to_dict.return_value': {'name': 'bar'}}),
mock.Mock(**{'to_dict.return_value': {'name': 'baz'}}), mock.Mock(**{'to_dict.return_value': {'name': 'baz'}}),
@ -273,85 +283,85 @@ class CinderRestTestCase(test.TestCase):
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.content, self.assertEqual(response.content,
'{"items": [{"name": "foo"}, {"name": "bar"}]}') '{"items": [{"name": "foo"}, {"name": "bar"}]}')
cc.list_extensions.assert_called_once_with(request) self.mock_list_extensions.assert_called_once_with(request)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['qos_specs_list']})
def test_qos_specs_get(self, cc): def test_qos_specs_get(self):
request = self.mock_rest_request(GET={}) request = self.mock_rest_request(GET={})
cc.qos_specs_list.return_value = [ qos_specs = self.cinder_qos_specs.list()
mock.Mock(**{'to_dict.return_value': {'id': 'one'}}), self.mock_qos_specs_list.return_value = qos_specs
mock.Mock(**{'to_dict.return_value': {'id': 'two'}}),
]
response = cinder.QoSSpecs().get(request) response = cinder.QoSSpecs().get(request)
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
self.assertEqual(response.content.decode("utf-8"), self.assertEqual([spec.to_dict() for spec in qos_specs],
'{"items": [{"id": "one"}, {"id": "two"}]}') response.json['items'])
cc.qos_specs_list.assert_called_once_with(request) self.mock_qos_specs_list.assert_called_once_with(request)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['tenant_absolute_limits']})
def test_tenant_absolute_limits_get(self, cc): def test_tenant_absolute_limits_get(self):
request = self.mock_rest_request(GET={}) request = self.mock_rest_request(GET={})
cc.tenant_absolute_limits.return_value = \ limits = self.cinder_limits['absolute']
{'id': 'one', 'val': float('inf')} # Ensure to have float(inf) to test it is handled properly
limits['maxTotalVolumes'] = float('inf')
self.mock_tenant_absolute_limits.return_value = limits
response = cinder.TenantAbsoluteLimits().get(request) response = cinder.TenantAbsoluteLimits().get(request)
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
response_as_json = json.loads(response.content.decode('utf-8')) self.assertEqual(limits, response.json)
self.assertEqual(response_as_json['id'], 'one') self.mock_tenant_absolute_limits.assert_called_once_with(request)
self.assertEqual(response_as_json['val'], 1e+999)
cc.tenant_absolute_limits.assert_called_once_with(request)
# #
# Services # Services
# #
@mock.patch.object(api.base, 'is_service_enabled') @test.create_mocks({api.base: ['is_service_enabled'],
@mock.patch.object(cinder.api, 'cinder') api.cinder: ['service_list',
def test_services_get(self, cc, mock_service_enabled): 'extension_supported']})
def test_services_get(self):
request = self.mock_rest_request(GET={}) request = self.mock_rest_request(GET={})
cc.service_list.return_value = [mock.Mock( services = self.cinder_services.list()
binary='binary_1', self.mock_service_list.return_value = services
host='host_1', self.mock_is_service_enabled.return_value = True
zone='zone_1', self.mock_extension_supported.return_value = True
updated_at='updated_at_1',
status='status_1',
state='state_1'
), mock.Mock(
binary='binary_2',
host='host_2',
zone='zone_2',
updated_at='updated_at_2',
status='status_2',
state='state_2'
)]
mock_service_enabled.return_value = True
response = cinder.Services().get(request) response = cinder.Services().get(request)
self.assertStatusCode(response, 200)
response_as_json = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_as_json['items'][0]['id'], 1)
self.assertEqual(response_as_json['items'][0]['binary'], 'binary_1')
self.assertEqual(response_as_json['items'][1]['id'], 2)
self.assertEqual(response_as_json['items'][1]['binary'], 'binary_2')
cc.service_list.assert_called_once_with(request)
mock_service_enabled.assert_called_once_with(request, 'volume')
@mock.patch.object(api.base, 'is_service_enabled') def _convert_service(service, idx):
def test_services_get_disabled(self, mock_service_enabled): return {'binary': service.binary,
'host': service.host,
'zone': service.zone,
'updated_at': service.updated_at,
'status': service.status,
'state': service.state,
'id': idx + 1}
self.assertStatusCode(response, 200)
self.assertEqual([_convert_service(s, i)
for i, s in enumerate(services)],
response.json['items'])
self.mock_service_list.assert_called_once_with(request)
self.mock_is_service_enabled.assert_called_once_with(request, 'volume')
self.mock_extension_supported.assert_called_once_with(request,
'Services')
@test.create_mocks({api.base: ['is_service_enabled']})
def test_services_get_disabled(self):
request = self.mock_rest_request(GET={}) request = self.mock_rest_request(GET={})
mock_service_enabled.return_value = False self.mock_is_service_enabled.return_value = False
response = cinder.Services().get(request) response = cinder.Services().get(request)
self.assertStatusCode(response, 501) self.assertStatusCode(response, 501)
self.mock_is_service_enabled.assert_called_once_with(request, 'volume')
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['is_volume_service_enabled',
def test_quota_sets_defaults_get_when_service_is_enabled(self, cc): 'default_quota_get']})
def test_quota_sets_defaults_get_when_service_is_enabled(self):
self.maxDiff = None self.maxDiff = None
filters = {'user': {'tenant_id': 'tenant'}} filters = {'user': {'tenant_id': 'tenant'}}
request = self.mock_rest_request(**{'GET': dict(filters)}) request = self.mock_rest_request(**{'GET': dict(filters)})
cc.is_service_enabled.return_value = True self.mock_is_volume_service_enabled.return_value = True
cc.default_quota_get.return_value = [Quota("volumes", 1), self.mock_default_quota_get.return_value = [Quota("volumes", 1),
Quota("snapshots", 2), Quota("snapshots", 2),
Quota("gigabytes", 3), Quota("gigabytes", 3),
Quota("some_other_1", 100), Quota("some_other_1", 100),
@ -378,15 +388,17 @@ class CinderRestTestCase(test.TestCase):
"display_name": "Yet Another", "display_name": "Yet Another",
"name": "yet_another"}]}) "name": "yet_another"}]})
cc.default_quota_get.assert_called_once_with(request, self.mock_is_volume_service_enabled.assert_called_once_with(request)
request.user.tenant_id) self.mock_default_quota_get.assert_called_once_with(
request, request.user.tenant_id)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['is_volume_service_enabled',
def test_quota_sets_defaults_get_when_service_is_disabled(self, cc): 'default_quota_get']})
def test_quota_sets_defaults_get_when_service_is_disabled(self):
filters = {'user': {'tenant_id': 'tenant'}} filters = {'user': {'tenant_id': 'tenant'}}
request = self.mock_rest_request(**{'GET': dict(filters)}) request = self.mock_rest_request(**{'GET': dict(filters)})
cc.is_volume_service_enabled.return_value = False self.mock_is_volume_service_enabled.return_value = False
response = cinder.DefaultQuotaSets().get(request) response = cinder.DefaultQuotaSets().get(request)
@ -394,35 +406,38 @@ class CinderRestTestCase(test.TestCase):
self.assertEqual(response.content.decode('utf-8'), self.assertEqual(response.content.decode('utf-8'),
'"Service Cinder is disabled."') '"Service Cinder is disabled."')
cc.default_quota_get.assert_not_called() self.mock_is_volume_service_enabled.assert_called_once_with(request)
self.mock_default_quota_get.assert_not_called()
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['is_volume_service_enabled',
def test_quota_sets_defaults_patch_when_service_is_enabled(self, cc): 'default_quota_update']})
def test_quota_sets_defaults_patch_when_service_is_enabled(self):
request = self.mock_rest_request(body=''' request = self.mock_rest_request(body='''
{"volumes": "15", "snapshots": "5000", {"volumes": "15", "snapshots": "5000",
"gigabytes": "5", "cores": "10"} "gigabytes": "5"}
''') ''')
cc.is_volume_service_enabled.return_value = True self.mock_is_volume_service_enabled.return_value = True
self.mock_default_quota_update.return_value = None
response = cinder.DefaultQuotaSets().patch(request) response = cinder.DefaultQuotaSets().patch(request)
self.assertStatusCode(response, 204) self.assertStatusCode(response, 204)
self.assertEqual(response.content.decode('utf-8'), '') self.assertEqual(response.content.decode('utf-8'), '')
cc.default_quota_update.assert_called_once_with(request, self.mock_is_volume_service_enabled.assert_called_once_with(request)
volumes='15', self.mock_default_quota_update.assert_called_once_with(
snapshots='5000', request, volumes='15', snapshots='5000', gigabytes='5')
gigabytes='5')
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['is_volume_service_enabled',
def test_quota_sets_defaults_patch_when_service_is_disabled(self, cc): 'default_quota_update']})
def test_quota_sets_defaults_patch_when_service_is_disabled(self):
request = self.mock_rest_request(body=''' request = self.mock_rest_request(body='''
{"volumes": "15", "snapshots": "5000", {"volumes": "15", "snapshots": "5000",
"gigabytes": "5", "cores": "10"} "gigabytes": "5"}
''') ''')
cc.is_volume_service_enabled.return_value = False self.mock_is_volume_service_enabled.return_value = False
response = cinder.DefaultQuotaSets().patch(request) response = cinder.DefaultQuotaSets().patch(request)
@ -430,72 +445,62 @@ class CinderRestTestCase(test.TestCase):
self.assertEqual(response.content.decode('utf-8'), self.assertEqual(response.content.decode('utf-8'),
'"Service Cinder is disabled."') '"Service Cinder is disabled."')
cc.default_quota_update.assert_not_called() self.mock_is_volume_service_enabled.assert_called_once_with(request)
self.mock_default_quota_update.assert_not_called()
@mock.patch.object(cinder.api, 'cinder')
@mock.patch.object(cinder, 'quotas')
def test_quota_sets_patch(self, qc, cc):
quota_set = self.cinder_quotas.list()[0]
quota_data = {}
for quota in quota_set:
quota_data[quota.name] = quota.limit
@test.create_mocks({api.cinder: ['tenant_quota_update',
'is_volume_service_enabled'],
quotas: ['get_disabled_quotas']})
def test_quota_sets_patch(self):
request = self.mock_rest_request(body=''' request = self.mock_rest_request(body='''
{"volumes": "15", "snapshots": "5000", {"volumes": "15", "snapshots": "5000",
"gigabytes": "5", "cores": "10"} "gigabytes": "5"}
''') ''')
qc.get_disabled_quotas.return_value = [] self.mock_get_disabled_quotas.return_value = []
qc.CINDER_QUOTA_FIELDS = {n for n in quota_data} self.mock_is_volume_service_enabled.return_value = True
cc.is_volume_service_enabled.return_value = True self.mock_tenant_quota_update.return_value = None
response = cinder.QuotaSets().patch(request, 'spam123') response = cinder.QuotaSets().patch(request, 'spam123')
self.assertStatusCode(response, 204) self.assertStatusCode(response, 204)
self.assertEqual(response.content.decode('utf-8'), '') self.assertEqual(response.content.decode('utf-8'), '')
cc.tenant_quota_update.assert_called_once_with(request, 'spam123', self.mock_get_disabled_quotas.assert_called_once_with(request)
volumes='15', self.mock_is_volume_service_enabled.assert_called_once_with(request)
snapshots='5000', self.mock_tenant_quota_update.assert_called_once_with(
gigabytes='5') request, 'spam123', volumes='15', snapshots='5000', gigabytes='5')
@mock.patch.object(cinder.api, 'cinder')
@mock.patch.object(cinder, 'quotas')
def test_quota_sets_when_service_is_disabled(self, qc, cc):
quota_set = self.cinder_quotas.list()[0]
quota_data = {}
for quota in quota_set:
quota_data[quota.name] = quota.limit
@test.create_mocks({api.cinder: ['tenant_quota_update',
'is_volume_service_enabled'],
quotas: ['get_disabled_quotas']})
def test_quota_sets_when_service_is_disabled(self):
request = self.mock_rest_request(body=''' request = self.mock_rest_request(body='''
{"volumes": "15", "snapshots": "5000", {"volumes": "15", "snapshots": "5000",
"gigabytes": "5", "cores": "10"} "gigabytes": "5"}
''') ''')
qc.get_disabled_quotas.return_value = [] self.mock_get_disabled_quotas.return_value = []
qc.CINDER_QUOTA_FIELDS = {n for n in quota_data} self.mock_is_volume_service_enabled.return_value = False
cc.is_volume_service_enabled.return_value = False self.mock_tenant_quota_update.return_value = None
response = cinder.QuotaSets().patch(request, 'spam123') response = cinder.QuotaSets().patch(request, 'spam123')
self.assertStatusCode(response, 501) self.assertStatusCode(response, 501)
self.assertEqual(response.content.decode('utf-8'), self.assertEqual(response.content.decode('utf-8'),
'"Service Cinder is disabled."') '"Service Cinder is disabled."')
cc.tenant_quota_update.assert_not_called() self.mock_tenant_quota_update.assert_not_called()
self.mock_get_disabled_quotas.assert_called_once_with(request)
self.mock_is_volume_service_enabled.assert_called_once_with(request)
@mock.patch.object(cinder.api, 'cinder') @test.create_mocks({api.cinder: ['availability_zone_list']})
def test_availability_zones_get(self, cc): def test_availability_zones_get(self):
request = self.mock_rest_request(GET={}) request = self.mock_rest_request(GET={})
mock_az = mock.Mock() az_list = self.cinder_availability_zones.list()
mock_az.to_dict.return_value = { self.mock_availability_zone_list.return_value = az_list
'name': 'cinder',
'status': 'available'
}
cc.availability_zone_list.return_value = [mock_az]
response = cinder.AvailabilityZones().get(request) response = cinder.AvailabilityZones().get(request)
self.assertStatusCode(response, 200) self.assertStatusCode(response, 200)
response_as_json = json.loads(response.content.decode('utf-8')) self.assertEqual([az.to_dict() for az in az_list],
self.assertEqual(response_as_json['items'][0]['name'], 'cinder') response.json['items'])
cc.availability_zone_list.assert_called_once_with(request, False) self.mock_availability_zone_list.assert_called_once_with(
request, False)