Make sure the converted image is imported

Turns out the converted image is being completely ignored. This is
because the `file_path` provided by the ImportToFS task cannot be
overriden by other tasks. In order to fix this, I've dropped all the
extra suffixes that we've been using in favor of working on the actual
`file_path` directly.

Change-Id: Ifbd2308057f33c1f712fa06f9bb3e57f90068fdc
Closes-bug: #1452750

glance/async/flows/base_import.py

@@ -145,10 +145,7 @@ class _ImportToFS(task.Task):
         # refer to the comment in the `_ImportToStore.execute` method.
         data = script_utils.get_image_data_iter(self.uri)
 
-        # NOTE(jokke): Using .tasks_import to ease debugging. The file name
-        # is specific so we know exactly where it's coming from.
-        tmp_id = "%s.tasks_import" % image_id
-        path = self.store.add(tmp_id, data, 0, context=None)[0]
+        path = self.store.add(image_id, data, 0, context=None)[0]
         return path
 
     def revert(self, image_id, result=None, **kwargs):

glance/async/flows/convert.py

@@ -14,6 +14,7 @@
 #    under the License.
 
 import logging
+import os
 
 from oslo_concurrency import processutils as putils
 from oslo_config import cfg
@@ -75,7 +76,7 @@ class _Convert(task.Task):
         # TODO(flaper87): Check whether the image is in the desired
         # format already. Probably using `qemu-img` just like the
         # `Introspection` task.
-        dest_path = "%s.converted"
+        dest_path = os.path.join(CONF.task.work_dir, "%s.converted" % image_id)
         stdout, stderr = putils.trycmd('qemu-img', 'convert', '-O',
                                        conversion_format, file_path, dest_path,
                                        log_errors=putils.LOG_ALL_ERRORS)
@@ -83,6 +84,20 @@ class _Convert(task.Task):
         if stderr:
             raise RuntimeError(stderr)
 
+        os.rename(dest_path, file_path.split("file://")[-1])
+        return file_path
+
+    def revert(self, image_id, result=None, **kwargs):
+        # NOTE(flaper87): If result is None, it probably
+        # means this task failed. Otherwise, we would have
+        # a result from its execution.
+        if result is None:
+            return
+
+        fs_path = result.split("file://")[-1]
+        if os.path.exists(fs_path):
+            os.path.remove(fs_path)
+
 
 def get_flow(**kwargs):
     task_id = kwargs.get('task_id')

glance/tests/unit/async/flows/test_convert.py

@@ -90,7 +90,9 @@ class TestImportTask(test_utils.BaseTestCase):
 
         with mock.patch.object(processutils, 'execute') as exc_mock:
             exc_mock.return_value = ("", None)
-            image_convert.execute(image, '/test/path.raw')
+            with mock.patch.object(os, 'rename') as rm_mock:
+                rm_mock.return_value = None
+                image_convert.execute(image, 'file:///test/path.raw')
 
     def test_import_flow_with_convert_and_introspect(self):
         self.config(engine_mode='serial',
@@ -117,11 +119,16 @@ class TestImportTask(test_utils.BaseTestCase):
         self.img_repo.get.return_value = image
         img_factory.new_image.side_effect = create_image
 
-        with mock.patch.object(script_utils, 'get_image_data_iter') as dmock:
-            dmock.return_value = StringIO.StringIO("TEST_IMAGE")
+        image_path = os.path.join(self.work_dir, image.image_id)
 
-            with mock.patch.object(processutils, 'execute') as exc_mock:
-                result = json.dumps({
+        def fake_execute(*args, **kwargs):
+            if 'info' in args:
+                # NOTE(flaper87): Make sure the file actually
+                # exists. Extra check to verify previous tasks did
+                # what they were supposed to do.
+                assert os.path.exists(args[3].split("file://")[-1])
+
+                return (json.dumps({
                     "virtual-size": 10737418240,
                     "filename": "/tmp/image.qcow2",
                     "cluster-size": 65536,
@@ -134,19 +141,24 @@ class TestImportTask(test_utils.BaseTestCase):
                         }
                     },
                     "dirty-flag": False
-                })
+                }), None)
 
-                # NOTE(flaper87): First result for the conversion step and
-                # the second one for the introspection one. The later *must*
-                # come after the former. If not, the current builtin flow
-                # process will be unsound.
-                # Follow-up work will fix this by having a better way to handle
-                # task's dependencies and activation.
-                exc_mock.side_effect = [("", None), (result, None)]
+            open("%s.converted" % image_path, 'a').close()
+            return ("", None)
+
+        with mock.patch.object(script_utils, 'get_image_data_iter') as dmock:
+            dmock.return_value = StringIO.StringIO("TEST_IMAGE")
+
+            with mock.patch.object(processutils, 'execute') as exc_mock:
+                exc_mock.side_effect = fake_execute
                 executor.begin_processing(self.task.task_id)
-                image_path = os.path.join(self.test_dir, image.image_id)
-                tmp_image_path = "%s.tasks_import" % image_path
-                self.assertFalse(os.path.exists(tmp_image_path))
-                self.assertTrue(os.path.exists(image_path))
+
+                # NOTE(flaper87): DeleteFromFS should've deleted this
+                # file. Make sure it doesn't exist.
+                self.assertFalse(os.path.exists(image_path))
+
+                # NOTE(flaper87): Workdir should be empty after all
+                # the tasks have been executed.
+                self.assertEqual([], os.listdir(self.work_dir))
                 self.assertEqual('qcow2', image.disk_format)
                 self.assertEqual(10737418240, image.virtual_size)

glance/tests/unit/async/flows/test_import.py

@@ -274,8 +274,7 @@ class TestImportTask(test_utils.BaseTestCase):
             self.assertEqual(dmock.return_value, "".join(reader))
 
             image_path = os.path.join(self.work_dir, image_id)
-            tmp_image_path = os.path.join(self.work_dir,
-                                          "%s.tasks_import" % image_path)
+            tmp_image_path = os.path.join(self.work_dir, image_path)
             self.assertTrue(os.path.exists(tmp_image_path))
 
     def test_delete_from_fs(self):