Utilizes assertIsNone and assertIsNotNone

Using assertEqual/assertNotEqual to test if an element
is or is not None is too python2.4.
Our unit testing framework supports assertIsNone and assertIsNotNone
which were created for these types of tests.

Change-Id: I2e430e87dd0222b194a84f2928ad36aac26aaba6
This commit is contained in:
Zhongyue Luo 2013-10-09 09:54:30 +08:00
parent b00e307d70
commit 7567cb5fb6
26 changed files with 148 additions and 148 deletions

View File

@ -303,7 +303,7 @@ class AdminActionsTest(test.TestCase):
volume = db.volume_get(ctx, volume['id'])
self.assertEqual(volume['status'], 'in-use')
self.assertEqual(volume['instance_uuid'], stubs.FAKE_UUID)
self.assertEqual(volume['attached_host'], None)
self.assertIsNone(volume['attached_host'])
self.assertEqual(volume['mountpoint'], mountpoint)
self.assertEqual(volume['attach_status'], 'attached')
admin_metadata = volume['volume_admin_metadata']
@ -330,9 +330,9 @@ class AdminActionsTest(test.TestCase):
volume = db.volume_get(ctx, volume['id'])
# status changed to 'available'
self.assertEqual(volume['status'], 'available')
self.assertEqual(volume['instance_uuid'], None)
self.assertEqual(volume['attached_host'], None)
self.assertEqual(volume['mountpoint'], None)
self.assertIsNone(volume['instance_uuid'])
self.assertIsNone(volume['attached_host'])
self.assertIsNone(volume['mountpoint'])
self.assertEqual(volume['attach_status'], 'detached')
admin_metadata = volume['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 1)
@ -357,7 +357,7 @@ class AdminActionsTest(test.TestCase):
# volume is attached
volume = db.volume_get(ctx, volume['id'])
self.assertEqual(volume['status'], 'in-use')
self.assertEqual(volume['instance_uuid'], None)
self.assertIsNone(volume['instance_uuid'])
self.assertEqual(volume['attached_host'], host_name)
self.assertEqual(volume['mountpoint'], mountpoint)
self.assertEqual(volume['attach_status'], 'attached')
@ -385,9 +385,9 @@ class AdminActionsTest(test.TestCase):
volume = db.volume_get(ctx, volume['id'])
# status changed to 'available'
self.assertEqual(volume['status'], 'available')
self.assertEqual(volume['instance_uuid'], None)
self.assertEqual(volume['attached_host'], None)
self.assertEqual(volume['mountpoint'], None)
self.assertIsNone(volume['instance_uuid'])
self.assertIsNone(volume['attached_host'])
self.assertIsNone(volume['mountpoint'])
self.assertEqual(volume['attach_status'], 'detached')
admin_metadata = volume['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 1)

View File

@ -25,7 +25,7 @@ class RequestTest(test.TestCase):
def test_content_type_missing(self):
request = wsgi.Request.blank('/tests/123', method='POST')
request.body = "<body />"
self.assertEqual(None, request.get_content_type())
self.assertIsNone(request.get_content_type())
def test_content_type_unsupported(self):
request = wsgi.Request.blank('/tests/123', method='POST')
@ -100,12 +100,12 @@ class RequestTest(test.TestCase):
self.stubs.SmartSet(request.accept_language,
'best_match', fake_best_match)
self.assertEqual(request.best_match_language(), None)
self.assertIsNone(request.best_match_language())
# If accept-language is not included or empty, match should be None
request.headers = {'Accept-Language': ''}
self.assertEqual(request.best_match_language(), None)
self.assertIsNone(request.best_match_language())
request.headers.pop('Accept-Language')
self.assertEqual(request.best_match_language(), None)
self.assertIsNone(request.best_match_language())
class ActionDispatcherTest(test.TestCase):
@ -352,7 +352,7 @@ class ResourceTest(test.TestCase):
request.body = 'foo'
content_type, body = resource.get_body(request)
self.assertEqual(content_type, None)
self.assertIsNone(content_type)
self.assertEqual(body, '')
def test_get_body_no_content_type(self):
@ -367,7 +367,7 @@ class ResourceTest(test.TestCase):
request.body = 'foo'
content_type, body = resource.get_body(request)
self.assertEqual(content_type, None)
self.assertIsNone(content_type)
self.assertEqual(body, '')
def test_get_body_no_content_body(self):
@ -383,7 +383,7 @@ class ResourceTest(test.TestCase):
request.body = ''
content_type, body = resource.get_body(request)
self.assertEqual(content_type, None)
self.assertIsNone(content_type)
self.assertEqual(body, '')
def test_get_body(self):
@ -594,7 +594,7 @@ class ResourceTest(test.TestCase):
extensions = [extension1, extension2]
response, post = resource.pre_process_extensions(extensions, None, {})
self.assertEqual(called, [])
self.assertEqual(response, None)
self.assertIsNone(response)
self.assertEqual(list(post), [extension2, extension1])
def test_pre_process_extensions_generator(self):
@ -621,7 +621,7 @@ class ResourceTest(test.TestCase):
response, post = resource.pre_process_extensions(extensions, None, {})
post = list(post)
self.assertEqual(called, ['pre1', 'pre2'])
self.assertEqual(response, None)
self.assertIsNone(response)
self.assertEqual(len(post), 2)
self.assertTrue(inspect.isgenerator(post[0]))
self.assertTrue(inspect.isgenerator(post[1]))
@ -678,7 +678,7 @@ class ResourceTest(test.TestCase):
response = resource.post_process_extensions([extension2, extension1],
None, None, {})
self.assertEqual(called, [2, 1])
self.assertEqual(response, None)
self.assertIsNone(response)
def test_post_process_extensions_regular_response(self):
class Controller(object):
@ -730,7 +730,7 @@ class ResourceTest(test.TestCase):
None, None, {})
self.assertEqual(called, [2, 1])
self.assertEqual(response, None)
self.assertIsNone(response)
def test_post_process_extensions_generator_response(self):
class Controller(object):

View File

@ -64,7 +64,7 @@ class SelectorTest(test.TestCase):
def test_missing_key_selector(self):
sel = xmlutil.Selector('test2', 'attrs')
self.assertEqual(sel(self.obj_for_test), None)
self.assertIsNone(sel(self.obj_for_test))
self.assertRaises(KeyError, sel, self.obj_for_test, True)
def test_constant_selector(self):
@ -163,7 +163,7 @@ class TemplateElementTest(test.TestCase):
# Create a template element with no subselector
elem = xmlutil.TemplateElement('test')
self.assertEqual(elem.subselector, None)
self.assertIsNone(elem.subselector)
def test_element_subselector_string(self):
# Create a template element with a string subselector
@ -310,7 +310,7 @@ class TemplateElementTest(test.TestCase):
elem = xmlutil.TemplateElement('test')
# Ensure that it has no text
self.assertEqual(elem.text, None)
self.assertIsNone(elem.text)
# Try setting it to a string and ensure it becomes a selector
elem.text = 'test'
@ -320,7 +320,7 @@ class TemplateElementTest(test.TestCase):
# Try resetting the text to None
elem.text = None
self.assertEqual(elem.text, None)
self.assertIsNone(elem.text)
# Now make up a selector and try setting the text to that
sel = xmlutil.Selector()
@ -329,7 +329,7 @@ class TemplateElementTest(test.TestCase):
# Finally, try deleting the text and see what happens
del elem.text
self.assertEqual(elem.text, None)
self.assertIsNone(elem.text)
def test_apply_attrs(self):
# Create a template element
@ -672,13 +672,13 @@ class SlaveTemplateBuilder(xmlutil.TemplateBuilder):
class TemplateBuilderTest(test.TestCase):
def test_master_template_builder(self):
# Make sure the template hasn't been built yet
self.assertEqual(MasterTemplateBuilder._tmpl, None)
self.assertIsNone(MasterTemplateBuilder._tmpl)
# Now, construct the template
tmpl1 = MasterTemplateBuilder()
# Make sure that there is a template cached...
self.assertNotEqual(MasterTemplateBuilder._tmpl, None)
self.assertIsNotNone(MasterTemplateBuilder._tmpl)
# Make sure it wasn't what was returned...
self.assertNotEqual(MasterTemplateBuilder._tmpl, tmpl1)
@ -697,13 +697,13 @@ class TemplateBuilderTest(test.TestCase):
def test_slave_template_builder(self):
# Make sure the template hasn't been built yet
self.assertEqual(SlaveTemplateBuilder._tmpl, None)
self.assertIsNone(SlaveTemplateBuilder._tmpl)
# Now, construct the template
tmpl1 = SlaveTemplateBuilder()
# Make sure there is a template cached...
self.assertNotEqual(SlaveTemplateBuilder._tmpl, None)
self.assertIsNotNone(SlaveTemplateBuilder._tmpl)
# Make sure it was what was returned...
self.assertEqual(SlaveTemplateBuilder._tmpl, tmpl1)

View File

@ -309,7 +309,7 @@ class LimitTest(BaseLimitTestSuite):
"""Test a limit handles 1 GET per second."""
limit = limits.Limit("GET", "*", ".*", 1, 1)
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
self.assertEqual(0, limit.next_request)
self.assertEqual(0, limit.last_request)
@ -317,7 +317,7 @@ class LimitTest(BaseLimitTestSuite):
"""Test two calls to 1 GET per second limit."""
limit = limits.Limit("GET", "*", ".*", 1, 1)
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
delay = limit("GET", "/anything")
self.assertEqual(1, delay)
@ -327,7 +327,7 @@ class LimitTest(BaseLimitTestSuite):
self.time += 4
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
self.assertEqual(4, limit.next_request)
self.assertEqual(4, limit.last_request)
@ -623,25 +623,25 @@ class WsgiLimiterTest(BaseLimitTestSuite):
def test_good_url(self):
delay = self._request("GET", "/something")
self.assertEqual(delay, None)
self.assertIsNone(delay)
def test_escaping(self):
delay = self._request("GET", "/something/jump%20up")
self.assertEqual(delay, None)
self.assertIsNone(delay)
def test_response_to_delays(self):
delay = self._request("GET", "/delayed")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed")
self.assertEqual(delay, '60.00')
def test_response_to_delays_usernames(self):
delay = self._request("GET", "/delayed", "user1")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed", "user2")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed", "user1")
self.assertEqual(delay, '60.00')

