Fix update share network tests and skip exception issue

After fixing a bug that was allowing share network update when it
shouldn't, the test that validates the share network creation and
update started failing. An additional validation was introduced and
it caused the new bug. This patch fixes the test and make it work
properly.
In addition this patch fixes test classes that are raising a skip
exception inside the setUpClass function and failing due to testtool
issue #272.

Closes-Bug: #1849728
Closes-Bug: #1849377
Related-Bug: #1846836
Co-Author: Carlos Eduardo <ces.eduardo98@gmail.com>
Change-Id: I5b27ee4a9e844ea48dc9324bcf38f5767223717f
This commit is contained in:
Douglas Viroel 2019-10-24 17:48:22 -03:00
parent e6e8cad240
commit 0ca91287ac
17 changed files with 196 additions and 212 deletions

View File

@ -202,7 +202,7 @@ class BaseTestCase(base.ClientTestBase):
snapshot_support=None,
create_share_from_snapshot=None,
revert_to_snapshot=None, mount_snapshot=None,
is_public=True, client=None, cleanup_in_class=True,
is_public=True, client=None, cleanup_in_class=False,
microversion=None, extra_specs=None,
description=None):
if client is None:
@ -257,7 +257,7 @@ class BaseTestCase(base.ClientTestBase):
neutron_net_id=None,
neutron_subnet_id=None,
availability_zone=None, client=None,
cleanup_in_class=True, microversion=None):
cleanup_in_class=False, microversion=None):
if client is None:
client = cls.get_admin_client()
share_network = client.create_share_network(
@ -284,7 +284,7 @@ class BaseTestCase(base.ClientTestBase):
def add_share_network_subnet(cls, share_network,
neutron_net_id=None, neutron_subnet_id=None,
availability_zone=None, client=None,
cleanup_in_class=True, microversion=None):
cleanup_in_class=False, microversion=None):
if client is None:
client = cls.get_admin_client()
share_network_subnet = client.add_share_network_subnet(

View File

@ -22,12 +22,10 @@ from manilaclient.tests.functional import base
@ddt.ddt
class ExportLocationReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(ExportLocationReadWriteTest, cls).setUpClass()
cls.share = cls.create_share(
client=cls.get_user_client(),
cleanup_in_class=True)
def setUp(self):
super(ExportLocationReadWriteTest, self).setUp()
self.share = self.create_share(
client=self.get_user_client())
@ddt.data('admin', 'user')
def test_list_share_export_locations(self, role):

View File

@ -31,10 +31,9 @@ class MessagesReadOnlyTest(base.BaseTestCase):
@ddt.ddt
class MessagesReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(MessagesReadWriteTest, cls).setUpClass()
cls.message = cls.create_message(cleanup_in_class=True)
def setUp(self):
super(MessagesReadWriteTest, self).setUp()
self.message = self.create_message()
def test_list_messages(self):
self.skip_if_microversion_not_supported('2.37')

View File

@ -50,8 +50,7 @@ class QuotasReadWriteTest(base.BaseTestCase):
name=data_utils.rand_name("manilaclient_functional_test"),
driver_handles_share_servers=False,
is_public=True,
microversion=self.microversion,
cleanup_in_class=False,
microversion=self.microversion
)
self.st_id = self.share_type["ID"]

View File

