Added Backup/Restore validations

Added checks for backup-create API call to ensure
datastore supports backup strategy
Also for trove-create API call (when restoring from backup)
added checks to ensure that datastore version of backup
being restored matches with datastore version of
instance being created

Change-Id: I9f9cf76ce3d2533612515e6382a184e955e09a32
Closes-Bug: #1285876

4 files changed, 50 insertions, 2 deletions

diff --git a/trove/backup/models.py b/trove/backup/models.py
index da9397b4..25adb8b1 100644
--- a/trove/backup/models.py
+++ b/trove/backup/models.py
@@ -45,6 +45,17 @@ class BackupState(object):
 class Backup(object):
 
     @classmethod
+    def validate_can_perform_action(cls, instance, operation):
+        """
+        Raises exception if backup strategy is not supported
+        """
+        datastore_cfg = CONF.get(instance.datastore_version.manager)
+        if not datastore_cfg or not (
+                datastore_cfg.get('backup_strategy', None)):
+            raise exception.DatastoreOperationNotSupported(
+                operation=operation, datastore=instance.datastore.name)
+
+    @classmethod
     def create(cls, context, instance, name, description=None, parent_id=None):
         """
         create db record for Backup
@@ -64,7 +75,8 @@ class Backup(object):
             from trove.instance.models import Instance
             instance_model = Instance.load(context, instance_id)
             instance_model.validate_can_perform_action()
-
+            cls.validate_can_perform_action(
+                instance_model, 'backup_create')
             cls.verify_swift_auth_token(context)
 
             parent = None
diff --git a/trove/common/exception.py b/trove/common/exception.py
index 245c2a48..b6eb31b8 100644
--- a/trove/common/exception.py
+++ b/trove/common/exception.py
@@ -349,8 +349,13 @@ class BackupFileNotFound(NotFound):
                 "storage.")
 
 
-class SwiftAuthError(TroveError):
+class BackupDatastoreVersionMismatchError(TroveError):
+    message = _("The datastore-version from which the backup was"
+                " taken, %(version1)s, does not match the destination"
+                " datastore-version of %(version2)s")
+
 
+class SwiftAuthError(TroveError):
     message = _("Swift account not accessible for tenant %(tenant_id)s.")
 
 
diff --git a/trove/instance/models.py b/trove/instance/models.py
index cb6f5d10..6d17e720 100644
--- a/trove/instance/models.py
+++ b/trove/instance/models.py
@@ -639,6 +639,17 @@ class Instance(BuiltInstance):
                 raise exception.BackupFileNotFound(
                     location=backup_info.location)
 
+            backup_db_info = DBInstance.find_by(
+                context=context, id=backup_info.instance_id)
+            if (backup_db_info.datastore_version_id
+                    != datastore_version.id):
+                ds_version = (datastore_models.DatastoreVersion.
+                              load_by_uuid(backup_db_info.datastore_version_id)
+                              )
+                raise exception.BackupDatastoreVersionMismatchError(
+                    version1=ds_version.name,
+                    version2=datastore_version.name)
+
         if not nics and CONF.default_neutron_networks:
             nics = []
             for net_id in CONF.default_neutron_networks:
diff --git a/trove/tests/unittests/backup/test_backup_models.py b/trove/tests/unittests/backup/test_backup_models.py
index 451b84e8..a24d4f89 100644
--- a/trove/tests/unittests/backup/test_backup_models.py
+++ b/trove/tests/unittests/backup/test_backup_models.py
@@ -58,6 +58,8 @@ class BackupCreateTest(testtools.TestCase):
         when(instance_models.BuiltInstance).load(any(), any()).thenReturn(
             instance)
         when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).validate_can_perform_action(
+            any(), any()).thenReturn(None)
         when(models.Backup).verify_swift_auth_token(any()).thenReturn(
             None)
         when(api.API).create_backup(any()).thenReturn(None)
@@ -84,6 +86,8 @@ class BackupCreateTest(testtools.TestCase):
         when(instance_models.BuiltInstance).load(any(), any()).thenReturn(
             instance)
         when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).validate_can_perform_action(
+            any(), any()).thenReturn(None)
         when(models.Backup).verify_swift_auth_token(any()).thenReturn(
             None)
         when(api.API).create_backup(any()).thenReturn(None)
@@ -114,6 +118,8 @@ class BackupCreateTest(testtools.TestCase):
         when(instance_models.BuiltInstance).load(any(), any()).thenReturn(
             instance)
         when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).validate_can_perform_action(
+            any(), any()).thenReturn(None)
         when(models.Backup).verify_swift_auth_token(any()).thenReturn(
             None)
         self.assertRaises(exception.NotFound, models.Backup.create,
@@ -135,12 +141,26 @@ class BackupCreateTest(testtools.TestCase):
         when(instance_models.BuiltInstance).load(any(), any()).thenReturn(
             instance)
         when(instance).validate_can_perform_action().thenReturn(None)
+        when(models.Backup).validate_can_perform_action(
+            any(), any()).thenReturn(None)
         when(models.Backup).verify_swift_auth_token(any()).thenRaise(
             exception.SwiftAuthError)
         self.assertRaises(exception.SwiftAuthError, models.Backup.create,
                           self.context, self.instance_id,
                           BACKUP_NAME, BACKUP_DESC)
 
+    def test_create_backup_datastore_operation_not_supported(self):
+        instance = mock(instance_models.Instance)
+        when(instance_models.BuiltInstance).load(any(), any()).thenReturn(
+            instance)
+        when(models.Backup).validate_can_perform_action(
+            any(), any()).thenRaise(
+                exception.DatastoreOperationNotSupported)
+        self.assertRaises(exception.DatastoreOperationNotSupported,
+                          models.Backup.create,
+                          self.context, self.instance_id,
+                          BACKUP_NAME, BACKUP_DESC)
+
 
 class BackupDeleteTest(testtools.TestCase):
     def setUp(self):