View File

@ -309,7 +309,7 @@ class LimitTest(BaseLimitTestSuite):
"""Test a limit handles 1 GET per second."""
limit = limits.Limit("GET", "*", ".*", 1, 1)
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
self.assertEqual(0, limit.next_request)
self.assertEqual(0, limit.last_request)
@ -317,7 +317,7 @@ class LimitTest(BaseLimitTestSuite):
"""Test two calls to 1 GET per second limit."""
limit = limits.Limit("GET", "*", ".*", 1, 1)
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
delay = limit("GET", "/anything")
self.assertEqual(1, delay)
@ -327,7 +327,7 @@ class LimitTest(BaseLimitTestSuite):
self.time += 4
delay = limit("GET", "/anything")
self.assertEqual(None, delay)
self.assertIsNone(delay)
self.assertEqual(4, limit.next_request)
self.assertEqual(4, limit.last_request)
@ -622,25 +622,25 @@ class WsgiLimiterTest(BaseLimitTestSuite):
def test_good_url(self):
delay = self._request("GET", "/something")
self.assertEqual(delay, None)
self.assertIsNone(delay)
def test_escaping(self):
delay = self._request("GET", "/something/jump%20up")
self.assertEqual(delay, None)
self.assertIsNone(delay)
def test_response_to_delays(self):
delay = self._request("GET", "/delayed")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed")
self.assertEqual(delay, '60.00')
def test_response_to_delays_usernames(self):
delay = self._request("GET", "/delayed", "user1")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed", "user2")
self.assertEqual(delay, None)
self.assertIsNone(delay)
delay = self._request("GET", "/delayed", "user1")
self.assertEqual(delay, '60.00')

