Adding test case for migration policy 'never'

There doesn't exist any test case in tempest for retype
volume without migration, so adding new test for it.

Related-Bug: #1740544

Change-Id: I58c348b954a679bc809906b2c73df9b1ab4459c5
diff --git a/tempest/api/volume/admin/test_volume_retype_with_migration.py b/tempest/api/volume/admin/test_volume_retype.py
similarity index 68%
rename from tempest/api/volume/admin/test_volume_retype_with_migration.py
rename to tempest/api/volume/admin/test_volume_retype.py
index 025c1be..423104b 100644
--- a/tempest/api/volume/admin/test_volume_retype_with_migration.py
+++ b/tempest/api/volume/admin/test_volume_retype.py
@@ -10,6 +10,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import abc
 
 from oslo_log import log as logging
 
@@ -23,31 +24,7 @@
 LOG = logging.getLogger(__name__)
 
 
-class VolumeRetypeWithMigrationTest(base.BaseVolumeAdminTest):
-
-    @classmethod
-    def skip_checks(cls):
-        super(VolumeRetypeWithMigrationTest, cls).skip_checks()
-
-        if not CONF.volume_feature_enabled.multi_backend:
-            raise cls.skipException("Cinder multi-backend feature disabled.")
-
-        if len(set(CONF.volume.backend_names)) < 2:
-            raise cls.skipException("Requires at least two different "
-                                    "backend names")
-
-    @classmethod
-    def resource_setup(cls):
-        super(VolumeRetypeWithMigrationTest, cls).resource_setup()
-        # read backend name from a list.
-        backend_src = CONF.volume.backend_names[0]
-        backend_dst = CONF.volume.backend_names[1]
-
-        extra_specs_src = {"volume_backend_name": backend_src}
-        extra_specs_dst = {"volume_backend_name": backend_dst}
-
-        cls.src_vol_type = cls.create_volume_type(extra_specs=extra_specs_src)
-        cls.dst_vol_type = cls.create_volume_type(extra_specs=extra_specs_dst)
+class VolumeRetypeTest(base.BaseVolumeAdminTest):
 
     def _wait_for_internal_volume_cleanup(self, vol):
         # When retyping a volume, Cinder creates an internal volume in the
@@ -70,10 +47,28 @@
                     fetched_vol['id'])
                 break
 
-    def _retype_volume(self, volume):
-        keys_with_no_change = ('id', 'size', 'description', 'name', 'user_id',
-                               'os-vol-tenant-attr:tenant_id')
-        keys_with_change = ('volume_type', 'os-vol-host-attr:host')
+    @abc.abstractmethod
+    def _verify_migration(self, source_vol, dest_vol):
+        pass
+
+    def _create_volume_from_snapshot(self):
+        # Create a volume in the first backend
+        src_vol = self.create_volume(volume_type=self.src_vol_type['name'])
+
+        # Create a volume snapshot
+        snapshot = self.create_snapshot(src_vol['id'])
+
+        # Create a volume from the snapshot
+        src_vol = self.create_volume(volume_type=self.src_vol_type['name'],
+                                     snapshot_id=snapshot['id'])
+
+        # Delete the snapshot
+        self.snapshots_client.delete_snapshot(snapshot['id'])
+        self.snapshots_client.wait_for_resource_deletion(snapshot['id'])
+
+        return src_vol
+
+    def _retype_volume(self, volume, migration_policy):
 
         volume_source = self.admin_volume_client.show_volume(
             volume['id'])['volume']
@@ -81,7 +76,7 @@
         self.volumes_client.retype_volume(
             volume['id'],
             new_type=self.dst_vol_type['name'],
-            migration_policy='on-demand')
+            migration_policy=migration_policy)
         self.addCleanup(self._wait_for_internal_volume_cleanup, volume)
         waiters.wait_for_volume_retype(self.volumes_client, volume['id'],
                                        self.dst_vol_type['name'])
@@ -89,6 +84,41 @@
         volume_dest = self.admin_volume_client.show_volume(
             volume['id'])['volume']
 
