openstack/cinder
Code Issues Proposed changes

Use assertEqual/Less/Greater/In/IsNone

Browse Source 
Instead of using assertFalse(A * B) / assertTrue(A * B),
developers should use assertEqual/NotEqual/Less/LessEqual
/Greater/GreaterEqual/In/NotIn/IsNone/IsNotNone
The * operator: '==', '!=', '<', '<=', '>', '>='

Change-Id: I11b60a93c089f688c6b243da01d20c205c4f2df9
This commit is contained in:
Béla Vancsics 2016-07-04 10:42:59 +02:00
parent 6f2112caa8
commit bcaa956028
32 changed files with 211 additions and 237 deletions

2
cinder/tests/compliance/test_backup_drivers.py

@ -34,7 +34,7 @@ class TestBackupDrivers(test.TestCase):
returns at least one registered driver, else the compliance test will
never even run.
"""
self.assertTrue(len(BACKUP_DRIVERS) > 0)
self.assertGreater(len(BACKUP_DRIVERS), 0)
@ddt.data(*BACKUP_DRIVERS)
def test_backup_driver_compliance(self, driver):

2
cinder/tests/compliance/test_fczm_drivers.py

@ -34,7 +34,7 @@ class TestFibreChannelZoneManagerDrivers(test.TestCase):
returns at least one registered driver, else the compliance test will
never even run.
"""
self.assertTrue(len(FCZM_DRIVERS) > 0)
self.assertGreater(len(FCZM_DRIVERS), 0)
@ddt.data(*FCZM_DRIVERS)
def test_fczm_driver_compliance(self, driver):

2
cinder/tests/compliance/test_volume_drivers.py

@ -34,7 +34,7 @@ class TestVolumeDrivers(test.TestCase):
returns at least one registered driver, else the compliance test will
never even run.
"""
self.assertTrue(len(VOLUME_DRIVERS) > 0)
self.assertGreater(len(VOLUME_DRIVERS), 0)
@ddt.data(*VOLUME_DRIVERS)
def test_volume_driver_compliance(self, driver):

18
cinder/tests/unit/api/openstack/test_api_version_request.py

@ -87,15 +87,15 @@ class APIVersionRequestTests(test.TestCase):
v4 = api_version_request.APIVersionRequest('2.0')
v_null = api_version_request.APIVersionRequest()
self.assertTrue(v1 < v2)
self.assertTrue(v1 <= v2)
self.assertTrue(v3 > v2)
self.assertTrue(v3 >= v2)
self.assertTrue(v1 != v2)
self.assertTrue(v1 == v4)
self.assertTrue(v1 != v_null)
self.assertTrue(v_null == v_null)
self.assertFalse(v1 == '2.0')
self.assertLess(v1, v2)
self.assertLessEqual(v1, v2)
self.assertGreater(v3, v2)
self.assertGreaterEqual(v3, v2)
self.assertNotEqual(v1, v2)
self.assertEqual(v1, v4)
self.assertNotEqual(v1, v_null)
self.assertEqual(v_null, v_null)
self.assertNotEqual('2.0', v1)
def test_version_matches(self):
v1 = api_version_request.APIVersionRequest('2.0')

10
cinder/tests/unit/api/v2/test_volumes.py

@ -974,7 +974,7 @@ class VolumeApiTest(test.TestCase):
self.assertEqual('/v2/%s/volumes/detail' % fake.PROJECT_ID,
href_parts.path)
params = urllib.parse.parse_qs(href_parts.query)
self.assertTrue('marker' in params)
self.assertIn('marker', params)
self.assertEqual('1', params['limit'][0])
def test_volume_detail_limit_negative(self):
@ -1125,7 +1125,7 @@ class VolumeApiTest(test.TestCase):
viewable_admin_meta=False,
offset=0):
self.assertTrue(filters['no_migration_targets'])
self.assertFalse('all_tenants' in filters)
self.assertNotIn('all_tenants', filters)
return [stubs.stub_volume(fake.VOLUME_ID, display_name='vol1')]
def stub_volume_get_all(context, marker, limit,
@ -1151,7 +1151,7 @@ class VolumeApiTest(test.TestCase):
filters=None,
viewable_admin_meta=False,
offset=0):
self.assertFalse('no_migration_targets' in filters)
self.assertNotIn('no_migration_targets', filters)
return [stubs.stub_volume(fake.VOLUME_ID, display_name='vol2')]
def stub_volume_get_all2(context, marker, limit,
@ -1181,8 +1181,8 @@ class VolumeApiTest(test.TestCase):
sort_keys=None, sort_dirs=None,
filters=None,
viewable_admin_meta=False, offset=0):
self.assertFalse('no_migration_targets' in filters)
self.assertFalse('all_tenants' in filters)
self.assertNotIn('no_migration_targets', filters)
self.assertNotIn('all_tenants', filters)
return [stubs.stub_volume(fake.VOLUME3_ID, display_name='vol3')]
self.stubs.Set(db, 'volume_get_all_by_project',
stub_volume_get_all_by_project3)

2
cinder/tests/unit/backup/drivers/test_backup_ceph.py

@ -1143,7 +1143,7 @@ class VolumeMetadataBackupTestCase(test.TestCase):
def mock_read(*args):
called.append('read')
self.assertTrue(len(obj_data) == 1)
self.assertEqual(1, len(obj_data))
return obj_data[0]
def _mock_write(data):

2
cinder/tests/unit/backup/drivers/test_backup_google.py

@ -540,7 +540,7 @@ class GoogleBackupDriverTestCase(test.TestCase):
result = service._prepare_output_data(fake_data)
self.assertEqual('zlib', result[0])
self.assertTrue(len(result) < len(fake_data))
self.assertGreater(len(fake_data), len(result))
@gcs_client
def test_prepare_output_data_no_compresssion(self):

2
cinder/tests/unit/backup/drivers/test_backup_nfs.py

@ -634,7 +634,7 @@ class BackupNFSSwiftBasedTestCase(test.TestCase):
result = service._prepare_output_data(fake_data)
self.assertEqual('zlib', result[0])
self.assertTrue(len(result) < len(fake_data))
self.assertGreater(len(fake_data), len(result))
def test_prepare_output_data_no_compresssion(self):
self.flags(backup_compression_algorithm='none')

2
cinder/tests/unit/backup/drivers/test_backup_swift.py

@ -811,7 +811,7 @@ class BackupSwiftTestCase(test.TestCase):
result = service._prepare_output_data(fake_data)
self.assertEqual('zlib', result[0])
self.assertTrue(len(result) < len(fake_data))
self.assertGreater(len(fake_data), len(result))
def test_prepare_output_data_no_compresssion(self):
self.flags(backup_compression_algorithm='none')

2
cinder/tests/unit/backup/test_backup.py

@ -1056,7 +1056,7 @@ class BackupTestCase(BaseBackupTest):
export = self.backup_mgr.export_record(self.ctxt, backup)
self.assertEqual(CONF.backup_driver, export['backup_service'])
self.assertTrue('backup_url' in export)
self.assertIn('backup_url', export)
def test_import_record_with_verify_not_implemented(self):
"""Test normal backup record import.