View File

@ -157,7 +157,7 @@ class ISCSIConnectorTestCase(ConnectorTestCase):
self.stubs.Set(self.connector, '_execute', initiator_no_file)
initiator = self.connector.get_initiator()
self.assertEqual(initiator, None)
self.assertIsNone(initiator)
self.stubs.Set(self.connector, '_execute', initiator_get_text)
initiator = self.connector.get_initiator()
self.assertEqual(initiator, 'iqn.1234-56.foo.bar:01:23456789abc')

View File

@ -50,4 +50,4 @@ class FinishVolumeMigrationTestCase(test.TestCase):
self.assertEqual(src_volume['name'], expected_name)
self.assertEqual(src_volume['host'], 'dest')
self.assertEqual(src_volume['status'], 'available')
self.assertEqual(src_volume['migration_status'], None)
self.assertIsNone(src_volume['migration_status'])

View File

@ -199,7 +199,7 @@ class QualityOfServiceSpecsTableTestCase(test.TestCase):
res = db.qos_specs_associations_get(self.ctxt, specs_id)
self.assertEqual(len(res), 0)
res = db.volume_type_get(self.ctxt, type_id)
self.assertEqual(res['qos_specs_id'], None)
self.assertIsNone(res['qos_specs_id'])
def test_qos_specs_disassociate_all(self):
specs_id = self._create_qos_specs('FakeQos')

View File