+        self._verify_migration(volume_source, volume_dest)
+
+
+class VolumeRetypeWithMigrationTest(VolumeRetypeTest):
+
+    @classmethod
+    def skip_checks(cls):
+        super(VolumeRetypeTest, cls).skip_checks()
+
+        if not CONF.volume_feature_enabled.multi_backend:
+            raise cls.skipException("Cinder multi-backend feature disabled.")
+
+        if len(set(CONF.volume.backend_names)) < 2:
+            raise cls.skipException("Requires at least two different "
+                                    "backend names")
+
+    @classmethod
+    def resource_setup(cls):
+        super(VolumeRetypeWithMigrationTest, cls).resource_setup()
+        # read backend name from a list.
+        backend_src = CONF.volume.backend_names[0]
+        backend_dst = CONF.volume.backend_names[1]
+
+        extra_specs_src = {"volume_backend_name": backend_src}
+        extra_specs_dst = {"volume_backend_name": backend_dst}
+
+        cls.src_vol_type = cls.create_volume_type(extra_specs=extra_specs_src)
+        cls.dst_vol_type = cls.create_volume_type(extra_specs=extra_specs_dst)
+
+    def _verify_migration(self, volume_source, volume_dest):
+
+        keys_with_no_change = ('id', 'size', 'description', 'name',
+                               'user_id', 'os-vol-tenant-attr:tenant_id')
+        keys_with_change = ('volume_type', 'os-vol-host-attr:host')
+
         # Check the volume information after the migration.
         self.assertEqual('success',
                          volume_dest['os-vol-mig-status-attr:migstat'])
@@ -103,23 +133,44 @@
     @decorators.idempotent_id('a1a41f3f-9dad-493e-9f09-3ff197d477cd')
     def test_available_volume_retype_with_migration(self):
         src_vol = self.create_volume(volume_type=self.src_vol_type['name'])
-        self._retype_volume(src_vol)
+        self._retype_volume(src_vol, migration_policy='on-demand')
 
     @decorators.idempotent_id('d0d9554f-e7a5-4104-8973-f35b27ccb60d')
     def test_volume_from_snapshot_retype_with_migration(self):
-        # Create a volume in the first backend
-        src_vol = self.create_volume(volume_type=self.src_vol_type['name'])
-
-        # Create a volume snapshot
-        snapshot = self.create_snapshot(src_vol['id'])
-
-        # Create a volume from the snapshot
-        src_vol = self.create_volume(volume_type=self.src_vol_type['name'],
-                                     snapshot_id=snapshot['id'])
-
-        # Delete the snapshot
-        self.snapshots_client.delete_snapshot(snapshot['id'])
-        self.snapshots_client.wait_for_resource_deletion(snapshot['id'])
+        src_vol = self._create_volume_from_snapshot()
 
         # Migrate the volume from snapshot to the second backend
-        self._retype_volume(src_vol)
+        self._retype_volume(src_vol, migration_policy='on-demand')
+
+
+class VolumeRetypeWithoutMigrationTest(VolumeRetypeTest):
+
+    @classmethod
+    def resource_setup(cls):
+        super(VolumeRetypeWithoutMigrationTest, cls).resource_setup()
+        cls.src_vol_type = cls.create_volume_type('volume-type-1')
+        cls.dst_vol_type = cls.create_volume_type('volume-type-2')
+
+    def _verify_migration(self, volume_source, volume_dest):
+
+        keys_with_no_change = ('id', 'size', 'description', 'name',
+                               'user_id', 'os-vol-tenant-attr:tenant_id',
+                               'os-vol-host-attr:host')
+        keys_with_change = ('volume_type',)
+
+        # Check the volume information after the retype
+        self.assertIsNone(volume_dest['os-vol-mig-status-attr:migstat'])
+        self.assertIsNone(volume_dest['migration_status'])
+
+        for key in keys_with_no_change:
+            self.assertEqual(volume_source[key], volume_dest[key])
+
+        for key in keys_with_change:
+            self.assertNotEqual(volume_source[key], volume_dest[key])
+
+    @decorators.idempotent_id('b90412ee-465d-46e9-b249-ec84a47d5f25')
+    def test_available_volume_retype(self):
+        src_vol = self.create_volume(volume_type=self.src_vol_type['name'])
+
+        # Retype the volume from snapshot
+        self._retype_volume(src_vol, migration_policy='never')