6
cinder/tests/unit/objects/test_base.py

@ -708,9 +708,9 @@ class TestCinderDictObject(test_objects.BaseObjectsTestCase):
self.assertEqual(42, obj.get('foo'))
self.assertEqual(42, obj.get('foo', None))
self.assertTrue('foo' in obj)
self.assertTrue('abc' in obj)
self.assertFalse('def' in obj)
self.assertIn('foo', obj)
self.assertIn('abc', obj)
self.assertNotIn('def', obj)
@mock.patch('cinder.objects.base.OBJ_VERSIONS', fake_objects.MyHistory())

8
cinder/tests/unit/scheduler/test_host_filters.py

@ -1262,10 +1262,10 @@ class BasicFiltersTestCase(HostFiltersTestCase):
def test_all_filters(self):
# Double check at least a couple of known filters exist
self.assertTrue('JsonFilter' in self.class_map)
self.assertTrue('CapabilitiesFilter' in self.class_map)
self.assertTrue('AvailabilityZoneFilter' in self.class_map)
self.assertTrue('IgnoreAttemptedHostsFilter' in self.class_map)
self.assertIn('JsonFilter', self.class_map)
self.assertIn('CapabilitiesFilter', self.class_map)
self.assertIn('AvailabilityZoneFilter', self.class_map)
self.assertIn('IgnoreAttemptedHostsFilter', self.class_map)
def _do_test_type_filter_extra_specs(self, ecaps, especs, passes):
filt_cls = self.class_map['CapabilitiesFilter']()

2
cinder/tests/unit/scheduler/test_host_manager.py

@ -294,7 +294,7 @@ class HostManagerTestCase(test.TestCase):
disabled=False)
self.assertEqual(expected, _mock_service_is_up.call_args_list)
self.assertTrue(_mock_warning.call_count > 0)
self.assertGreater(_mock_warning.call_count, 0)
# Get host_state_map and make sure we have the first 2 hosts (host3 is
# down, host4 is missing capabilities)

4
cinder/tests/unit/scheduler/test_scheduler.py

@ -310,7 +310,7 @@ class SchedulerManagerTestCase(test.TestCase):
self.context,
'volume',
consistencygroup_obj)
self.assertTrue(LOG.exception.call_count > 0)
self.assertGreater(LOG.exception.call_count, 0)
db.consistencygroup_update.assert_called_once_with(
self.context, group_id, {'status': (
fields.ConsistencyGroupStatus.ERROR)})
@ -323,7 +323,7 @@ class SchedulerManagerTestCase(test.TestCase):
reason="No weighed hosts available")
self.manager.create_consistencygroup(
self.context, 'volume', consistencygroup_obj)
self.assertTrue(LOG.error.call_count > 0)
self.assertGreater(LOG.error.call_count, 0)
db.consistencygroup_update.assert_called_once_with(
self.context, group_id, {'status': (
fields.ConsistencyGroupStatus.ERROR)})

4
cinder/tests/unit/test_context.py

