¡@
OpenStack Study: test_backup.py**** CubicPower OpenStack Study ****
# Copyright (C) 2012 Hewlett-Packard Development Company, L.P. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. """ Tests for Backup code. """ import mock import tempfile from oslo.config import cfg from cinder import context from cinder import db from cinder import exception from cinder.openstack.common import importutils from cinder.openstack.common import log as logging from cinder.openstack.common import timeutils from cinder import test CONF = cfg.CONF LOG = logging.getLogger(__name__)
**** CubicPower OpenStack Study **** class FakeBackupException(Exception): pass
**** CubicPower OpenStack Study **** class BackupTestCase(test.TestCase): """Test Case for backups."""
**** CubicPower OpenStack Study **** def setUp(self): super(BackupTestCase, self).setUp() vol_tmpdir = tempfile.mkdtemp() self.flags(volumes_dir=vol_tmpdir) self.backup_mgr = \ importutils.import_object(CONF.backup_manager) self.backup_mgr.host = 'testhost' self.ctxt = context.get_admin_context() self.backup_mgr.driver.set_initialized()
**** CubicPower OpenStack Study **** def tearDown(self): super(BackupTestCase, self).tearDown()
**** CubicPower OpenStack Study **** def _create_backup_db_entry(self, volume_id=1, display_name='test_backup', display_description='this is a test backup', container='volumebackups', status='creating', size=1, object_count=0, project_id='fake'): """Create a backup entry in the DB. Return the entry ID """ backup = {} backup['volume_id'] = volume_id backup['user_id'] = 'fake' backup['project_id'] = project_id backup['host'] = 'testhost' backup['availability_zone'] = '1' backup['display_name'] = display_name backup['display_description'] = display_description backup['container'] = container backup['status'] = status backup['fail_reason'] = '' backup['service'] = CONF.backup_driver backup['size'] = size backup['object_count'] = object_count return db.backup_create(self.ctxt, backup)['id']
**** CubicPower OpenStack Study **** def _create_volume_db_entry(self, display_name='test_volume', display_description='this is a test volume', status='backing-up', size=1): """Create a volume entry in the DB. Return the entry ID """ vol = {} vol['size'] = size vol['host'] = 'testhost' vol['user_id'] = 'fake' vol['project_id'] = 'fake' vol['status'] = status vol['display_name'] = display_name vol['display_description'] = display_description vol['attach_status'] = 'detached' return db.volume_create(self.ctxt, vol)['id']
**** CubicPower OpenStack Study **** def _create_exported_record_entry(self, vol_size=1): """Create backup metadata export entry.""" vol_id = self._create_volume_db_entry(status='available', size=vol_size) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) export = self.backup_mgr.export_record(self.ctxt, backup_id) return export
**** CubicPower OpenStack Study **** def _create_export_record_db_entry(self, volume_id='0000', status='creating', project_id='fake'): """Create a backup entry in the DB. Return the entry ID """ backup = {} backup['volume_id'] = volume_id backup['user_id'] = 'fake' backup['project_id'] = project_id backup['status'] = status return db.backup_create(self.ctxt, backup)['id']
**** CubicPower OpenStack Study **** def test_init_host(self): """Make sure stuck volumes and backups are reset to correct states when backup_manager.init_host() is called """ vol1_id = self._create_volume_db_entry(status='backing-up') vol2_id = self._create_volume_db_entry(status='restoring-backup') backup1_id = self._create_backup_db_entry(status='creating') backup2_id = self._create_backup_db_entry(status='restoring') backup3_id = self._create_backup_db_entry(status='deleting') self.backup_mgr.init_host() vol1 = db.volume_get(self.ctxt, vol1_id) self.assertEqual(vol1['status'], 'available') vol2 = db.volume_get(self.ctxt, vol2_id) self.assertEqual(vol2['status'], 'error_restoring') backup1 = db.backup_get(self.ctxt, backup1_id) self.assertEqual(backup1['status'], 'error') backup2 = db.backup_get(self.ctxt, backup2_id) self.assertEqual(backup2['status'], 'available') self.assertRaises(exception.BackupNotFound, db.backup_get, self.ctxt, backup3_id)
**** CubicPower OpenStack Study **** def test_create_backup_with_bad_volume_status(self): """Test error handling when creating a backup from a volume with a bad status """ vol_id = self._create_volume_db_entry(status='available', size=1) backup_id = self._create_backup_db_entry(volume_id=vol_id) self.assertRaises(exception.InvalidVolume, self.backup_mgr.create_backup, self.ctxt, backup_id)
**** CubicPower OpenStack Study **** def test_create_backup_with_bad_backup_status(self): """Test error handling when creating a backup with a backup with a bad status """ vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) self.assertRaises(exception.InvalidBackup, self.backup_mgr.create_backup, self.ctxt, backup_id) @mock.patch('%s.%s' % (CONF.volume_driver, 'backup_volume'))
**** CubicPower OpenStack Study **** def test_create_backup_with_error(self, _mock_volume_backup): """Test error handling when error occurs during backup creation.""" vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(volume_id=vol_id) _mock_volume_backup.side_effect = FakeBackupException('fake') self.assertRaises(FakeBackupException, self.backup_mgr.create_backup, self.ctxt, backup_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'available') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'error') self.assertTrue(_mock_volume_backup.called) @mock.patch('%s.%s' % (CONF.volume_driver, 'backup_volume'))
**** CubicPower OpenStack Study **** def test_create_backup(self, _mock_volume_backup): """Test normal backup creation.""" vol_size = 1 vol_id = self._create_volume_db_entry(size=vol_size) backup_id = self._create_backup_db_entry(volume_id=vol_id) self.backup_mgr.create_backup(self.ctxt, backup_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'available') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'available') self.assertEqual(backup['size'], vol_size) self.assertTrue(_mock_volume_backup.called)
**** CubicPower OpenStack Study **** def test_restore_backup_with_bad_volume_status(self): """Test error handling when restoring a backup to a volume with a bad status. """ vol_id = self._create_volume_db_entry(status='available', size=1) backup_id = self._create_backup_db_entry(volume_id=vol_id) self.assertRaises(exception.InvalidVolume, self.backup_mgr.restore_backup, self.ctxt, backup_id, vol_id) backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'available')
**** CubicPower OpenStack Study **** def test_restore_backup_with_bad_backup_status(self): """Test error handling when restoring a backup with a backup with a bad status. """ vol_id = self._create_volume_db_entry(status='restoring-backup', size=1) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) self.assertRaises(exception.InvalidBackup, self.backup_mgr.restore_backup, self.ctxt, backup_id, vol_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'error') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'error') @mock.patch('%s.%s' % (CONF.volume_driver, 'restore_backup'))
**** CubicPower OpenStack Study **** def test_restore_backup_with_driver_error(self, _mock_volume_restore): """Test error handling when an error occurs during backup restore.""" vol_id = self._create_volume_db_entry(status='restoring-backup', size=1) backup_id = self._create_backup_db_entry(status='restoring', volume_id=vol_id) _mock_volume_restore.side_effect = FakeBackupException('fake') self.assertRaises(FakeBackupException, self.backup_mgr.restore_backup, self.ctxt, backup_id, vol_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'error_restoring') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'available') self.assertTrue(_mock_volume_restore.called)
**** CubicPower OpenStack Study **** def test_restore_backup_with_bad_service(self): """Test error handling when attempting a restore of a backup with a different service to that used to create the backup. """ vol_id = self._create_volume_db_entry(status='restoring-backup', size=1) backup_id = self._create_backup_db_entry(status='restoring', volume_id=vol_id) service = 'cinder.tests.backup.bad_service' db.backup_update(self.ctxt, backup_id, {'service': service}) self.assertRaises(exception.InvalidBackup, self.backup_mgr.restore_backup, self.ctxt, backup_id, vol_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'error') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'available') @mock.patch('%s.%s' % (CONF.volume_driver, 'restore_backup'))
**** CubicPower OpenStack Study **** def test_restore_backup(self, _mock_volume_restore): """Test normal backup restoration.""" vol_size = 1 vol_id = self._create_volume_db_entry(status='restoring-backup', size=vol_size) backup_id = self._create_backup_db_entry(status='restoring', volume_id=vol_id) self.backup_mgr.restore_backup(self.ctxt, backup_id, vol_id) vol = db.volume_get(self.ctxt, vol_id) self.assertEqual(vol['status'], 'available') backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'available') self.assertTrue(_mock_volume_restore.called)
**** CubicPower OpenStack Study **** def test_delete_backup_with_bad_backup_status(self): """Test error handling when deleting a backup with a backup with a bad status. """ vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) self.assertRaises(exception.InvalidBackup, self.backup_mgr.delete_backup, self.ctxt, backup_id) backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'error')
**** CubicPower OpenStack Study **** def test_delete_backup_with_error(self): """Test error handling when an error occurs during backup deletion.""" vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='deleting', display_name='fail_on_delete', volume_id=vol_id) self.assertRaises(IOError, self.backup_mgr.delete_backup, self.ctxt, backup_id) backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'error')
**** CubicPower OpenStack Study **** def test_delete_backup_with_bad_service(self): """Test error handling when attempting a delete of a backup with a different service to that used to create the backup. """ vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='deleting', volume_id=vol_id) service = 'cinder.tests.backup.bad_service' db.backup_update(self.ctxt, backup_id, {'service': service}) self.assertRaises(exception.InvalidBackup, self.backup_mgr.delete_backup, self.ctxt, backup_id) backup = db.backup_get(self.ctxt, backup_id) self.assertEqual(backup['status'], 'error')
**** CubicPower OpenStack Study **** def test_delete_backup_with_no_service(self): """Test error handling when attempting a delete of a backup with no service defined for that backup, relates to bug #1162908 """ vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='deleting', volume_id=vol_id) db.backup_update(self.ctxt, backup_id, {'service': None}) self.backup_mgr.delete_backup(self.ctxt, backup_id)
**** CubicPower OpenStack Study **** def test_delete_backup(self): """Test normal backup deletion.""" vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='deleting', volume_id=vol_id) self.backup_mgr.delete_backup(self.ctxt, backup_id) self.assertRaises(exception.BackupNotFound, db.backup_get, self.ctxt, backup_id) ctxt_read_deleted = context.get_admin_context('yes') backup = db.backup_get(ctxt_read_deleted, backup_id) self.assertEqual(backup.deleted, True) self.assertGreaterEqual(timeutils.utcnow(), backup.deleted_at) self.assertEqual(backup.status, 'deleted')
**** CubicPower OpenStack Study **** def test_list_backup(self): backups = db.backup_get_all_by_project(self.ctxt, 'project1') self.assertEqual(len(backups), 0) b1 = self._create_backup_db_entry() b2 = self._create_backup_db_entry(project_id='project1') backups = db.backup_get_all_by_project(self.ctxt, 'project1') self.assertEqual(len(backups), 1) self.assertEqual(backups[0].id, b2)
**** CubicPower OpenStack Study **** def test_backup_get_all_by_project_with_deleted(self): """Test deleted backups don't show up in backup_get_all_by_project. Unless context.read_deleted is 'yes'. """ backups = db.backup_get_all_by_project(self.ctxt, 'fake') self.assertEqual(len(backups), 0) backup_id_keep = self._create_backup_db_entry() backup_id = self._create_backup_db_entry() db.backup_destroy(self.ctxt, backup_id) backups = db.backup_get_all_by_project(self.ctxt, 'fake') self.assertEqual(len(backups), 1) self.assertEqual(backups[0].id, backup_id_keep) ctxt_read_deleted = context.get_admin_context('yes') backups = db.backup_get_all_by_project(ctxt_read_deleted, 'fake') self.assertEqual(len(backups), 2)
**** CubicPower OpenStack Study **** def test_backup_get_all_by_host_with_deleted(self): """Test deleted backups don't show up in backup_get_all_by_project. Unless context.read_deleted is 'yes' """ backups = db.backup_get_all_by_host(self.ctxt, 'testhost') self.assertEqual(len(backups), 0) backup_id_keep = self._create_backup_db_entry() backup_id = self._create_backup_db_entry() db.backup_destroy(self.ctxt, backup_id) backups = db.backup_get_all_by_host(self.ctxt, 'testhost') self.assertEqual(len(backups), 1) self.assertEqual(backups[0].id, backup_id_keep) ctxt_read_deleted = context.get_admin_context('yes') backups = db.backup_get_all_by_host(ctxt_read_deleted, 'testhost') self.assertEqual(len(backups), 2)
**** CubicPower OpenStack Study **** def test_backup_manager_driver_name(self): """"Test mapping between backup services and backup drivers.""" old_setting = CONF.backup_driver setattr(cfg.CONF, 'backup_driver', "cinder.backup.services.swift") backup_mgr = \ importutils.import_object(CONF.backup_manager) self.assertEqual('cinder.backup.drivers.swift', backup_mgr.driver_name) setattr(cfg.CONF, 'backup_driver', old_setting)
**** CubicPower OpenStack Study **** def test_export_record_with_bad_service(self): """Test error handling when attempting an export of a backup record with a different service to that used to create the backup. """ vol_id = self._create_volume_db_entry(size=1) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) service = 'cinder.tests.backup.bad_service' db.backup_update(self.ctxt, backup_id, {'service': service}) self.assertRaises(exception.InvalidBackup, self.backup_mgr.export_record, self.ctxt, backup_id)
**** CubicPower OpenStack Study **** def test_export_record_with_bad_backup_status(self): """Test error handling when exporting a backup record with a backup with a bad status. """ vol_id = self._create_volume_db_entry(status='available', size=1) backup_id = self._create_backup_db_entry(status='error', volume_id=vol_id) self.assertRaises(exception.InvalidBackup, self.backup_mgr.export_record, self.ctxt, backup_id)
**** CubicPower OpenStack Study **** def test_export_record(self): """Test normal backup record export.""" vol_size = 1 vol_id = self._create_volume_db_entry(status='available', size=vol_size) backup_id = self._create_backup_db_entry(status='available', volume_id=vol_id) export = self.backup_mgr.export_record(self.ctxt, backup_id) self.assertEqual(export['backup_service'], CONF.backup_driver) self.assertTrue('backup_url' in export)
**** CubicPower OpenStack Study **** def test_import_record_with_verify_not_implemented(self): """Test normal backup record import. Test the case when import succeeds for the case that the driver does not support verify. """ vol_size = 1 export = self._create_exported_record_entry(vol_size=vol_size) imported_record = self._create_export_record_db_entry() backup_hosts = [] backup_driver = self.backup_mgr.service.get_backup_driver(self.ctxt) _mock_backup_verify_class = ('%s.%s.%s' % (backup_driver.__module__, backup_driver.__class__.__name__, 'verify')) with mock.patch(_mock_backup_verify_class) as _mock_record_verify: _mock_record_verify.side_effect = NotImplementedError() self.backup_mgr.import_record(self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) backup = db.backup_get(self.ctxt, imported_record) self.assertEqual(backup['status'], 'available') self.assertEqual(backup['size'], vol_size)
**** CubicPower OpenStack Study **** def test_import_record_with_verify(self): """Test normal backup record import. Test the case when import succeeds for the case that the driver implements verify. """ vol_size = 1 export = self._create_exported_record_entry(vol_size=vol_size) imported_record = self._create_export_record_db_entry() backup_hosts = [] backup_driver = self.backup_mgr.service.get_backup_driver(self.ctxt) _mock_backup_verify_class = ('%s.%s.%s' % (backup_driver.__module__, backup_driver.__class__.__name__, 'verify')) with mock.patch(_mock_backup_verify_class) as _mock_record_verify: self.backup_mgr.import_record(self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) backup = db.backup_get(self.ctxt, imported_record) self.assertEqual(backup['status'], 'available') self.assertEqual(backup['size'], vol_size)
**** CubicPower OpenStack Study **** def test_import_record_with_bad_service(self): """Test error handling when attempting an import of a backup record with a different service to that used to create the backup. """ export = self._create_exported_record_entry() export['backup_service'] = 'cinder.tests.backup.bad_service' imported_record = self._create_export_record_db_entry() #Test the case where the additional hosts list is empty backup_hosts = [] self.assertRaises(exception.ServiceNotFound, self.backup_mgr.import_record, self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) #Test that the import backup keeps calling other hosts to find a #suitable host for the backup service backup_hosts = ['fake1', 'fake2'] BackupAPI_import = 'cinder.backup.rpcapi.BackupAPI.import_record' with mock.patch(BackupAPI_import) as _mock_backup_import: self.backup_mgr.import_record(self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) self.assertTrue(_mock_backup_import.called)
**** CubicPower OpenStack Study **** def test_import_record_with_invalid_backup(self): """Test error handling when attempting an import of a backup record where the backup driver returns an exception. """ export = self._create_exported_record_entry() backup_driver = self.backup_mgr.service.get_backup_driver(self.ctxt) _mock_record_import_class = ('%s.%s.%s' % (backup_driver.__module__, backup_driver.__class__.__name__, 'import_record')) imported_record = self._create_export_record_db_entry() backup_hosts = [] with mock.patch(_mock_record_import_class) as _mock_record_import: _mock_record_import.side_effect = FakeBackupException('fake') self.assertRaises(exception.InvalidBackup, self.backup_mgr.import_record, self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) self.assertTrue(_mock_record_import.called) backup = db.backup_get(self.ctxt, imported_record) self.assertEqual(backup['status'], 'error')
**** CubicPower OpenStack Study **** def test_import_record_with_verify_invalid_backup(self): """Test error handling when attempting an import of a backup record where the backup driver returns an exception. """ vol_size = 1 export = self._create_exported_record_entry(vol_size=vol_size) imported_record = self._create_export_record_db_entry() backup_hosts = [] backup_driver = self.backup_mgr.service.get_backup_driver(self.ctxt) _mock_backup_verify_class = ('%s.%s.%s' % (backup_driver.__module__, backup_driver.__class__.__name__, 'verify')) with mock.patch(_mock_backup_verify_class) as _mock_record_verify: _mock_record_verify.side_effect = \ exception.InvalidBackup(reason='fake') self.assertRaises(exception.InvalidBackup, self.backup_mgr.import_record, self.ctxt, imported_record, export['backup_service'], export['backup_url'], backup_hosts) self.assertTrue(_mock_record_verify.called) backup = db.backup_get(self.ctxt, imported_record) self.assertEqual(backup['status'], 'error')
|