Fix incorrect task status with wrong parameter

Now, when use wrong parameter to create a import task, the task's
status will be processing forever. Actually the task has been stopped
and returned error.
So in this situation, the task's status should be changed into failure.

Closes-bug: #1495519
Change-Id: Ie4b3b92aaddcdbb839aadf61c75fc0c915e2040e

glance/async/taskflow_executor.py

@@ -20,11 +20,13 @@ from oslo_config import cfg
 from oslo_log import log as logging
 from oslo_utils import excutils
 import six
+from six.moves import urllib
 from stevedore import driver
 from taskflow import engines
 from taskflow.listeners import logging as llistener
 
 import glance.async
+from glance.common import exception
 from glance.common.scripts import utils as script_utils
 from glance import i18n
 
@@ -101,9 +103,23 @@ class TaskExecutor(glance.async.TaskExecutor):
             return driver.DriverManager('glance.flows', task.type,
                                         invoke_on_load=True,
                                         invoke_kwds=kwds).driver
+        except urllib.error.URLError as exc:
+            raise exception.ImportTaskError(message=exc.reason)
+        except exception.BadStoreUri as exc:
+            raise exception.ImportTaskError(message=exc.msg)
         except RuntimeError:
             raise NotImplementedError()
 
+    def begin_processing(self, task_id):
+        try:
+            super(TaskExecutor, self).begin_processing(task_id)
+        except exception.ImportTaskError as exc:
+            LOG.error(_LE('Failed to execute task %(task_id)s: %(exc)s') %
+                      {'task_id': task_id, 'exc': exc.msg})
+            task = self.task_repo.get(task_id)
+            task.fail(exc.msg)
+            self.task_repo.save(task)
+
     def _run(self, task_id, task_type):
         LOG.debug('Taskflow executor picked up the execution of task ID '
                   '%(task_id)s of task type '

glance/common/exception.py

@@ -352,6 +352,10 @@ class InvalidTaskStatusTransition(TaskException, Invalid):
                 " %(new_status)s is not allowed")
 
 
+class ImportTaskError(TaskException, Invalid):
+    message = _("An import task exception occurred")
+
+
 class DuplicateLocation(Duplicate):
     message = _("The location %(location)s already exists")
 

glance/tests/unit/v2/test_tasks_resource.py

@@ -366,6 +366,47 @@ class TestTasksController(test_utils.BaseTestCase):
                                task_live_time.seconds / 3600)
         self.assertEqual(CONF.task.task_time_to_live, task_live_time_hour)
 
+    def test_create_with_wrong_import_form(self):
+        request = unit_test_utils.get_fake_request()
+        wrong_import_from = [
+            "swift://cloud.foo/myaccount/mycontainer/path",
+            "file:///path",
+            "s3://accesskey:secretkey@s3.amazonaws.com/bucket/key-id",
+            "cinder://volume-id"
+        ]
+        executor_factory = self.gateway.get_task_executor_factory(
+            request.context)
+        task_repo = self.gateway.get_task_repo(request.context)
+
+        for import_from in wrong_import_from:
+            task = {
+                "type": "import",
+                "input": {
+                    "import_from": import_from,
+                    "import_from_format": "qcow2",
+                    "image_properties": {
+                        "disk_format": "qcow2",
+                        "container_format": "bare",
+                        "name": "test-task"
+                    }
+                }
+            }
+            new_task = self.controller.create(request, task=task)
+            task_executor = executor_factory.new_task_executor(request.context)
+            task_executor.begin_processing(new_task.task_id)
+            final_task = task_repo.get(new_task.task_id)
+
+            self.assertEqual('failure', final_task.status)
+            if import_from.startswith("file:///"):
+                msg = ("File based imports are not allowed. Please use a "
+                       "non-local source of image data.")
+            else:
+                supported = ['http', ]
+                msg = ("The given uri is not valid. Please specify a "
+                       "valid uri from the following list of supported uri "
+                       "%(supported)s") % {'supported': supported}
+            self.assertEqual(msg, final_task.message)
+
     @mock.patch.object(glance.gateway.Gateway, 'get_task_factory')
     def test_notifications_on_create(self, mock_get_task_factory):
         request = unit_test_utils.get_fake_request()