Use requests-mock for the list/add/remove volume types tests

Change-Id: I03aac75375b0f34678f8dc0b953fb2a0cd797150
Signed-off-by: Rosario Di Somma <rosario.disomma@gmail.com>
This commit is contained in:
Rosario Di Somma 2017-04-21 16:16:40 -07:00
parent 0bfdaf4e0b
commit 258a6a33ba
1 changed files with 118 additions and 47 deletions

View File

@ -13,35 +13,42 @@
# under the License.
import mock
import testtools
import shade
from shade.tests.unit import base
class TestVolumeAccess(base.TestCase):
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_list_volume_types(self, mock_cinder):
class TestVolumeAccess(base.RequestsMockTestCase):
def test_list_volume_types(self):
volume_type = dict(
id='voltype01', description='volume type description',
name='name', is_public=False)
mock_cinder.volume_types.list.return_value = [volume_type]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]})])
self.assertTrue(self.cloud.list_volume_types())
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_get_volume_type(self, mock_cinder):
def test_get_volume_type(self):
volume_type = dict(
id='voltype01', description='volume type description', name='name',
is_public=False)
mock_cinder.volume_types.list.return_value = [volume_type]
volume_type_got = self.cloud.get_volume_type('name')
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]})])
volume_type_got = self.cloud.get_volume_type(volume_type['name'])
self.assertEqual(volume_type_got.id, volume_type['id'])
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_get_volume_type_access(self, mock_cinder):
def test_get_volume_type_access(self):
volume_type = dict(
id='voltype01', description='volume type description', name='name',
is_public=False)
@ -49,14 +56,24 @@ class TestVolumeAccess(base.TestCase):
dict(volume_type_id='voltype01', name='name', project_id='prj01'),
dict(volume_type_id='voltype01', name='name', project_id='prj02')
]
mock_cinder.volume_types.list.return_value = [volume_type]
mock_cinder.volume_type_access.list.return_value = volume_type_access
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'],
'os-volume-type-access']),
json={'volume_type_access': volume_type_access})])
self.assertEqual(
len(self.op_cloud.get_volume_type_access('name')), 2)
len(self.op_cloud.get_volume_type_access(volume_type['name'])), 2)
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_remove_volume_type_access(self, mock_cinder):
def test_remove_volume_type_access(self):
volume_type = dict(
id='voltype01', description='volume type description', name='name',
is_public=False)
@ -65,26 +82,55 @@ class TestVolumeAccess(base.TestCase):
project_002 = dict(volume_type_id='voltype01', name='name',
project_id='prj02')
volume_type_access = [project_001, project_002]
mock_cinder.volume_types.list.return_value = [volume_type]
mock_cinder.volume_type_access.list.return_value = volume_type_access
def _fake_remove(*args, **kwargs):
volume_type_access.pop()
mock_cinder.volume_type_access.remove_project_access.side_effect = \
_fake_remove
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'],
'os-volume-type-access']),
json={'volume_type_access': volume_type_access}),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'], qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='POST',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'], 'action']),
json={'removeProjectAccess': {
'project': project_001['project_id']}},
validate=dict(
json={'removeProjectAccess': {
'project': project_001['project_id']}})),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'],
'os-volume-type-access']),
json={'volume_type_access': [project_001]})])
self.assertEqual(
len(self.op_cloud.get_volume_type_access(
volume_type['name'])), 2)
self.op_cloud.remove_volume_type_access(
volume_type['name'], project_001['project_id'])
self.assertEqual(
len(self.op_cloud.get_volume_type_access('name')), 1)
len(self.op_cloud.get_volume_type_access(volume_type['name'])), 1)
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_add_volume_type_access(self, mock_cinder):
def test_add_volume_type_access(self):
volume_type = dict(
id='voltype01', description='volume type description', name='name',
is_public=False)
@ -92,31 +138,56 @@ class TestVolumeAccess(base.TestCase):
project_id='prj01')
project_002 = dict(volume_type_id='voltype01', name='name',
project_id='prj02')
volume_type_access = [project_001]
mock_cinder.volume_types.list.return_value = [volume_type]
mock_cinder.volume_type_access.list.return_value = volume_type_access
mock_cinder.volume_type_access.add_project_access.return_value = None
def _fake_add(*args, **kwargs):
volume_type_access.append(project_002)
mock_cinder.volume_type_access.add_project_access.side_effect = \
_fake_add
volume_type_access = [project_001, project_002]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='POST',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'], 'action']),
json={'addProjectAccess': {
'project': project_002['project_id']}},
validate=dict(
json={'addProjectAccess': {
'project': project_002['project_id']}})),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]}),
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types', volume_type['id'],
'os-volume-type-access']),
json={'volume_type_access': volume_type_access})])
self.op_cloud.add_volume_type_access(
volume_type['name'], project_002['project_id'])
self.assertEqual(
len(self.op_cloud.get_volume_type_access('name')), 2)
len(self.op_cloud.get_volume_type_access(volume_type['name'])), 2)
self.assert_calls()
@mock.patch.object(shade.OpenStackCloud, 'cinder_client')
def test_add_volume_type_access_missing(self, mock_cinder):
def test_add_volume_type_access_missing(self):
volume_type = dict(
id='voltype01', description='volume type description', name='name',
is_public=False)
project_001 = dict(volume_type_id='voltype01', name='name',
project_id='prj01')
mock_cinder.volume_types.list.return_value = [volume_type]
self.register_uris([
dict(method='GET',
uri=self.get_mock_url(
'volumev2', 'public',
append=['types'],
qs_elements=['is_public=None']),
json={'volume_types': [volume_type]})])
with testtools.ExpectedException(shade.OpenStackCloudException,
"VolumeType not found: MISSING"):
self.op_cloud.add_volume_type_access(
"MISSING", project_001['project_id'])
self.assert_calls()