@ -63,8 +63,8 @@ class ContextTestCase(test.TestCase):
admin_context = user_context.elevated()
self.assertFalse(user_context.is_admin)
self.assertTrue(admin_context.is_admin)
self.assertFalse('admin' in user_context.roles)
self.assertTrue('admin' in admin_context.roles)
self.assertNotIn('admin', user_context.roles)
self.assertIn('admin', admin_context.roles)
def test_service_catalog_nova_and_swift(self):
service_catalog = [

2
cinder/tests/unit/test_db_api.py

@ -131,7 +131,7 @@ class DBAPIServiceTestCase(BaseTest):
def test_service_create(self):
service = self._create_service({})
self.assertFalse(service['id'] is None)
self.assertIsNotNone(service['id'])
for key, value in self._get_base_values().items():
self.assertEqual(value, service[key])

2
cinder/tests/unit/test_drbdmanagedrv.py

@ -456,7 +456,7 @@ class DrbdManageIscsiTestCase(test.TestCase):
self.assertEqual("list_volumes", dmd.odm.calls[0][0])
self.assertEqual("list_assignments", dmd.odm.calls[1][0])
self.assertEqual("create_snapshot", dmd.odm.calls[2][0])
self.assertTrue('node' in dmd.odm.calls[2][3])
self.assertIn('node', dmd.odm.calls[2][3])
def test_delete_snapshot(self):
testsnap = {'id': 'ca253fd0-8068-11e4-98c0-5254008ea111'}

36
cinder/tests/unit/test_emc_vmax.py

@ -1998,7 +1998,7 @@ class EMCVMAXISCSIDriverNoFastTestCase(test.TestCase):
portGroupNames = ['pg1', 'pg2', 'pg3', 'pg4']
portGroupName = (
self.driver.common.utils._get_random_pg_from_list(portGroupNames))
self.assertTrue('pg' in portGroupName)
self.assertIn('pg', portGroupName)
portGroupNames = ['pg1']
portGroupName = (
@ -2017,7 +2017,7 @@ class EMCVMAXISCSIDriverNoFastTestCase(test.TestCase):
"</EMC>")
dom = minidom.parseString(data)
portgroup = self.driver.common.utils._get_random_portgroup(dom)
self.assertTrue('OS-PG' in portgroup)
self.assertIn('OS-PG', portgroup)
# Duplicate portgroups
data = ("<?xml version='1.0' encoding='UTF-8'?>\n<EMC>\n"
@ -2030,7 +2030,7 @@ class EMCVMAXISCSIDriverNoFastTestCase(test.TestCase):
"</EMC>")
dom = minidom.parseString(data)
portgroup = self.driver.common.utils._get_random_portgroup(dom)
self.assertTrue('OS-PG' in portgroup)
self.assertIn('OS-PG', portgroup)
def test_get_random_portgroup_exception(self):
# Missing PortGroup values
@ -2789,7 +2789,7 @@ class EMCVMAXISCSIDriverNoFastTestCase(test.TestCase):
conn, self.data.initiatorNames, self.data.storage_system))
# The hardware id list has not been found as it has been removed
# externally.
self.assertTrue(len(foundHardwareIdInstanceNames2) == 0)
self.assertEqual(0, len(foundHardwareIdInstanceNames2))
# Bug 1393555 - controller has been deleted by another process.
def test_find_lunmasking_scsi_protocol_controller(self):
@ -2910,8 +2910,7 @@ class EMCVMAXISCSIDriverNoFastTestCase(test.TestCase):
'FC_SLVR1', arrayInfo[0]['PoolName'])
self.assertEqual(
'SILVER1', arrayInfo[0]['FastPolicy'])
self.assertTrue(
'OS-PORTGROUP' in arrayInfo[0]['PortGroup'])
self.assertIn('OS-PORTGROUP', arrayInfo[0]['PortGroup'])
bExists = os.path.exists(file_name)
if bExists:
os.remove(file_name)
@ -6644,12 +6643,11 @@ class EMCV2MultiPoolDriverTestCase(test.TestCase):
arrayInfo = self.driver.utils.parse_file_to_get_array_map(
self.config_file_path)
self.assertTrue(len(arrayInfo) == 2)
self.assertEqual(2, len(arrayInfo))
for arrayInfoRec in arrayInfo:
self.assertEqual(
'1234567891011', arrayInfoRec['SerialNumber'])
self.assertTrue(
self.data.port_group in arrayInfoRec['PortGroup'])
self.assertIn(self.data.port_group, arrayInfoRec['PortGroup'])
self.assertTrue(
self.data.poolname in arrayInfoRec['PoolName'] or
'SATA_BRONZE1' in arrayInfoRec['PoolName'])
@ -6932,13 +6930,12 @@ class EMCV3MultiSloDriverTestCase(test.TestCase):
arrayInfo = self.driver.utils.parse_file_to_get_array_map(
self.config_file_path)
self.assertTrue(len(arrayInfo) == 2)
self.assertEqual(2, len(arrayInfo))
for arrayInfoRec in arrayInfo:
self.assertEqual(
'1234567891011', arrayInfoRec['SerialNumber'])
self.assertTrue(
self.data.port_group in arrayInfoRec['PortGroup'])
self.assertTrue('SRP_1' in arrayInfoRec['PoolName'])
self.assertIn(self.data.port_group, arrayInfoRec['PortGroup'])
self.assertIn('SRP_1', arrayInfoRec['PoolName'])
self.assertTrue(
'Bronze' in arrayInfoRec['SLO'] or
'Silver' in arrayInfoRec['SLO'])
@ -7263,7 +7260,7 @@ class EMCV2MultiPoolDriverMultipleEcomsTestCase(test.TestCase):
pool = 'gold+1234567891011'
arrayInfo = self.driver.utils.parse_file_to_get_array_map(
self.config_file_path)
self.assertTrue(len(arrayInfo) == 4)
self.assertEqual(4, len(arrayInfo))
poolRec = self.driver.utils.extract_record(arrayInfo, pool)
self.assertEqual('1234567891011', poolRec['SerialNumber'])
@ -7279,7 +7276,7 @@ class EMCV2MultiPoolDriverMultipleEcomsTestCase(test.TestCase):
arrayInfo = self.driver.utils.parse_file_to_get_array_map(
self.config_file_path)
self.assertTrue(len(arrayInfo) == 4)
self.assertEqual(4, len(arrayInfo))
poolRec = self.driver.utils.extract_record(arrayInfo, pool)
self.assertEqual('1234567891011', poolRec['SerialNumber'])
@ -7393,8 +7390,7 @@ class EMCVMAXProvisionV3Test(test.TestCase):
conn, poolInstanceName)
storagepoolsetting = provisionv3.get_storage_pool_setting(
conn, storagePoolCapability, slo, workload)
self.assertTrue(
'Bronze:DSS' in storagepoolsetting['InstanceID'])
self.assertIn('Bronze:DSS', storagepoolsetting['InstanceID'])
def test_get_storage_pool_setting_exception(self):
provisionv3 = self.driver.common.provisionv3
@ -7685,7 +7681,7 @@ class EMCVMAXFCTest(test.TestCase):
mvInstances = self.driver._get_common_masking_views(
portGroupInstanceName, initiatorGroupInstanceName)
self.assertTrue(len(mvInstances) == 2)
self.assertEqual(2, len(mvInstances))
def test_get_common_masking_views_one_overlap(self):
common = self.driver.common
@ -7714,7 +7710,7 @@ class EMCVMAXFCTest(test.TestCase):
mvInstances = self.driver._get_common_masking_views(
portGroupInstanceName, initiatorGroupInstanceName)
self.assertTrue(len(mvInstances) == 1)
self.assertEqual(1, len(mvInstances))
def test_get_common_masking_views_no_overlap(self):
common = self.driver.common
@ -7741,7 +7737,7 @@ class EMCVMAXFCTest(test.TestCase):
mvInstances = self.driver._get_common_masking_views(
portGroupInstanceName, initiatorGroupInstanceName)
self.assertTrue(len(mvInstances) == 0)
self.assertEqual(0, len(mvInstances))
class EMCVMAXUtilsTest(test.TestCase):

162
cinder/tests/unit/test_emc_vnx.py