@ -199,7 +199,7 @@ class TestGlanceImageService(test.TestCase):
num_images = len(self.service.detail(self.context))
image_id = self.service.create(self.context, fixture)['id']
self.assertNotEquals(None, image_id)
self.assertIsNotNone(image_id)
self.assertEqual(num_images + 1,
len(self.service.detail(self.context)))
@ -207,7 +207,7 @@ class TestGlanceImageService(test.TestCase):
fixture = self._make_fixture(name='test image')
image_id = self.service.create(self.context, fixture)['id']
self.assertNotEquals(None, image_id)
self.assertIsNotNone(image_id)
self.assertRaises(exception.ImageNotFound,
self.service.show,
self.context,

View File

@ -208,7 +208,7 @@ class HostStateTestCase(test.TestCase):
def test_update_from_volume_capability(self):
fake_host = host_manager.HostState('host1')
self.assertEqual(fake_host.free_capacity_gb, None)
self.assertIsNone(fake_host.free_capacity_gb)
volume_capability = {'total_capacity_gb': 1024,
'free_capacity_gb': 512,
@ -220,7 +220,7 @@ class HostStateTestCase(test.TestCase):
def test_update_from_volume_infinite_capability(self):
fake_host = host_manager.HostState('host1')
self.assertEqual(fake_host.free_capacity_gb, None)
self.assertIsNone(fake_host.free_capacity_gb)
volume_capability = {'total_capacity_gb': 'infinite',
'free_capacity_gb': 'infinite',
@ -233,7 +233,7 @@ class HostStateTestCase(test.TestCase):
def test_update_from_volume_unknown_capability(self):
fake_host = host_manager.HostState('host1')
self.assertEqual(fake_host.free_capacity_gb, None)
self.assertIsNone(fake_host.free_capacity_gb)
volume_capability = {'total_capacity_gb': 'infinite',
'free_capacity_gb': 'unknown',

View File

@ -189,7 +189,7 @@ class BackupSwiftTestCase(test.TestCase):
def test_get_compressor(self):
service = SwiftBackupDriver(self.ctxt)
compressor = service._get_compressor('None')
self.assertEqual(compressor, None)
self.assertIsNone(compressor)
compressor = service._get_compressor('zlib')
self.assertEqual(compressor, zlib)
compressor = service._get_compressor('bz2')

View File

@ -282,7 +282,7 @@ class DBAPIVolumeTestCase(BaseTest):
self.assertEqual(volume['mountpoint'], '/tmp')
self.assertEqual(volume['attach_status'], 'attached')
self.assertEqual(volume['instance_uuid'], instance_uuid)
self.assertEqual(volume['attached_host'], None)
self.assertIsNone(volume['attached_host'])
def test_volume_attached_to_host(self):
volume = db.volume_create(self.ctxt, {'host': 'host1'})
@ -293,7 +293,7 @@ class DBAPIVolumeTestCase(BaseTest):
self.assertEqual(volume['status'], 'in-use')
self.assertEqual(volume['mountpoint'], '/tmp')
self.assertEqual(volume['attach_status'], 'attached')
self.assertEqual(volume['instance_uuid'], None)
self.assertIsNone(volume['instance_uuid'])
self.assertEqual(volume['attached_host'], host_name)
def test_volume_data_get_for_host(self):
@ -621,7 +621,7 @@ class DBAPIEncryptionTestCase(BaseTest):
encryption_get = \
db.volume_type_encryption_get(self.ctxt,
encryption['volume_type_id'])
self.assertEqual(None, encryption_get)
self.assertIsNone(encryption_get)
class DBAPIReservationTestCase(BaseTest):

View File

@ -193,10 +193,10 @@ class HUSiSCSIDriverTest(test.TestCase):
def test_create_volume(self):
loc = self.driver.create_volume(_VOLUME)
self.assertNotEqual(loc, None)
self.assertIsNotNone(loc)
vol = _VOLUME.copy()
vol['provider_location'] = loc['provider_location']
self.assertNotEqual(loc['provider_location'], None)
self.assertIsNotNone(loc['provider_location'])
return vol
def test_delete_volume(self):
@ -231,7 +231,7 @@ class HUSiSCSIDriverTest(test.TestCase):
svol = vol.copy()
svol['volume_size'] = svol['size']
loc = self.driver.create_snapshot(svol)
self.assertNotEqual(loc, None)
self.assertIsNotNone(loc)
svol['provider_location'] = loc['provider_location']
return svol
@ -243,7 +243,7 @@ class HUSiSCSIDriverTest(test.TestCase):
svol = vol.copy()
svol['volume_size'] = svol['size']
loc = self.driver.create_snapshot(svol)
self.assertNotEqual(loc, None)
self.assertIsNotNone(loc)
svol['provider_location'] = loc['provider_location']
return svol
@ -266,7 +266,7 @@ class HUSiSCSIDriverTest(test.TestCase):
def test_create_volume_from_snapshot(self):
svol = self.test_create_snapshot()
vol = self.driver.create_volume_from_snapshot(_VOLUME, svol)
self.assertNotEqual(vol, None)
self.assertIsNotNone(vol)
return vol
def test_initialize_connection(self):

View File

@ -516,7 +516,7 @@ class HVSRESTiSCSIDriverTestCase(test.TestCase):
def test_log_in_success(self):
deviceid = self.driver.common.login()
self.assertNotEqual(deviceid, None)
self.assertIsNotNone(deviceid)
def test_log_out_success(self):
self.driver.common.login()
@ -535,12 +535,12 @@ class HVSRESTiSCSIDriverTestCase(test.TestCase):
def test_delete_volume_success(self):
self.driver.common.login()
self.driver.delete_volume(test_volume)
self.assertEqual(self.driver.common.lun_id, None)
self.assertIsNone(self.driver.common.lun_id)
def test_delete_snapshot_success(self):
self.driver.common.login()
self.driver.delete_snapshot(test_snap)
self.assertEqual(self.driver.common.snapshot_id, None)
self.assertIsNone(self.driver.common.snapshot_id)
def test_colone_volume_success(self):
self.driver.common.login()
@ -577,7 +577,7 @@ class HVSRESTiSCSIDriverTestCase(test.TestCase):
def test_get_volume_stats(self):
self.driver.common.login()
status = self.driver.get_volume_stats()
self.assertNotEqual(status['free_capacity_gb'], None)
self.assertIsNotNone(status['free_capacity_gb'])
def test_create_snapshot_fail(self):
self.driver.common.login()
@ -705,7 +705,7 @@ class HVSRESTFCDriverTestCase(test.TestCase):
def test_log_in_Success(self):
deviceid = self.driver.common.login()
self.assertNotEqual(deviceid, None)
self.assertIsNotNone(deviceid)
def test_create_volume_success(self):
self.driver.common.login()
@ -720,12 +720,12 @@ class HVSRESTFCDriverTestCase(test.TestCase):
def test_delete_volume_success(self):
self.driver.common.login()
self.driver.delete_volume(test_volume)
self.assertEqual(self.driver.common.lun_id, None)
self.assertIsNone(self.driver.common.lun_id)
def test_delete_snapshot_success(self):
self.driver.common.login()
self.driver.delete_snapshot(test_snap)
self.assertEqual(self.driver.common.snapshot_id, None)
self.assertIsNone(self.driver.common.snapshot_id)
def test_colone_volume_success(self):
self.driver.common.login()
@ -762,7 +762,7 @@ class HVSRESTFCDriverTestCase(test.TestCase):
def test_get_volume_stats(self):
self.driver.common.login()
status = self.driver.get_volume_stats()
self.assertNotEqual(status['free_capacity_gb'], None)
self.assertIsNotNone(status['free_capacity_gb'])
def test_create_snapshot_fail(self):
self.driver.common.login()

View File

@ -1132,8 +1132,8 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.driver.delete_volume, FAKE_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertEqual(FAKE_VOLUME['provider_location'], None)
self.assertIsNone(LUN_INFO['ID'])
self.assertIsNone(FAKE_VOLUME['provider_location'])
def test_create_delete_cloned_volume(self):
# Test no source volume
@ -1150,14 +1150,14 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
# Test start luncopy failed
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
set_error_flg('chgluncopystatus')
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
# Test luncopy status abnormal
LUNCOPY_SETTING['Status'] = 'Disable'
@ -1165,7 +1165,7 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_cloned_volume,
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
LUNCOPY_SETTING['Status'] = 'Normal'
# Test delete luncopy failed
@ -1175,7 +1175,7 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
FAKE_CLONED_VOLUME, FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
# need to clean up LUNCopy
LUNCOPY_INFO['Name'] = None
LUNCOPY_INFO['ID'] = None
@ -1189,10 +1189,10 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertEqual(CLONED_LUN_INFO['ID'], VOLUME_SNAP_ID['vol_copy'])
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(FAKE_CLONED_VOLUME['provider_location'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertIsNone(FAKE_CLONED_VOLUME['provider_location'])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def test_create_delete_snapshot(self):
# Test no resource pool
@ -1215,8 +1215,8 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertRaises(exception.VolumeBackendAPIException,
self.driver.create_snapshot,
FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['Status'], None)
self.assertIsNone(SNAPSHOT_INFO['ID'])
self.assertIsNone(SNAPSHOT_INFO['Status'])
# Test disable snapshot failed
set_error_flg('disablesnapshot')
self.driver.create_snapshot(FAKE_SNAPSHOT)
@ -1241,8 +1241,8 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertEqual(SNAPSHOT_INFO['Status'], 'Active')
self.assertEqual(ret['provider_location'], SNAPSHOT_INFO['ID'])
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['Status'], None)
self.assertIsNone(SNAPSHOT_INFO['ID'])
self.assertIsNone(SNAPSHOT_INFO['Status'])
def test_create_delete_snapshot_volume(self):
# Test no source snapshot
@ -1261,9 +1261,9 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.driver.delete_volume(FAKE_VOLUME)
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertIsNone(SNAPSHOT_INFO['ID'])
def test_initialize_connection(self):
# Test can not get iscsi iqn
@ -1326,7 +1326,7 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertEqual(LUN_INFO['Owner Controller'], 'B')
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def test_terminate_connection(self):
# Test no host was found
@ -1365,9 +1365,9 @@ class HuaweiTISCSIDriverTestCase(test.TestCase):
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.driver.initialize_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.assertEqual(MAP_INFO["DEV LUN ID"], None)
self.assertIsNone(MAP_INFO["DEV LUN ID"])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def test_get_volume_stats(self):
stats = self.driver.get_volume_stats(True)
@ -1418,16 +1418,16 @@ class HuaweiTFCDriverTestCase(test.TestCase):
self.driver.create_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], VOLUME_SNAP_ID['vol'])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def test_create_delete_snapshot(self):
self.driver.create_volume(FAKE_VOLUME)
self.driver.create_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], VOLUME_SNAP_ID['snap'])
self.driver.delete_snapshot(FAKE_SNAPSHOT)
self.assertEqual(SNAPSHOT_INFO['ID'], None)
self.assertIsNone(SNAPSHOT_INFO['ID'])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def test_create_cloned_volume(self):
self.driver.create_volume(FAKE_VOLUME)
@ -1436,8 +1436,8 @@ class HuaweiTFCDriverTestCase(test.TestCase):
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertIsNone(LUN_INFO['ID'])
def test_create_snapshot_volume(self):
self.driver.create_volume(FAKE_VOLUME)
@ -1448,8 +1448,8 @@ class HuaweiTFCDriverTestCase(test.TestCase):
self.assertEqual(ret['provider_location'], CLONED_LUN_INFO['ID'])
self.driver.delete_volume(FAKE_CLONED_VOLUME)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(CLONED_LUN_INFO['ID'], None)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(CLONED_LUN_INFO['ID'])
self.assertIsNone(LUN_INFO['ID'])
def test_initialize_terminitat_connection(self):
self.driver.create_volume(FAKE_VOLUME)
@ -1460,10 +1460,10 @@ class HuaweiTFCDriverTestCase(test.TestCase):
self.assertEqual(MAP_INFO["DEV LUN ID"], LUN_INFO['ID'])
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.assertEqual(MAP_INFO["DEV LUN ID"], None)
self.assertEqual(MAP_INFO["Host LUN ID"], None)
self.assertIsNone(MAP_INFO["DEV LUN ID"])
self.assertIsNone(MAP_INFO["Host LUN ID"])
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
def _test_get_volume_stats(self):
stats = self.driver.get_volume_stats(True)
@ -1597,7 +1597,7 @@ class HuaweiDorado2100G2ISCSIDriverTestCase(HuaweiTISCSIDriverTestCase):
FAKE_CONNECTOR['initiator'])
self.driver.terminate_connection(FAKE_VOLUME, FAKE_CONNECTOR)
self.driver.delete_volume(FAKE_VOLUME)
self.assertEqual(LUN_INFO['ID'], None)
self.assertIsNone(LUN_INFO['ID'])
class SSHMethodTestCase(test.TestCase):

