self.oldHTTPConnection = (
wire_HTTPConnection_to_WSGI("169.254.0.1:80", self.app))
self.proxy = limits.WsgiLimiterProxy("169.254.0.1:80")
+ self.addCleanup(self._restore, self.oldHTTPConnection)
+
+ def _restore(self, oldHTTPConnection):
+ # restore original HTTPConnection object
+ httplib.HTTPConnection = oldHTTPConnection
def test_200(self):
"""Successful request test."""
self.assertEqual((delay, error), expected)
- def tearDown(self):
- # restore original HTTPConnection object
- httplib.HTTPConnection = self.oldHTTPConnection
- super(WsgiLimiterProxyTest, self).tearDown()
-
class LimitsViewBuilderTest(test.TestCase):
def setUp(self):
class VolumeApiTest(test.TestCase):
def setUp(self):
super(VolumeApiTest, self).setUp()
+ self.addCleanup(fake_notifier.reset)
self.ext_mgr = extensions.ExtensionManager()
self.ext_mgr.extensions = {}
fake_image.stub_out_image_service(self.stubs)
self.flags(host='fake',
notification_driver=[fake_notifier.__name__])
-
self.stubs.Set(db, 'volume_get_all', stubs.stub_volume_get_all)
self.stubs.Set(db, 'service_get_all_by_topic',
stubs.stub_service_get_all_by_topic)
self.stubs.Set(volume_api.API, 'delete', stubs.stub_volume_delete)
- def tearDown(self):
- super(VolumeApiTest, self).tearDown()
- fake_notifier.reset()
-
def test_volume_create(self):
self.stubs.Set(volume_api.API, 'get', stubs.stub_volume_get)
self.stubs.Set(volume_api.API, "create", stubs.stub_volume_create)
class FinishVolumeMigrationTestCase(test.TestCase):
"""Test cases for finish_volume_migration."""
- def setUp(self):
- super(FinishVolumeMigrationTestCase, self).setUp()
-
- def tearDown(self):
- super(FinishVolumeMigrationTestCase, self).tearDown()
-
def test_finish_volume_migration(self):
ctxt = context.RequestContext(user_id='user_id',
project_id='project_id',
self.ctxt = context.RequestContext(user_id='user_id',
project_id='project_id')
- def tearDown(self):
- super(NameIDsTestCase, self).tearDown()
-
def test_name_id_same(self):
"""New volume should have same 'id' and 'name_id'."""
vol_ref = testutils.create_volume(self.ctxt, size=1)
project_id='project_id',
is_admin=True)
- def tearDown(self):
- super(QualityOfServiceSpecsTableTestCase, self).tearDown()
-
def _create_qos_specs(self, name, values=None):
"""Create a transfer object."""
if values:
self.ctxt = context.RequestContext(user_id='user_id',
project_id='project_id')
- def tearDown(self):
- super(TransfersTableTestCase, self).tearDown()
-
def _create_transfer(self, volume_id=None):
"""Create a transfer object."""
transfer = {'display_name': 'display_name',
self.ctxt = context.get_admin_context()
self.backup_mgr.driver.set_initialized()
- def tearDown(self):
- super(BackupTestCase, self).tearDown()
-
def _create_backup_db_entry(self, volume_id=1, display_name='test_backup',
display_description='this is a test backup',
container='volumebackups',
self.backup_id = str(uuid.uuid4())
self.mb = ceph.VolumeMetadataBackup(mock.Mock(), self.backup_id)
- def tearDown(self):
- super(VolumeMetadataBackupTestCase, self).tearDown()
-
@common_meta_backup_mocks
def test_name(self):
self.assertEqual(self.mb.name, 'backup.%s.meta' % (self.backup_id))
self.assertRaises(NotImplementedError,
self.driver.verify, self.backup)
- def tearDown(self):
- super(BackupBaseDriverTestCase, self).tearDown()
-
class BackupMetadataAPITestCase(test.TestCase):
mock_dumps.side_effect = TypeError
self.assertFalse(self.bak_meta_api._is_serializable(data))
mock_dumps.assert_called_once()
-
- def tearDown(self):
- super(BackupMetadataAPITestCase, self).tearDown()
self.stubs.Set(utils, 'execute', fake_exec)
self.stubs.Set(os, 'stat', fake_stat_image)
- def tearDown(self):
- super(BackupTSMTestCase, self).tearDown()
-
def _create_volume_db_entry(self, volume_id):
vol = {'id': volume_id,
'size': 1,
self.manager = importutils.import_object(CONF.volume_manager)
self.context = context.get_admin_context()
- def tearDown(self):
- super(VolumeDriverCompatibility, self).tearDown()
-
def _load_driver(self, driver):
if 'SolidFire' in driver:
# SolidFire driver does update_cluster stat on init
self.data = EMCSMISCommonData()
self.tempdir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tempdir)
super(EMCSMISFCDriverTestCase, self).setUp()
self.config_file_path = None
self.create_fake_config_file()
+ self.addCleanup(os.remove, self.config_file_path)
configuration = mock.Mock()
configuration.cinder_emc_config_file = self.config_file_path
volume_with_vt = self.data.test_volume
volume_with_vt['volume_type_id'] = 1
self.driver.create_volume(volume_with_vt)
-
- def _cleanup(self):
- bExists = os.path.exists(self.config_file_path)
- if bExists:
- os.remove(self.config_file_path)
- shutil.rmtree(self.tempdir)
-
- def tearDown(self):
- self._cleanup()
- super(EMCSMISFCDriverTestCase, self).tearDown()
super(HuaweiUtilsTestCase, self).setUp()
self.tmp_dir = tempfile.mkdtemp()
+ self.addCleanup(shutil.rmtree, self.tmp_dir)
self.fake_conf_file = self.tmp_dir + '/cinder_huawei_conf.xml'
+ self.addCleanup(os.remove, self.fake_conf_file)
create_fake_conf_file(self.fake_conf_file)
- def tearDown(self):
- if os.path.exists(self.fake_conf_file):
- os.remove(self.fake_conf_file)
- shutil.rmtree(self.tmp_dir)
- super(HuaweiUtilsTestCase, self).tearDown()
-
def test_parse_xml_file_ioerror(self):
tmp_fonf_file = '/xxx/cinder_huawei_conf.xml'
self.assertRaises(IOError, huawei_utils.parse_xml_file, tmp_fonf_file)
self.snapshot = dict(volume_name=self.volume_name,
name=self.snapshot_name)
- def tearDown(self):
- super(RBDTestCase, self).tearDown()
-
@common_mocks
def test_create_volume(self):
client = self.mock_client.return_value
self.data_length = 1024
self.full_data = 'abcd' * 256
- def tearDown(self):
- super(RBDImageIOWrapperTestCase, self).tearDown()
-
def test_init(self):
self.assertEqual(self.mock_rbd_wrapper._rbd_meta, self.meta)
self.assertEqual(self.mock_rbd_wrapper._offset, 0)
class VolumeConfigurationTest(test.TestCase):
- def setUp(self):
- super(VolumeConfigurationTest, self).setUp()
-
- def tearDown(self):
- super(VolumeConfigurationTest, self).tearDown()
def test_group_grafts_opts(self):
c = configuration.Configuration(volume_opts, config_group='foo')
self.stubs.Set(httplib, 'HTTPSConnection', FakeHTTPSConnection)
self.driver.do_setup(None)
- def tearDown(self):
- super(ZadaraVPSADriverTestCase, self).tearDown()
-
def test_create_destroy(self):
"""Create/Delete volume."""
volume = {'name': 'test_volume_01', 'size': 1}