From 2e3c1cd77b3d523c48ff4a086cccf68fd11c4c19 Mon Sep 17 00:00:00 2001 From: Michael Kerrin Date: Wed, 17 Apr 2013 15:36:28 +0000 Subject: [PATCH] Dont delete backup record from database Instead we should be marking the backup as deleted so that we have an audit of all the backups. Backups are now marked deleted in the way that volumes are. Fixes bug: 1169943 Fixes bug: 1169857 Change-Id: I52c603be138c0f1d6d411d167977041255ac9053 (cherry picked from commit a4ba17b668bd893d50fd4f3a5453da89966a77e3) --- cinder/db/sqlalchemy/api.py | 22 +++++++++----- cinder/tests/test_backup.py | 58 +++++++++++++++++++++++++++++++++++-- 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/cinder/db/sqlalchemy/api.py b/cinder/db/sqlalchemy/api.py index f5801a84fd..6b03046a1b 100644 --- a/cinder/db/sqlalchemy/api.py +++ b/cinder/db/sqlalchemy/api.py @@ -1920,28 +1920,32 @@ def sm_volume_get_all(context): @require_context def backup_get(context, backup_id, session=None): result = model_query(context, models.Backup, - read_deleted="yes").filter_by(id=backup_id).first() + session=session, project_only=True).\ + filter_by(id=backup_id).\ + first() + if not result: raise exception.BackupNotFound(backup_id=backup_id) + return result @require_admin_context def backup_get_all(context): - return model_query(context, models.Backup, read_deleted="yes").all() + return model_query(context, models.Backup).all() @require_admin_context def backup_get_all_by_host(context, host): - return model_query(context, models.Backup, - read_deleted="yes").filter_by(host=host).all() + return model_query(context, models.Backup).filter_by(host=host).all() @require_context def backup_get_all_by_project(context, project_id): authorize_project_context(context, project_id) - return model_query(context, models.Backup, read_deleted="yes").all() + return model_query(context, models.Backup).\ + filter_by(project_id=project_id).all() @require_context @@ -1975,5 +1979,9 @@ def backup_update(context, backup_id, values): def backup_destroy(context, backup_id): session = get_session() with session.begin(): - model_query(context, models.Backup, - read_deleted="yes").filter_by(id=backup_id).delete() + session.query(models.Backup).\ + filter_by(id=backup_id).\ + update({'status': 'deleted', + 'deleted': True, + 'deleted_at': timeutils.utcnow(), + 'updated_at': literal_column('updated_at')}) diff --git a/cinder/tests/test_backup.py b/cinder/tests/test_backup.py index b97a76f4c1..b42cfb17f9 100644 --- a/cinder/tests/test_backup.py +++ b/cinder/tests/test_backup.py @@ -25,6 +25,7 @@ from cinder import flags from cinder.openstack.common import importutils from cinder.openstack.common import log as logging +from cinder.openstack.common import timeutils from cinder import test FLAGS = flags.FLAGS @@ -56,7 +57,8 @@ def _create_backup_db_entry(self, volume_id=1, display_name='test_backup', container='volumebackups', status='creating', size=0, - object_count=0): + object_count=0, + project_id='fake'): """ Create a backup entry in the DB. Return the entry ID @@ -64,7 +66,7 @@ def _create_backup_db_entry(self, volume_id=1, display_name='test_backup', backup = {} backup['volume_id'] = volume_id backup['user_id'] = 'fake' - backup['project_id'] = 'fake' + backup['project_id'] = project_id backup['host'] = 'testhost' backup['availability_zone'] = '1' backup['display_name'] = display_name @@ -330,3 +332,55 @@ def test_delete_backup(self): 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.assertTrue(timeutils.utcnow() > backup.deleted_at) + self.assertEqual(backup.status, 'deleted') + + 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) + + 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) + + 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)