View File

@ -609,7 +609,7 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
mox.VerifyAll()
self.assertFalse(result)
self.assertFalse(vol_dict['bootable'])
self.assertEqual(vol_dict['provider_location'], None)
self.assertIsNone(vol_dict['provider_location'])
def test_clone_image_resizefails(self):
drv = self._driver
@ -656,7 +656,7 @@ class NetappDirectCmodeNfsDriverTestCase(test.TestCase):
mox.VerifyAll()
self.assertFalse(result)
self.assertFalse(vol_dict['bootable'])
self.assertEqual(vol_dict['provider_location'], None)
self.assertIsNone(vol_dict['provider_location'])
def test_is_cloneable_share_badformats(self):
drv = self._driver

View File

@ -114,7 +114,7 @@ class PolicyTestCase(test.TestCase):
action = "example:get_http"
target = {}
result = policy.enforce(self.context, action, target)
self.assertEqual(result, None)
self.assertIsNone(result)
def test_enforce_http_false(self):

View File

@ -166,7 +166,7 @@ class QuotaIntegrationTestCase(test.TestCase):
# Make sure no reservation was created for snapshot gigabytes.
reservations = db.reservation_get_all_by_project(self.context,
self.project_id)
self.assertEqual(reservations.get('gigabytes'), None)
self.assertIsNone(reservations.get('gigabytes'))
# Make sure the snapshot volume_size isn't included in usage.
vol_ref2 = volume.API().create(self.context, 10, '', '')
@ -277,7 +277,7 @@ class BaseResourceTestCase(test.TestCase):
resource = quota.BaseResource('test_resource')
self.assertEqual(resource.name, 'test_resource')
self.assertEqual(resource.flag, None)
self.assertIsNone(resource.flag)
self.assertEqual(resource.default, -1)
def test_with_flag(self):
@ -371,7 +371,7 @@ class VolumeTypeResourceTestCase(test.TestCase):
resource = quota.VolumeTypeResource('volumes', volume)
self.assertEqual(resource.name, 'volumes_%s' % volume_type_name)
self.assertEqual(resource.flag, None)
self.assertIsNone(resource.flag)
self.assertEqual(resource.default, -1)

