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)

diff --git a/cinder/db/sqlalchemy/api.py b/cinder/db/sqlalchemy/api.py
index f5801a84fdc59f96d2b4a13282a990e5eb0a8095..6b03046a1b54a154bec5bdf13441830cff2a3710 100644 (file)
--- 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 b97a76f4c12586e9eac63899ff552af4663c242a..b42cfb17f94dd1167f3e1d46f5cedb0d147b6dc9 100644 (file)
--- a/cinder/tests/test_backup.py
+++ b/cinder/tests/test_backup.py
@@ -25,6 +25,7 @@ from cinder import exception
 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 @@ class BackupTestCase(test.TestCase):
                                 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 @@ class BackupTestCase(test.TestCase):
         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 @@ class BackupTestCase(test.TestCase):
                           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)