@ -1931,17 +1931,14 @@ class EMCVNXCLIDriverISCSITestCase(DriverTestCaseBase):
self.driverSetup(commands, results)
stats = self.driver.get_volume_stats(True)
self.assertTrue(stats['driver_version'] == VERSION,
"driver_version is incorrect")
self.assertTrue(
stats['storage_protocol'] == 'iSCSI',
"storage_protocol is incorrect")
self.assertTrue(
stats['vendor_name'] == "EMC",
"vendor name is incorrect")
self.assertTrue(
stats['volume_backend_name'] == "namedbackend",
"volume backend name is incorrect")
self.assertEqual(VERSION, stats['driver_version'],
"driver_version is incorrect")
self.assertEqual('iSCSI', stats['storage_protocol'],
"storage_protocol is incorrect")
self.assertEqual("EMC", stats['vendor_name'],
"vendor name is incorrect")
self.assertEqual("namedbackend", stats['volume_backend_name'],
"volume backend name is incorrect")
pool_stats = stats['pools'][0]
@ -2002,9 +1999,8 @@ class EMCVNXCLIDriverISCSITestCase(DriverTestCaseBase):
self.driver.cli.check_max_pool_luns_threshold = True
stats = self.driver.get_volume_stats(True)
pool_stats = stats['pools'][0]
self.assertTrue(
pool_stats['free_capacity_gb'] == 0,
"free_capacity_gb is incorrect")
self.assertEqual(0, pool_stats['free_capacity_gb'],
"free_capacity_gb is incorrect")
expect_cmd = [
mock.call(*self.testData.POOL_FEATURE_INFO_POOL_LUNS_CMD(),
poll=False)]
@ -2013,11 +2009,10 @@ class EMCVNXCLIDriverISCSITestCase(DriverTestCaseBase):
self.driver.cli.check_max_pool_luns_threshold = False
stats = self.driver.get_volume_stats(True)
pool_stats = stats['pools'][0]
self.assertTrue(stats['driver_version'] is not None,
"driver_version is not returned")
self.assertTrue(
pool_stats['free_capacity_gb'] == 3105.303,
"free_capacity_gb is incorrect")
self.assertIsNotNone(stats['driver_version'],
"driver_version is not returned")
self.assertEqual(3105.303, pool_stats['free_capacity_gb'],
"free_capacity_gb is incorrect")
@mock.patch("cinder.volume.drivers.emc.emc_vnx_cli."
"CommandLineHelper.create_lun_by_cmd",
@ -2611,8 +2606,8 @@ Time Remaining: 0 second(s)
self.testData.test_volume,
self.testData.connector
)
self.assertTrue(iscsi_data['data']['target_lun'] == 2,
"iSCSI initialize connection returned wrong HLU")
self.assertEqual(2, iscsi_data['data']['target_lun'],
"iSCSI initialize connection returned wrong HLU")
expected = [mock.call(*self.testData.STORAGEGROUP_LIST_CMD('fakehost'),
poll=False),
mock.call('storagegroup', '-addhlu', '-hlu', 2, '-alu', 3,
@ -2722,8 +2717,8 @@ Time Remaining: 0 second(s)
iscsi_data = self.driver.initialize_connection(
self.testData.test_volume,
self.testData.connector)
self.assertTrue(iscsi_data['data']['target_lun'] == 2,
"iSCSI initialize connection returned wrong HLU")
self.assertEqual(2, iscsi_data['data']['target_lun'],
"iSCSI initialize connection returned wrong HLU")
expected = [mock.call(*self.testData.STORAGEGROUP_LIST_CMD('fakehost'),
poll=False),
mock.call(*self.testData.STORAGEGROUP_LIST_CMD('fakehost'),
@ -4184,7 +4179,7 @@ Time Remaining: 0 second(s)
new_type_data,
diff_data,
host_test_data)
self.assertTrue(type(ret) == tuple)
self.assertEqual(tuple, type(ret))
self.assertTrue(ret[0])
self.assertIn('type^lun', ret[1]['provider_location'])
expect_cmd = [
@ -4693,10 +4688,10 @@ Time Remaining: 0 second(s)
snap2_in_src_cgsnap],
source_cg=None, source_vols=None))
self.assertEqual(2, len(volumes_model_update))
self.assertTrue('id^%s' % 6231 in
volumes_model_update[0]['provider_location'])
self.assertTrue('id^%s' % 6232 in
volumes_model_update[1]['provider_location'])
self.assertIn('id^%s' % 6231,
volumes_model_update[0]['provider_location'])
self.assertIn('id^%s' % 6232,
volumes_model_update[1]['provider_location'])
expect_cmd = [
mock.call(*td.SNAP_COPY_CMD(src_cgsnap['id'], copied_snap_name)),
@ -4939,10 +4934,10 @@ Time Remaining: 0 second(s)
source_cg=src_cg, source_vols=[vol1_in_src_cg,
vol2_in_src_cg]))
self.assertEqual(2, len(volumes_model_update))
self.assertTrue('id^%s' % 6231 in
volumes_model_update[0]['provider_location'])
self.assertTrue('id^%s' % 6232 in
volumes_model_update[1]['provider_location'])
self.assertIn('id^%s' % 6231,
volumes_model_update[0]['provider_location'])
self.assertIn('id^%s' % 6232,
volumes_model_update[1]['provider_location'])
delete_temp_snap_cmd = [
mock.call(*td.DELETE_CG_SNAPSHOT(temp_snap_name))]
@ -5067,8 +5062,8 @@ Time Remaining: 0 second(s)
'fakehost',
{('A', 2, 0), ('B', 1, 0)},
all_targets)
self.assertTrue(port_a2 in targets)
self.assertTrue(port_b1 in targets)
self.assertIn(port_a2, targets)
self.assertIn(port_b1, targets)
@mock.patch.object(emc_vnx_cli.CommandLineHelper,
'ping_node')
@ -5098,9 +5093,9 @@ Time Remaining: 0 second(s)
'fakehost',
{('A', 2, 0), ('A', 1, 0), ('B', 1, 0)},
all_targets)
self.assertTrue(port_a1 in targets)
self.assertTrue(port_a2 in targets)
self.assertTrue(port_b1 in targets)
self.assertIn(port_a1, targets)
self.assertIn(port_a2, targets)
self.assertIn(port_b1, targets)
@mock.patch('cinder.volume.drivers.emc.emc_vnx_cli.'
'EMCVnxCliBase.get_lun_owner',
@ -5170,17 +5165,14 @@ class EMCVNXCLIDArrayBasedDriverTestCase(DriverTestCaseBase):
self.driverSetup(commands, results)
stats = self.driver.get_volume_stats(True)
self.assertTrue(stats['driver_version'] == VERSION,
"driver_version is incorrect")
self.assertTrue(
stats['storage_protocol'] == 'iSCSI',
"storage_protocol is not correct")
self.assertTrue(
stats['vendor_name'] == "EMC",
"vendor name is not correct")
self.assertTrue(
stats['volume_backend_name'] == "namedbackend",
"volume backend name is not correct")
self.assertEqual(VERSION, stats['driver_version'],
"driver_version is incorrect")
self.assertEqual('iSCSI', stats['storage_protocol'],
"storage_protocol is not correct")
self.assertEqual("EMC", stats['vendor_name'],
"vendor name is not correct")
self.assertEqual("namedbackend", stats['volume_backend_name'],
"volume backend name is not correct")
self.assertEqual(2, len(stats['pools']))
pool_stats1 = stats['pools'][0]
@ -5281,21 +5273,16 @@ class EMCVNXCLIDArrayBasedDriverTestCase(DriverTestCaseBase):
self.driverSetup(commands, results)
stats = self.driver.get_volume_stats(True)
self.assertTrue(
stats['pools'][0]['free_capacity_gb'] == 0,
"free_capacity_gb is incorrect")
self.assertTrue(
stats['pools'][1]['free_capacity_gb'] != 0,
"free_capacity_gb is incorrect")
self.assertTrue(
stats['pools'][2]['free_capacity_gb'] != 0,
"free_capacity_gb is incorrect")
self.assertTrue(
stats['pools'][3]['free_capacity_gb'] == 0,
"free_capacity_gb is incorrect")
self.assertTrue(
stats['pools'][4]['free_capacity_gb'] == 0,
"free_capacity_gb is incorrect")
self.assertEqual(0, stats['pools'][0]['free_capacity_gb'],
"free_capacity_gb is incorrect")
self.assertNotEqual(0, stats['pools'][1]['free_capacity_gb'],
"free_capacity_gb is incorrect")
self.assertNotEqual(0, stats['pools'][2]['free_capacity_gb'],
"free_capacity_gb is incorrect")
self.assertEqual(0, stats['pools'][3]['free_capacity_gb'],
"free_capacity_gb is incorrect")
self.assertEqual(0, stats['pools'][4]['free_capacity_gb'],
"free_capacity_gb is incorrect")
@mock.patch('oslo_service.loopingcall.FixedIntervalLoopingCall',
new=utils.ZeroIntervalLoopingCall)
@ -5759,8 +5746,8 @@ class EMCVNXCLIDriverFCTestCase(DriverTestCaseBase):
connection_info = self.driver.terminate_connection(
self.testData.test_volume,
self.testData.connector)
self.assertTrue('initiator_target_map' in connection_info['data'],
'initiator_target_map should be populated.')
self.assertIn('initiator_target_map', connection_info['data'],
'initiator_target_map should be populated.')
self.assertEqual(EMCVNXCLIDriverTestData.i_t_map,
connection_info['data']['initiator_target_map'])
@ -5772,17 +5759,14 @@ class EMCVNXCLIDriverFCTestCase(DriverTestCaseBase):
self.driverSetup(commands, results)
stats = self.driver.get_volume_stats(True)
self.assertTrue(stats['driver_version'] == VERSION,
"driver_version is incorrect")
self.assertTrue(
stats['storage_protocol'] == 'FC',
"storage_protocol is incorrect")
self.assertTrue(
stats['vendor_name'] == "EMC",
"vendor name is incorrect")
self.assertTrue(
stats['volume_backend_name'] == "namedbackend",
"volume backend name is incorrect")
self.assertEqual(VERSION, stats['driver_version'],
"driver_version is incorrect")
self.assertEqual('FC', stats['storage_protocol'],
"storage_protocol is incorrect")
self.assertEqual("EMC", stats['vendor_name'],
"vendor name is incorrect")
self.assertEqual("namedbackend", stats['volume_backend_name'],
"volume backend name is incorrect")
pool_stats = stats['pools'][0]
@ -5817,9 +5801,8 @@ class EMCVNXCLIDriverFCTestCase(DriverTestCaseBase):
self.driver.cli.check_max_pool_luns_threshold = True
stats = self.driver.get_volume_stats(True)
pool_stats = stats['pools'][0]
self.assertTrue(
pool_stats['free_capacity_gb'] == 0,
"free_capacity_gb is incorrect")
self.assertEqual(0, pool_stats['free_capacity_gb'],
"free_capacity_gb is incorrect")
expect_cmd = [
mock.call(*self.testData.POOL_FEATURE_INFO_POOL_LUNS_CMD(),
poll=False)]
@ -5828,11 +5811,10 @@ class EMCVNXCLIDriverFCTestCase(DriverTestCaseBase):
self.driver.cli.check_max_pool_luns_threshold = False
stats = self.driver.get_volume_stats(True)
pool_stats = stats['pools'][0]
self.assertTrue(stats['driver_version'] is not None,
"driver_version is incorrect")
self.assertTrue(
pool_stats['free_capacity_gb'] == 3105.303,
"free_capacity_gb is incorrect")
self.assertIsNotNone(stats['driver_version'],
"driver_version is incorrect")
self.assertEqual(3105.303, pool_stats['free_capacity_gb'],
"free_capacity_gb is incorrect")
def test_deregister_initiator(self):
fake_cli = self.driverSetup()
@ -6221,9 +6203,9 @@ class EMCVNXCLIDriverReplicationV2TestCase(DriverTestCaseBase):
'create_lun_with_advance_feature',
mock.Mock(return_value={'lun_id': 5})):
model_update = self.driver.create_volume(rep_volume)
self.assertTrue(model_update['replication_status'] == 'enabled')
self.assertTrue(model_update['replication_driver_data'] ==
build_replication_data(self.configuration))
self.assertEqual('enabled', model_update['replication_status'])
self.assertEqual(build_replication_data(self.configuration),
model_update['replication_driver_data'])
self.assertDictMatch({'system': self.backend_id,
'snapcopy': 'False'},
model_update['metadata'])
@ -6253,9 +6235,9 @@ class EMCVNXCLIDriverReplicationV2TestCase(DriverTestCaseBase):
'create_lun_with_advance_feature',
mock.Mock(return_value={'lun_id': 5})):
model_update = self.driver.create_volume(rep_volume)
self.assertTrue(model_update['replication_status'] == 'enabled')
self.assertTrue(model_update['replication_driver_data'] ==
build_replication_data(self.configuration))
self.assertEqual('enabled', model_update['replication_status'])
self.assertEqual(build_replication_data(self.configuration),
model_update['replication_driver_data'])
fake_cli.assert_has_calls(
[mock.call(*self.testData.MIRROR_CREATE_CMD(mirror_name, 5),
poll=True),

8
cinder/tests/unit/test_hitachi_hnas_iscsi.py

@ -350,7 +350,7 @@ class HNASiSCSIDriverTest(test.TestCase):
self.driver.delete_volume(vol)
# should not be deletable twice
prov_loc = self.backend.getVolumebyProvider(vol['provider_location'])
self.assertTrue(prov_loc is None)
self.assertIsNone(prov_loc)
def test_extend_volume(self):
vol = self._create_volume()
@ -414,7 +414,7 @@ class HNASiSCSIDriverTest(test.TestCase):
lun = svol['provider_location']
m_id_to_vol.return_value = svol
self.driver.delete_snapshot(svol)
self.assertTrue(self.backend.getVolumebyProvider(lun) is None)
self.assertIsNone(self.backend.getVolumebyProvider(lun))
def test_create_volume_from_snapshot(self):
svol = self._create_volume()
@ -433,8 +433,8 @@ class HNASiSCSIDriverTest(test.TestCase):
connector['host'] = 'dut_1.lab.hds.com'
vol = self._create_volume()
conn = self.driver.initialize_connection(vol, connector)
self.assertTrue('3260' in conn['data']['target_portal'])
self.assertTrue(type(conn['data']['target_lun']) is int)
self.assertIn('3260', conn['data']['target_portal'])
self.assertIs(type(conn['data']['target_lun']), int)
self.backend.add_iscsi_conn = mock.MagicMock()
self.backend.add_iscsi_conn.side_effect = putils.ProcessExecutionError

4
cinder/tests/unit/test_hpelefthand.py

@ -534,7 +534,7 @@ class TestHPELeftHandISCSIDriver(HPELeftHandBaseDriver, test.TestCase):
self.assertEqual('iscsi', result['driver_volume_type'])
self.assertFalse(result['data']['target_discovered'])
self.assertEqual(self.volume_id, result['data']['volume_id'])
self.assertTrue('auth_method' not in result['data'])
self.assertNotIn('auth_method', result['data'])
expected = self.driver_startup_call_stack + [
mock.call.getServerByName('fakehost'),
@ -587,7 +587,7 @@ class TestHPELeftHandISCSIDriver(HPELeftHandBaseDriver, test.TestCase):
self.assertEqual('iscsi', result['driver_volume_type'])
self.assertFalse(result['data']['target_discovered'])
self.assertEqual(self.volume_id, result['data']['volume_id'])
self.assertTrue('auth_method' not in result['data'])
self.assertNotIn('auth_method', result['data'])
expected = self.driver_startup_call_stack + [
mock.call.getServerByName('fakehost'),

60
cinder/tests/unit/test_huawei_drivers.py

@ -3367,10 +3367,10 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
old_replica = driver.replica
secondary_id, volumes_update = driver.failover_host(
None, [test_volume], 'default')
self.assertTrue(driver.active_backend_id in ('', None))
self.assertTrue(old_client == driver.client)
self.assertTrue(old_replica_client == driver.replica_client)
self.assertTrue(old_replica == driver.replica)
self.assertIn(driver.active_backend_id, ('', None))
self.assertEqual(old_client, driver.client)
self.assertEqual(old_replica_client, driver.replica_client)
self.assertEqual(old_replica, driver.replica)
self.assertEqual('default', secondary_id)
self.assertEqual(0, len(volumes_update))
@ -3383,9 +3383,9 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
secondary_id, volumes_update = driver.failover_host(
None, [test_volume], REPLICA_BACKEND_ID)
self.assertEqual(REPLICA_BACKEND_ID, driver.active_backend_id)
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual(REPLICA_BACKEND_ID, secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']
@ -3405,9 +3405,9 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
secondary_id, volumes_update = driver.failover_host(
None, [test_volume], REPLICA_BACKEND_ID)
self.assertEqual(REPLICA_BACKEND_ID, driver.active_backend_id)
self.assertTrue(old_client == driver.client)
self.assertTrue(old_replica_client == driver.replica_client)
self.assertTrue(old_replica == driver.replica)
self.assertEqual(old_client, driver.client)
self.assertEqual(old_replica_client, driver.replica_client)
self.assertEqual(old_replica, driver.replica)
self.assertEqual(REPLICA_BACKEND_ID, secondary_id)
self.assertEqual(0, len(volumes_update))
@ -3424,17 +3424,17 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
old_replica = driver.replica
secondary_id, volumes_update = driver.failover_host(
None, [volume], 'default')
self.assertTrue(driver.active_backend_id in ('', None))
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertIn(driver.active_backend_id, ('', None))
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual('default', secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']
v_update = volumes_update[0]['updates']
self.assertEqual(volume['id'], v_id)
self.assertEqual('available', v_update['status'])
self.assertFalse('old_status' in v_update['metadata'])
self.assertNotIn('old_status', v_update['metadata'])
def test_failover_replica_volumes(self):
driver = FakeISCSIStorage(configuration=self.configuration)
@ -3449,9 +3449,9 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
secondary_id, volumes_update = driver.failover_host(
None, [replication_volume], REPLICA_BACKEND_ID)
self.assertEqual(REPLICA_BACKEND_ID, driver.active_backend_id)
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual(REPLICA_BACKEND_ID, secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']
@ -3479,10 +3479,10 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
return_value={'replication_enabled': 'true'}))
secondary_id, volumes_update = driver.failover_host(
None, [volume], REPLICA_BACKEND_ID)
self.assertTrue(driver.active_backend_id == REPLICA_BACKEND_ID)
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertEqual(driver.active_backend_id, REPLICA_BACKEND_ID)
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual(REPLICA_BACKEND_ID, secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']
@ -3508,10 +3508,10 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
old_replica = driver.replica
secondary_id, volumes_update = driver.failover_host(
None, [volume], 'default')
self.assertTrue(driver.active_backend_id in ('', None))
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertIn(driver.active_backend_id, ('', None))
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual('default', secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']
@ -3542,10 +3542,10 @@ class HuaweiISCSIDriverTestCase(test.TestCase):
old_replica = driver.replica
secondary_id, volumes_update = driver.failover_host(
None, [volume], 'default')
self.assertTrue(driver.active_backend_id in ('', None))
self.assertTrue(old_client == driver.replica_client)
self.assertTrue(old_replica_client == driver.client)
self.assertFalse(old_replica == driver.replica)
self.assertIn(driver.active_backend_id, ('', None))
self.assertEqual(old_client, driver.replica_client)
self.assertEqual(old_replica_client, driver.client)
self.assertNotEqual(old_replica, driver.replica)
self.assertEqual('default', secondary_id)
self.assertEqual(1, len(volumes_update))
v_id = volumes_update[0]['volume_id']

2
cinder/tests/unit/test_infortrend_common.py

@ -1604,7 +1604,7 @@ class InfortrendiSCSICommonTestCase(InfortrendTestCass):
rc, model_update = self.driver.migrate_volume(test_volume, fake_host)
self.assertFalse(rc)
self.assertTrue(model_update is None)
self.assertIsNone(model_update)
self.assertEqual(1, log_warning.call_count)
def test_migrate_volume_with_get_part_id_fail(self):

23
cinder/tests/unit/test_quobyte.py

@ -896,10 +896,9 @@ class QuobyteDriverTestCase(test.TestCase):
def test_set_nas_security_options_default(self):
drv = self._driver
self.assertTrue(drv.configuration.nas_secure_file_operations ==
"true")
self.assertTrue(drv.configuration.nas_secure_file_permissions ==
"true")
self.assertEqual("true", drv.configuration.nas_secure_file_operations)
self.assertEqual("true",
drv.configuration.nas_secure_file_permissions)
self.assertFalse(drv._execute_as_root)
def test_set_nas_security_options_insecure(self):
@ -909,10 +908,10 @@ class QuobyteDriverTestCase(test.TestCase):
drv.set_nas_security_options(is_new_cinder_install=True)
self.assertTrue(drv.configuration.nas_secure_file_operations ==
"false")
self.assertTrue(drv.configuration.nas_secure_file_permissions ==
"false")
self.assertEqual("false",
drv.configuration.nas_secure_file_operations)
self.assertEqual("false",
drv.configuration.nas_secure_file_permissions)
self.assertTrue(drv._execute_as_root)
def test_set_nas_security_options_explicitly_secure(self):
@ -922,8 +921,8 @@ class QuobyteDriverTestCase(test.TestCase):
drv.set_nas_security_options(is_new_cinder_install=True)
self.assertTrue(drv.configuration.nas_secure_file_operations ==
"true")
self.assertTrue(drv.configuration.nas_secure_file_permissions ==
"true")
self.assertEqual("true",
drv.configuration.nas_secure_file_operations)
self.assertEqual("true",
drv.configuration.nas_secure_file_permissions)
self.assertFalse(drv._execute_as_root)

51
cinder/tests/unit/test_utils.py

@ -616,26 +616,23 @@ class MonkeyPatchTestCase(test.TestCase):
self.assertEqual(8, ret_b)
package_a = self.example_package + 'example_a.'
self.assertTrue(
package_a + 'example_function_a'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertIn(package_a + 'example_function_a',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertTrue(
package_a + 'ExampleClassA.example_method'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertTrue(
package_a + 'ExampleClassA.example_method_add'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertIn(package_a + 'ExampleClassA.example_method',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertIn(package_a + 'ExampleClassA.example_method_add',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
package_b = self.example_package + 'example_b.'
self.assertFalse(
package_b + 'example_function_b'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertFalse(
package_b + 'ExampleClassB.example_method'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertFalse(
package_b + 'ExampleClassB.example_method_add'
in cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertNotIn(
package_b + 'example_function_b',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertNotIn(
package_b + 'ExampleClassB.example_method',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
self.assertNotIn(
package_b + 'ExampleClassB.example_method_add',
cinder.tests.unit.monkey_patch_example.CALLED_FUNCTION)
class AuditPeriodTest(test.TestCase):
@ -1183,8 +1180,8 @@ class LogTracingTestCase(test.TestCase):
self.assertEqual(2, mock_log.debug.call_count)
# Ensure the correct function name was logged
for call in mock_log.debug.call_args_list:
self.assertTrue('_trace_test_method' in str(call))
self.assertFalse('blah' in str(call))
self.assertIn('_trace_test_method', str(call))
self.assertNotIn('blah', str(call))
def test_utils_trace_method_outer_decorator(self):
mock_logging = self.mock_object(utils, 'logging')
@ -1210,8 +1207,8 @@ class LogTracingTestCase(test.TestCase):
self.assertEqual(2, mock_log.debug.call_count)
# Ensure the incorrect function name was logged
for call in mock_log.debug.call_args_list:
self.assertFalse('_trace_test_method' in str(call))
self.assertTrue('blah' in str(call))
self.assertNotIn('_trace_test_method', str(call))
self.assertIn('blah', str(call))
def test_utils_trace_method_outer_decorator_with_functools(self):
mock_log = mock.Mock()
@ -1238,8 +1235,8 @@ class LogTracingTestCase(test.TestCase):
self.assertEqual(2, mock_log.debug.call_count)
# Ensure the incorrect function name was logged
for call in mock_log.debug.call_args_list:
self.assertTrue('_trace_test_method' in str(call))
self.assertFalse('wraps' in str(call))
self.assertIn('_trace_test_method', str(call))
self.assertNotIn('wraps', str(call))
def test_utils_trace_method_with_exception(self):
self.LOG = self.mock_object(utils, 'LOG')
@ -1253,8 +1250,8 @@ class LogTracingTestCase(test.TestCase):
self.assertRaises(exception.APITimeout, _trace_test_method)
exception_log = self.LOG.debug.call_args_list[1]
self.assertTrue('exception' in str(exception_log))
self.assertTrue('test message' in str(exception_log))
self.assertIn('exception', str(exception_log))
self.assertIn('test message', str(exception_log))
def test_utils_trace_method_with_time(self):
mock_logging = self.mock_object(utils, 'logging')
@ -1275,7 +1272,7 @@ class LogTracingTestCase(test.TestCase):
self.assertEqual('OK', result)
return_log = mock_log.debug.call_args_list[1]
self.assertTrue('2900' in str(return_log))
self.assertIn('2900', str(return_log))
def test_utils_trace_wrapper_class(self):
mock_logging = self.mock_object(utils, 'logging')

4
cinder/tests/unit/test_v7000_common.py

@ -572,9 +572,9 @@ class V7000CommonTestCase(test.TestCase):
test_snap_id = "12345678-abcd-1234-cdef-0123456789ab"
expected = "12345678abcd1234cdef0123456789ab"
self.assertTrue(len(expected) == 32)
self.assertEqual(32, len(expected))
result = self.driver._compress_snapshot_id(test_snap_id)
self.assertTrue(result == expected)
self.assertEqual(expected, result)
def test_ensure_snapshot_resource_area(self):
result_dict = {'success': True, 'res': 'Successful'}

4
cinder/tests/unit/volume/drivers/emc/scaleio/test_consistencygroups.py

@ -136,8 +136,8 @@ class TestConsistencyGroups(scaleio.TestScaleIODriver):
self.volumes))
self.assertTrue(any(volume['status'] == 'error_deleting' for volume in
result_volumes_update))
self.assertTrue(result_model_update['status'] in ['error_deleting',
'error'])
self.assertIn(result_model_update['status'],
['error_deleting', 'error'])
def test_create_consistencygroup_from_cg(self):
self.set_https_response_mode(self.RESPONSE_MODE.Valid)

4
cinder/tests/unit/volume/drivers/netapp/dataontap/client/test_api.py

@ -64,7 +64,7 @@ class NetAppApiServerTests(test.TestCase):
expected_call_args = zapi_fakes.FAKE_CALL_ARGS_LIST
self.assertTrue(mock_invoke.call_args in expected_call_args)
self.assertIn(mock_invoke.call_args, expected_call_args)
@ddt.data('stor', 'STORE', '')
def test_set_server_type_value_error(self, server_type):
@ -545,7 +545,7 @@ class SSHUtilTests(test.TestCase):
mock.Mock(return_value=False))
self.sshutil._wait_on_stdout(stdout, 1)
exit_status.assert_any_call()
self.assertTrue(exit_status.call_count > 2)
self.assertGreater(exit_status.call_count, 2)
def _mock_ssh_channel_files(self, channel):
stdin = mock.Mock()

4
cinder/tests/unit/volume/drivers/netapp/dataontap/test_block_base.py

@ -595,7 +595,7 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
self.assertEqual(
fake.ISCSI_CONNECTION_PROPERTIES['data']['auth_password'],
target_info['data']['auth_password'])
self.assertTrue('auth_password' in target_info['data'])
self.assertIn('auth_password', target_info['data'])
self.assertEqual(
fake.ISCSI_CONNECTION_PROPERTIES['data']['discovery_auth_method'],
@ -604,7 +604,7 @@ class NetAppBlockStorageLibraryTestCase(test.TestCase):
fake.ISCSI_CONNECTION_PROPERTIES['data']
['discovery_auth_password'],
target_info['data']['discovery_auth_password'])
self.assertTrue('auth_password' in target_info['data'])
self.assertIn('auth_password', target_info['data'])
self.assertEqual(
fake.ISCSI_CONNECTION_PROPERTIES['data']
['discovery_auth_username'],

2
cinder/tests/unit/zonemanager/test_brcd_fc_zone_client_cli.py

@ -145,7 +145,7 @@ class TestBrcdFCZoneClientCLI(client_cli.BrcdFCZoneClientCLI, test.TestCase):
call_args = apply_zone_change_mock.call_args[0][0]
self.assertEqual(0, get_active_zs_mock.call_count)
self.assertEqual(2, apply_zone_change_mock.call_count)
self.assertTrue(zone_constant.CFG_ADD.strip() in call_args)
self.assertIn(zone_constant.CFG_ADD.strip(), call_args)
@mock.patch.object(client_cli.BrcdFCZoneClientCLI, '_ssh_execute')
def test_activate_zoneset(self, ssh_execute_mock):

8
cinder/tests/unit/zonemanager/test_brcd_fc_zone_driver.py

@ -138,7 +138,7 @@ class TestBrcdFcZoneDriver(BrcdFcZoneDriverBaseTest, test.TestCase):
GlobalVars._active_cfg = _active_cfg_before_add
get_southbound_client_mock.return_value = self.get_client("HTTPS")
self.driver.add_connection('BRCD_FAB_1', _initiator_target_map)
self.assertTrue(_zone_name in GlobalVars._zone_state)
self.assertIn(_zone_name, GlobalVars._zone_state)
@mock.patch.object(driver.BrcdFCZoneDriver, '_get_southbound_client')
def test_delete_connection(self, get_southbound_client_mock):
@ -147,7 +147,7 @@ class TestBrcdFcZoneDriver(BrcdFcZoneDriverBaseTest, test.TestCase):
GlobalVars._active_cfg = _active_cfg_before_delete
self.driver.delete_connection(
'BRCD_FAB_1', _initiator_target_map)
self.assertFalse(_zone_name in GlobalVars._zone_state)
self.assertNotIn(_zone_name, GlobalVars._zone_state)
@mock.patch.object(driver.BrcdFCZoneDriver, '_get_southbound_client')
def test_add_connection_for_initiator_mode(self, get_southbound_client_mk):
@ -157,7 +157,7 @@ class TestBrcdFcZoneDriver(BrcdFcZoneDriverBaseTest, test.TestCase):
GlobalVars._active_cfg = _active_cfg_before_add
self.setup_driver(self.setup_config(True, 2))
self.driver.add_connection('BRCD_FAB_1', _initiator_target_map)
self.assertTrue(_zone_name in GlobalVars._zone_state)
self.assertIn(_zone_name, GlobalVars._zone_state)
@mock.patch.object(driver.BrcdFCZoneDriver, '_get_southbound_client')
def test_delete_connection_for_initiator_mode(self,
@ -168,7 +168,7 @@ class TestBrcdFcZoneDriver(BrcdFcZoneDriverBaseTest, test.TestCase):
self.setup_driver(self.setup_config(True, 2))
self.driver.delete_connection(
'BRCD_FAB_1', _initiator_target_map)
self.assertFalse(_zone_name in GlobalVars._zone_state)
self.assertNotIn(_zone_name, GlobalVars._zone_state)
def test_add_connection_for_invalid_fabric(self):
"""Test abnormal flows."""

4
cinder/tests/unit/zonemanager/test_cisco_fc_zone_driver.py

@ -117,7 +117,7 @@ class TestCiscoFcZoneDriver(CiscoFcZoneDriverBaseTest, test.TestCase):
GlobalVars._active_cfg = _active_cfg_before_delete
self.driver.delete_connection(
'CISCO_FAB_1', _initiator_target_map)
self.assertFalse(_zone_name in GlobalVars._zone_state)
self.assertNotIn(_zone_name, GlobalVars._zone_state)
def test_delete_connection_for_initiator_mode(self):
GlobalVars._is_normal_test = True
@ -125,7 +125,7 @@ class TestCiscoFcZoneDriver(CiscoFcZoneDriverBaseTest, test.TestCase):
self.setup_driver(self.setup_config(True, 2))
self.driver.delete_connection(
'CISCO_FAB_1', _initiator_target_map)
self.assertFalse(_zone_name in GlobalVars._zone_state)
self.assertNotIn(_zone_name, GlobalVars._zone_state)
def test_add_connection_for_invalid_fabric(self):
"""Test abnormal flows."""