View File

@ -74,7 +74,7 @@ class FakeImageService:
class TestUtil(test.TestCase):
def test_ascii_str(self):
self.assertEqual(None, driver.ascii_str(None))
self.assertIsNone(driver.ascii_str(None))
self.assertEqual('foo', driver.ascii_str('foo'))
self.assertEqual('foo', driver.ascii_str(u'foo'))
self.assertRaises(UnicodeEncodeError,

View File

@ -158,7 +158,7 @@ class SolidFireVolumeTestCase(test.TestCase):
sfv = SolidFireDriver(configuration=self.configuration)
model_update = sfv.create_volume(testvol)
self.assertNotEqual(model_update, None)
self.assertIsNotNone(model_update)
def test_create_volume(self):
self.stubs.Set(SolidFireDriver, '_issue_api_request',
@ -172,8 +172,8 @@ class SolidFireVolumeTestCase(test.TestCase):
sfv = SolidFireDriver(configuration=self.configuration)
model_update = sfv.create_volume(testvol)
self.assertNotEqual(model_update, None)
self.assertEqual(model_update.get('provider_geometry', None), None)
self.assertIsNotNone(model_update)
self.assertIsNone(model_update.get('provider_geometry', None))
def test_create_volume_non_512(self):
self.stubs.Set(SolidFireDriver, '_issue_api_request',
@ -275,7 +275,7 @@ class SolidFireVolumeTestCase(test.TestCase):
sfv = SolidFireDriver(configuration=self.configuration)
model_update = sfv.create_volume(testvol)
self.assertNotEqual(model_update, None)
self.assertIsNotNone(model_update)
def test_create_volume_fails(self):
# NOTE(JDG) This test just fakes update_cluster_status
@ -301,28 +301,28 @@ class SolidFireVolumeTestCase(test.TestCase):
self.stubs.Set(SolidFireDriver, '_issue_api_request',
self.fake_issue_api_request)
account = sfv._create_sfaccount('project-id')
self.assertNotEqual(account, None)
self.assertIsNotNone(account)
def test_create_sfaccount_fails(self):
sfv = SolidFireDriver(configuration=self.configuration)
self.stubs.Set(SolidFireDriver, '_issue_api_request',
self.fake_issue_api_request_fails)
account = sfv._create_sfaccount('project-id')
self.assertEqual(account, None)
self.assertIsNone(account)
def test_get_sfaccount_by_name(self):
sfv = SolidFireDriver(configuration=self.configuration)
self.stubs.Set(SolidFireDriver, '_issue_api_request',
self.fake_issue_api_request)
account = sfv._get_sfaccount_by_name('some-name')
self.assertNotEqual(account, None)
self.assertIsNotNone(account)
def test_get_sfaccount_by_name_fails(self):
sfv = SolidFireDriver(configuration=self.configuration)
self.stubs.Set(SolidFireDriver, '_issue_api_request',
self.fake_issue_api_request_fails)
account = sfv._get_sfaccount_by_name('some-name')
self.assertEqual(account, None)
self.assertIsNone(account)
def test_delete_volume(self):
self.stubs.Set(SolidFireDriver, '_issue_api_request',

View File

@ -1845,16 +1845,16 @@ class StorwizeSVCDriverTestCase(test.TestCase):
'host': u'unicode.foo}.bar}.baz-%s' % rand_id}
self.driver.initialize_connection(volume1, conn)
host_name = self.driver._get_host_from_connector(conn)
self.assertNotEqual(host_name, None)
self.assertIsNotNone(host_name)
self.driver.terminate_connection(volume1, conn)
host_name = self.driver._get_host_from_connector(conn)
self.assertEqual(host_name, None)
self.assertIsNone(host_name)
self.driver.delete_volume(volume1)
# Clean up temporary hosts
for tmpconn in [tmpconn1, tmpconn2]:
host_name = self.driver._get_host_from_connector(tmpconn)
self.assertNotEqual(host_name, None)
self.assertIsNotNone(host_name)
self.driver._delete_host(host_name)
def test_storwize_svc_validate_connector(self):
@ -1908,7 +1908,7 @@ class StorwizeSVCDriverTestCase(test.TestCase):
# Check case where no hosts exist
if self.USESIM:
ret = self.driver._get_host_from_connector(self._connector)
self.assertEqual(ret, None)
self.assertIsNone(ret)
# Make sure that the volumes have been created
self._assert_vol_exists(volume1['name'], True)
@ -1936,7 +1936,7 @@ class StorwizeSVCDriverTestCase(test.TestCase):
self.driver.terminate_connection(volume1, self._connector)
if self.USESIM:
ret = self.driver._get_host_from_connector(self._connector)
self.assertEqual(ret, None)
self.assertIsNone(ret)
# Check cases with no auth set for host
if self.USESIM:
@ -1963,10 +1963,10 @@ class StorwizeSVCDriverTestCase(test.TestCase):
chap_ret = self.driver._get_chap_secret_for_host(host_name)
if auth_enabled or host_exists == 'yes-auth':
self.assertIn('auth_password', init_ret['data'])
self.assertNotEqual(chap_ret, None)
self.assertIsNotNone(chap_ret)
else:
self.assertNotIn('auth_password', init_ret['data'])
self.assertEqual(chap_ret, None)
self.assertIsNone(chap_ret)
self.driver.terminate_connection(volume1, conn_na)
self._set_flag('storwize_svc_iscsi_chap_enabled', True)
@ -2005,7 +2005,7 @@ class StorwizeSVCDriverTestCase(test.TestCase):
# Make sure our host still exists
host_name = self.driver._get_host_from_connector(self._connector)
self.assertNotEqual(host_name, None)
self.assertIsNotNone(host_name)
# Remove the mapping from the 2nd volume and delete it. The host should
# be automatically removed because there are no more mappings.
@ -2020,7 +2020,7 @@ class StorwizeSVCDriverTestCase(test.TestCase):
# Check if our host still exists (it should not)
if self.USESIM:
ret = self.driver._get_host_from_connector(self._connector)
self.assertEqual(ret, None)
self.assertIsNone(ret)
def test_storwize_svc_multi_host_maps(self):
# We can't test connecting to multiple hosts from a single host when

View File

@ -455,7 +455,7 @@ class GenericUtilsTestCase(test.TestCase):
def test_check_ssh_injection(self):
cmd_list = ['ssh', '-D', 'my_name@name_of_remote_computer']
self.assertEqual(utils.check_ssh_injection(cmd_list), None)
self.assertIsNone(utils.check_ssh_injection(cmd_list))
def test_check_ssh_injection_on_error(self):
with_space = ['shh', 'my_name@ name_of_remote_computer']

View File

@ -869,7 +869,7 @@ class VMwareEsxVmdkDriverTestCase(test.TestCase):
"""Test _get_snapshot_from_tree."""
volops = volumeops.VMwareVolumeOps
ret = volops._get_snapshot_from_tree(mox.IgnoreArg(), None)
self.assertEqual(ret, None)
self.assertIsNone(ret)
name = 'snapshot_name'
snapshot = FakeMor('VirtualMachineSnapshot', 'my_snap')
root = FakeSnapshotTree(name='snapshot_name', snapshot=snapshot)

View File

@ -284,8 +284,8 @@ class VolumeTestCase(BaseVolumeTestCase):
1,
'name',
'description')
self.assertEqual(volume['volume_type_id'], None)
self.assertEqual(volume['encryption_key_id'], None)
self.assertIsNone(volume['volume_type_id'])
self.assertIsNone(volume['encryption_key_id'])
# Create default volume type
vol_type = conf_fixture.def_vol_type
@ -635,7 +635,7 @@ class VolumeTestCase(BaseVolumeTestCase):
self.assertEqual(vol['attach_status'], "attached")
self.assertEqual(vol['mountpoint'], mountpoint)
self.assertEqual(vol['instance_uuid'], instance_uuid)
self.assertEqual(vol['attached_host'], None)
self.assertIsNone(vol['attached_host'])
admin_metadata = vol['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 2)
self.assertEqual(admin_metadata[0]['key'], 'readonly')
@ -676,7 +676,7 @@ class VolumeTestCase(BaseVolumeTestCase):
self.assertEqual(vol['status'], "in-use")
self.assertEqual(vol['attach_status'], "attached")
self.assertEqual(vol['mountpoint'], mountpoint)
self.assertEqual(vol['instance_uuid'], None)
self.assertIsNone(vol['instance_uuid'])
# sanitized, conforms to RFC-952 and RFC-1123 specs.
self.assertEqual(vol['attached_host'], 'fake-host')
admin_metadata = vol['volume_admin_metadata']
@ -723,7 +723,7 @@ class VolumeTestCase(BaseVolumeTestCase):
self.assertEqual(vol['attach_status'], "attached")
self.assertEqual(vol['mountpoint'], mountpoint)
self.assertEqual(vol['instance_uuid'], instance_uuid)
self.assertEqual(vol['attached_host'], None)
self.assertIsNone(vol['attached_host'])
admin_metadata = vol['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 2)
self.assertEqual(admin_metadata[0]['key'], 'readonly')
@ -739,9 +739,9 @@ class VolumeTestCase(BaseVolumeTestCase):
vol = db.volume_get(self.context, volume_id)
self.assertEqual(vol['status'], "available")
self.assertEqual(vol['attach_status'], "detached")
self.assertEqual(vol['mountpoint'], None)
self.assertEqual(vol['instance_uuid'], None)
self.assertEqual(vol['attached_host'], None)
self.assertIsNone(vol['mountpoint'])
self.assertIsNone(vol['instance_uuid'])
self.assertIsNone(vol['attached_host'])
admin_metadata = vol['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 1)
self.assertEqual(admin_metadata[0]['key'], 'readonly')
@ -753,7 +753,7 @@ class VolumeTestCase(BaseVolumeTestCase):
self.assertEqual(vol['status'], "in-use")
self.assertEqual(vol['attach_status'], "attached")
self.assertEqual(vol['mountpoint'], mountpoint)
self.assertEqual(vol['instance_uuid'], None)
self.assertIsNone(vol['instance_uuid'])
self.assertEqual(vol['attached_host'], 'fake-host')
admin_metadata = vol['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 2)
@ -770,9 +770,9 @@ class VolumeTestCase(BaseVolumeTestCase):
vol = db.volume_get(self.context, volume_id)
self.assertEqual(vol['status'], "available")
self.assertEqual(vol['attach_status'], "detached")
self.assertEqual(vol['mountpoint'], None)
self.assertEqual(vol['instance_uuid'], None)
self.assertEqual(vol['attached_host'], None)
self.assertIsNone(vol['mountpoint'])
self.assertIsNone(vol['instance_uuid'])
self.assertIsNone(vol['attached_host'])
admin_metadata = vol['volume_admin_metadata']
self.assertEqual(len(admin_metadata), 1)
self.assertEqual(admin_metadata[0]['key'], 'readonly')
@ -1427,7 +1427,7 @@ class VolumeTestCase(BaseVolumeTestCase):
# create raw snapshot
volume = tests_utils.create_volume(self.context, **self.volume_params)
snapshot = self._create_snapshot(volume['id'])
self.assertEqual(snapshot['display_name'], None)
self.assertIsNone(snapshot['display_name'])
# use volume.api to update name
volume_api = cinder.volume.api.API()
update_dict = {'display_name': 'test update name'}
@ -1704,7 +1704,7 @@ class VolumeTestCase(BaseVolumeTestCase):
# check volume properties
volume = db.volume_get(context.get_admin_context(), volume['id'])
self.assertEqual(volume['host'], 'newhost')
self.assertEqual(volume['migration_status'], None)
self.assertIsNone(volume['migration_status'])
def test_migrate_volume_generic(self):
def fake_migr(vol, host):
@ -1733,7 +1733,7 @@ class VolumeTestCase(BaseVolumeTestCase):
host_obj, True)
volume = db.volume_get(context.get_admin_context(), volume['id'])
self.assertEqual(volume['host'], 'newhost')
self.assertEqual(volume['migration_status'], None)
self.assertIsNone(volume['migration_status'])
def test_update_volume_readonly_flag(self):
"""Test volume readonly flag can be updated at API level."""
@ -2098,7 +2098,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, False)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
def test_lvm_migrate_volume_bad_loc_info(self):
capabilities = {'location_info': 'foo'}
@ -2107,7 +2107,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, False)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
def test_lvm_migrate_volume_diff_driver(self):
capabilities = {'location_info': 'FooDriver:foo:bar'}
@ -2116,7 +2116,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, False)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
def test_lvm_migrate_volume_diff_host(self):
capabilities = {'location_info': 'LVMVolumeDriver:foo:bar'}
@ -2125,7 +2125,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, False)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
def test_lvm_migrate_volume_in_use(self):
hostname = socket.gethostname()
@ -2135,7 +2135,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, False)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
def test_lvm_migrate_volume_proceed(self):
hostname = socket.gethostname()
@ -2161,7 +2161,7 @@ class LVMISCSIVolumeDriverTestCase(DriverTestCase):
moved, model_update = self.volume.driver.migrate_volume(self.context,
vol, host)
self.assertEqual(moved, True)
self.assertEqual(model_update, None)
self.assertIsNone(model_update)
class LVMVolumeDriverTestCase(DriverTestCase):

View File

@ -68,4 +68,4 @@ class VolumeConfigurationTest(test.TestCase):
def test_safe_get(self):
c = configuration.Configuration(volume_opts, config_group='foo')
self.assertEqual(c.safe_get('none_opt'), None)
self.assertIsNone(c.safe_get('none_opt'))

View File

@ -225,7 +225,7 @@ class VolumeTypeTestCase(test.TestCase):
type_ref = volume_types.create(self.ctxt, "type1", {"key2": "val2",
"key3": "val3"})
res = volume_types.get_volume_type_qos_specs(type_ref['id'])
self.assertEqual(res['qos_specs'], None)
self.assertIsNone(res['qos_specs'])
qos_specs.associate_qos_with_type(self.ctxt,
qos_ref['id'],
type_ref['id'])