@ -22,17 +22,16 @@ from manilaclient.tests.functional import base
@ddt.ddt
class SecurityServiceReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SecurityServiceReadWriteTest, cls).setUpClass()
cls.name = data_utils.rand_name('autotest')
cls.description = 'fake_description'
cls.user = 'fake_user'
cls.password = 'fake_password'
cls.server = 'fake_server'
cls.domain = 'fake_domain'
cls.dns_ip = '1.2.3.4'
cls.ou = 'fake_ou'
def setUp(self):
super(SecurityServiceReadWriteTest, self).setUp()
self.name = data_utils.rand_name('autotest')
self.description = 'fake_description'
self.user = 'fake_user'
self.password = 'fake_password'
self.server = 'fake_server'
self.domain = 'fake_domain'
self.dns_ip = '1.2.3.4'
self.ou = 'fake_ou'
@ddt.data(
{'name': 'test_name'},

View File

@ -30,34 +30,32 @@ class ShareAccessReadWriteBase(base.BaseTestCase):
protocol = None
access_level = None
@classmethod
def setUpClass(cls):
super(ShareAccessReadWriteBase, cls).setUpClass()
if cls.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % cls.protocol
raise cls.skipException(message)
if cls.access_level not in CONF.access_levels_mapping.get(
cls.protocol, '').split(' '):
raise cls.skipException("%(level)s tests for %(protocol)s share "
"access are disabled." % {
'level': cls.access_level,
'protocol': cls.protocol
})
cls.access_types = CONF.access_types_mapping.get(
cls.protocol, '').split(' ')
if not cls.access_types:
raise cls.skipException("No access levels were provided for %s "
"share access tests." % cls.protoco)
def setUp(self):
super(ShareAccessReadWriteBase, self).setUp()
if self.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % self.protocol
raise self.skipException(message)
if self.access_level not in CONF.access_levels_mapping.get(
self.protocol, '').split(' '):
raise self.skipException("%(level)s tests for %(protocol)s share "
"access are disabled." % {
'level': self.access_level,
'protocol': self.protocol
})
self.access_types = CONF.access_types_mapping.get(
self.protocol, '').split(' ')
if not self.access_types:
raise self.skipException("No access levels were provided for %s "
"share access tests." % self.protocol)
cls.share = cls.create_share(share_protocol=cls.protocol,
public=True,
cleanup_in_class=True)
cls.share_id = cls.share['id']
self.share = self.create_share(share_protocol=self.protocol,
public=True)
self.share_id = self.share['id']
# NOTE(vponomaryov): increase following int range when significant
# amount of new tests is added.
int_range = range(20, 50)
cls.access_to = {
self.access_to = {
# NOTE(vponomaryov): list of unique values is required for ability
# to create lots of access rules for one share using different
# API microversions.

View File

@ -24,19 +24,18 @@ from tempest.lib import exceptions
@utils.skip_if_microversion_not_supported('2.51')
class ShareNetworkSubnetsReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(ShareNetworkSubnetsReadWriteTest, cls).setUpClass()
cls.name = data_utils.rand_name('autotest')
cls.description = 'fake_description'
cls.neutron_net_id = 'fake_neutron_net_id'
cls.neutron_subnet_id = 'fake_neutron_subnet_id'
def setUp(self):
super(ShareNetworkSubnetsReadWriteTest, self).setUp()
self.name = data_utils.rand_name('autotest')
self.description = 'fake_description'
self.neutron_net_id = 'fake_neutron_net_id'
self.neutron_subnet_id = 'fake_neutron_subnet_id'
cls.sn = cls.create_share_network(
name=cls.name,
description=cls.description,
neutron_net_id=cls.neutron_net_id,
neutron_subnet_id=cls.neutron_subnet_id,
self.sn = self.create_share_network(
name=self.name,
description=self.description,
neutron_net_id=self.neutron_net_id,
neutron_subnet_id=self.neutron_subnet_id,
)
def test_get_share_network_subnet(self):

View File

@ -26,19 +26,18 @@ from manilaclient.tests.functional import utils
@ddt.ddt
class ShareNetworksReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(ShareNetworksReadWriteTest, cls).setUpClass()
cls.name = data_utils.rand_name('autotest')
cls.description = 'fake_description'
cls.neutron_net_id = 'fake_neutron_net_id'
cls.neutron_subnet_id = 'fake_neutron_subnet_id'
def setUp(self):
super(ShareNetworksReadWriteTest, self).setUp()
self.name = data_utils.rand_name('autotest')
self.description = 'fake_description'
self.neutron_net_id = 'fake_neutron_net_id'
self.neutron_subnet_id = 'fake_neutron_subnet_id'
cls.sn = cls.create_share_network(
name=cls.name,
description=cls.description,
neutron_net_id=cls.neutron_net_id,
neutron_subnet_id=cls.neutron_subnet_id,
self.sn = self.create_share_network(
name=self.name,
description=self.description,
neutron_net_id=self.neutron_net_id,
neutron_subnet_id=self.neutron_subnet_id,
)
@ddt.data(
@ -125,33 +124,56 @@ class ShareNetworksReadWriteTest(base.BaseTestCase):
self.assertEqual(self.neutron_net_id, get['neutron_net_id'])
self.assertEqual(self.neutron_subnet_id, get['neutron_subnet_id'])
def _get_expected_update_data(self, net_data, net_creation_data):
# NOTE(dviroel): When subnets are supported, the outputs are converted
# from string to literal structures in order to process the content of
# 'share_network_subnets' field.
default_return_value = (
None if utils.share_network_subnets_are_supported() else 'None')
expected_nn_id = (
default_return_value
if net_data.get('neutron_net_id')
else net_creation_data.get('neutron_net_id', default_return_value))
expected_nsn_id = (
default_return_value
if net_data.get('neutron_subnet_id')
else net_creation_data.get('neutron_subnet_id',
default_return_value))
return expected_nn_id, expected_nsn_id
@ddt.data(
{'name': data_utils.rand_name('autotest_share_network_name')},
{'description': 'fake_description'},
{'neutron_net_id': 'fake_neutron_net_id',
'neutron_subnet_id': 'fake_neutron_subnet_id'},
{'name': '""'},
{'description': '""'},
{'neutron_net_id': '""'},
{'neutron_subnet_id': '""'},
({'name': data_utils.rand_name('autotest_share_network_name')}, {}),
({'description': 'fake_description'}, {}),
({'neutron_net_id': 'fake_neutron_net_id',
'neutron_subnet_id': 'fake_neutron_subnet_id'}, {}),
({'name': '""'}, {}),
({'description': '""'}, {}),
({'neutron_net_id': '""'},
{'neutron_net_id': 'fake_nn_id', 'neutron_subnet_id': 'fake_nsn_id'}),
({'neutron_subnet_id': '""'},
{'neutron_net_id': 'fake_nn_id', 'neutron_subnet_id': 'fake_nsn_id'})
)
def test_create_update_share_network(self, net_data):
sn = self.create_share_network(cleanup_in_class=False)
@ddt.unpack
def test_create_update_share_network(self, net_data, net_creation_data):
sn = self.create_share_network(
cleanup_in_class=False, **net_creation_data)
update = self.admin_client.update_share_network(sn['id'], **net_data)
expected_nn_id, expected_nsn_id = self._get_expected_update_data(
net_data, net_creation_data)
expected_data = {
'name': 'None',
'description': 'None',
'neutron_net_id': 'None',
'neutron_subnet_id': 'None',
'neutron_net_id': expected_nn_id,
'neutron_subnet_id': expected_nsn_id,
}
subnet_keys = []
if utils.share_network_subnets_are_supported():
subnet_keys = ['neutron_net_id', 'neutron_subnet_id']
subnet = ast.literal_eval(update['share_network_subnets'])
expected_data['neutron_net_id'] = None
expected_data['neutron_subnet_id'] = None
update_values = dict([(k, v) for k, v in net_data.items()
if v != '""'])

View File

@ -29,16 +29,11 @@ CONF = config.CONF
@utils.skip_if_microversion_not_supported('2.47')
class ShareReplicaExportLocationsTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(ShareReplicaExportLocationsTest, cls).setUpClass()
def _create_share_and_replica(self):
replication_type = CONF.replication_type
share_type = self.create_share_type(
driver_handles_share_servers=False,
extra_specs={'replication_type': replication_type},
cleanup_in_class=False)
extra_specs={'replication_type': replication_type})
share = self.create_share(share_type=share_type['ID'],
client=self.get_user_client())
share_replica = self.create_share_replica(share['id'])

View File

@ -30,10 +30,9 @@ CONF = config.CONF
@ddt.ddt
class ShareServersReadOnlyTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(ShareServersReadOnlyTest, cls).setUpClass()
cls.client = cls.get_admin_client()
def setUp(self):
super(ShareServersReadOnlyTest, self).setUp()
self.client = self.get_admin_client()
def test_share_server_list(self):
self.client.list_share_servers()
@ -68,20 +67,19 @@ class ShareServersReadWriteBase(base.BaseTestCase):
protocol = None
@classmethod
def setUpClass(cls):
super(ShareServersReadWriteBase, cls).setUpClass()
def setUp(self):
super(ShareServersReadWriteBase, self).setUp()
if not CONF.run_share_servers_tests:
message = "share-servers tests are disabled."
raise cls.skipException(message)
if cls.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % cls.protocol
raise cls.skipException(message)
raise self.skipException(message)
if self.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % self.protocol
raise self.skipException(message)
cls.client = cls.get_admin_client()
if not cls.client.share_network:
self.client = self.get_admin_client()
if not self.client.share_network:
message = "Can run only with DHSS=True mode"
raise cls.skipException(message)
raise self.skipException(message)
def _create_share_and_share_network(self):
name = data_utils.rand_name('autotest_share_name')

View File

@ -29,24 +29,22 @@ CONF = config.CONF
class SharesReadWriteBase(base.BaseTestCase):
protocol = None
@classmethod
def setUpClass(cls):
super(SharesReadWriteBase, cls).setUpClass()
if cls.protocol not in CONF.enable_protocols:
message = "%s tests are disabled" % cls.protocol
raise cls.skipException(message)
cls.name = data_utils.rand_name('autotest_share_name')
cls.description = data_utils.rand_name('autotest_share_description')
def setUp(self):
super(SharesReadWriteBase, self).setUp()
if self.protocol not in CONF.enable_protocols:
message = "%s tests are disabled" % self.protocol
raise self.skipException(message)
self.name = data_utils.rand_name('autotest_share_name')
self.description = data_utils.rand_name('autotest_share_description')
# NOTE(vponomaryov): following share is used only in one test
# until tests for snapshots appear.
cls.share = cls.create_share(
share_protocol=cls.protocol,
self.share = self.create_share(
share_protocol=self.protocol,
size=1,
name=cls.name,
description=cls.description,
client=cls.get_user_client(),
cleanup_in_class=True)
name=self.name,
description=self.description,
client=self.get_user_client())
def test_create_delete_share(self):
name = data_utils.rand_name('autotest_share_name')
@ -96,29 +94,28 @@ class SharesReadWriteBase(base.BaseTestCase):
@ddt.ddt
class SharesTestMigration(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SharesTestMigration, cls).setUpClass()
def setUp(self):
super(SharesTestMigration, self).setUp()
cls.old_type = cls.create_share_type(
self.old_type = self.create_share_type(
data_utils.rand_name('test_share_type'),
driver_handles_share_servers=True)
cls.new_type = cls.create_share_type(
self.new_type = self.create_share_type(
data_utils.rand_name('test_share_type'),
driver_handles_share_servers=True)
cls.error_type = cls.create_share_type(
self.error_type = self.create_share_type(
data_utils.rand_name('test_share_type'),
driver_handles_share_servers=True,
extra_specs={'cause_error': 'no_valid_host'})
cls.old_share_net = cls.get_user_client().get_share_network(
cls.get_user_client().share_network)
self.old_share_net = self.get_user_client().get_share_network(
self.get_user_client().share_network)
share_net_info = (
utils.get_default_subnet(cls.get_user_client(),
cls.old_share_net['id'])
utils.get_default_subnet(self.get_user_client(),
self.old_share_net['id'])
if utils.share_network_subnets_are_supported()
else cls.old_share_net)
cls.new_share_net = cls.create_share_network(
else self.old_share_net)
self.new_share_net = self.create_share_network(
neutron_net_id=share_net_info['neutron_net_id'],
neutron_subnet_id=share_net_info['neutron_subnet_id'])

View File

@ -112,47 +112,43 @@ class SharesListReadOnlyTest(base.BaseTestCase):
@ddt.ddt
class SharesListReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SharesListReadWriteTest, cls).setUpClass()
cls.private_name = data_utils.rand_name('autotest_share_name')
cls.private_description = data_utils.rand_name(
def setUp(self):
super(SharesListReadWriteTest, self).setUp()
self.private_name = data_utils.rand_name('autotest_share_name')
self.private_description = data_utils.rand_name(
'autotest_share_description')
cls.public_name = data_utils.rand_name('autotest_public_share_name')
cls.public_description = data_utils.rand_name(
self.public_name = data_utils.rand_name('autotest_public_share_name')
self.public_description = data_utils.rand_name(
'autotest_public_share_description')
cls.admin_private_name = data_utils.rand_name(
self.admin_private_name = data_utils.rand_name(
'autotest_admin_private_share_name')
cls.admin_private_description = data_utils.rand_name(
self.admin_private_description = data_utils.rand_name(
'autotest_admin_private_share_description')
cls.admin_private_share = cls.create_share(
name=cls.admin_private_name,
description=cls.admin_private_description,
self.admin_private_share = self.create_share(
name=self.admin_private_name,
description=self.admin_private_description,
public=False,
cleanup_in_class=True,
client=None,
wait_for_creation=False)
cls.private_share = cls.create_share(
name=cls.private_name,
description=cls.private_description,
self.private_share = self.create_share(
name=self.private_name,
description=self.private_description,
public=False,
cleanup_in_class=True,
client=cls.get_user_client(),
client=self.get_user_client(),
wait_for_creation=False)
cls.public_share = cls.create_share(
name=cls.public_name,
description=cls.public_description,
self.public_share = self.create_share(
name=self.public_name,
description=self.public_description,
public=True,
client=cls.get_user_client(),
cleanup_in_class=True)
client=self.get_user_client())
for share_id in (cls.private_share['id'], cls.public_share['id'],
cls.admin_private_share['id']):
cls.get_admin_client().wait_for_resource_status(
for share_id in (self.private_share['id'], self.public_share['id'],
self.admin_private_share['id']):
self.get_admin_client().wait_for_resource_status(
share_id, constants.STATUS_AVAILABLE)
def _list_shares(self, filters=None):
@ -319,8 +315,7 @@ class SharesListReadWriteTest(base.BaseTestCase):
name=u'共享名称',
description=u'共享描述',
public=True,
client=self.get_user_client(),
cleanup_in_class=True)
client=self.get_user_client())
filters = {'name~': u'名称'}
shares = self.user_client.list_shares(filters=filters)
self.assertGreater(len(shares), 0)

View File

@ -21,19 +21,15 @@ from manilaclient.tests.functional import base
@ddt.ddt
class SharesMetadataReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SharesMetadataReadWriteTest, cls).setUpClass()
cls.share = cls.create_share(
client=cls.get_user_client(),
cleanup_in_class=True)
def setUp(self):
super(SharesMetadataReadWriteTest, self).setUp()
self.share = self.create_share(client=self.get_user_client())
def test_set_metadata_in_share_creation(self):
md = {"key1": "value1", "key2": "value2"}
# Create share with metadata
share = self.create_share(
metadata=md, cleanup_in_class=False, client=self.get_user_client())
share = self.create_share(metadata=md, client=self.get_user_client())
# Read share metadata
metadata = self.user_client.get_share_metadata(share["id"])

View File

@ -26,25 +26,23 @@ CONF = config.CONF
class SnapshotAccessReadBase(base.BaseTestCase):
protocol = None
@classmethod
def setUpClass(cls):
super(SnapshotAccessReadBase, cls).setUpClass()
if cls.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % cls.protocol
raise cls.skipException(message)
cls.access_types = CONF.access_types_mapping.get(
cls.protocol, '').split(' ')
if not cls.access_types:
raise cls.skipException("No access types were provided for %s "
"snapshot access tests." % cls.protocol)
def setUp(self):
super(SnapshotAccessReadBase, self).setUp()
if self.protocol not in CONF.enable_protocols:
message = "%s tests are disabled." % self.protocol
raise self.skipException(message)
self.access_types = CONF.access_types_mapping.get(
self.protocol, '').split(' ')
if not self.access_types:
raise self.skipException("No access types were provided for %s "
"snapshot access tests." % self.protocol)
cls.share = cls.create_share(share_protocol=cls.protocol,
public=True,
cleanup_in_class=True,
client=cls.get_user_client())
self.share = self.create_share(share_protocol=self.protocol,
public=True,
client=self.get_user_client())
int_range = range(0, 10)
cls.access_to = {
self.access_to = {
'ip': ['99.88.77.%d' % i for i in int_range],
'user': ['foo_user_%d' % i for i in int_range],
'cert': ['tenant_%d.example.com' % i for i in int_range],

View File

@ -30,15 +30,12 @@ CONF = config.CONF
@utils.skip_if_microversion_not_supported('2.19')
class SnapshotInstancesTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SnapshotInstancesTest, cls).setUpClass()
cls.share = cls.create_share(
client=cls.get_user_client(),
cleanup_in_class=True)
cls.snapshot = cls.create_snapshot(share=cls.share['id'],
client=cls.get_user_client(),
cleanup_in_class=True)
def setUp(self):
super(SnapshotInstancesTest, self).setUp()
self.share = self.create_share(
client=self.get_user_client())
self.snapshot = self.create_snapshot(share=self.share['id'],
client=self.get_user_client())
def test_list_all_snapshot_instances(self):
snapshot_instances = self.admin_client.list_snapshot_instances()

View File

@ -31,15 +31,12 @@ CONF = config.CONF
@utils.skip_if_microversion_not_supported('2.32')
class SnapshotInstanceExportLocationReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SnapshotInstanceExportLocationReadWriteTest, cls).setUpClass()
cls.share = cls.create_share(
client=cls.get_user_client(),
cleanup_in_class=True)
cls.snapshot = cls.create_snapshot(share=cls.share['id'],
client=cls.get_user_client(),
cleanup_in_class=True)
def setUp(self):
super(SnapshotInstanceExportLocationReadWriteTest, self).setUp()
self.share = self.create_share(
client=self.get_user_client())
self.snapshot = self.create_snapshot(share=self.share['id'],
client=self.get_user_client())
def test_get_snapshot_instance_export_location(self):
client = self.admin_client

View File

@ -31,15 +31,12 @@ CONF = config.CONF
@utils.skip_if_microversion_not_supported('2.32')
class SnapshotExportLocationReadWriteTest(base.BaseTestCase):
@classmethod
def setUpClass(cls):
super(SnapshotExportLocationReadWriteTest, cls).setUpClass()
cls.share = cls.create_share(
client=cls.get_user_client(),
cleanup_in_class=True)
cls.snapshot = cls.create_snapshot(share=cls.share['id'],
client=cls.get_user_client(),
cleanup_in_class=True)
def setUp(self):
super(SnapshotExportLocationReadWriteTest, self).setUp()
self.share = self.create_share(
client=self.get_user_client())
self.snapshot = self.create_snapshot(share=self.share['id'],
client=self.get_user_client())
@ddt.data('admin', 'user')
def test_get_snapshot_export_location(self, role):