def test_adherence_to_output_schema(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential_with_schema.yaml')
        wf_input = {'who': 'Thanos'}
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
        wf_ex_db = wf_ex_dbs[0]
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
        tk1_ex_db = wf_db_access.TaskExecution.get_by_id(tk1_ex_db.id)
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task2'}
        tk2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk2_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)
        tk2_ex_db = wf_db_access.TaskExecution.get_by_id(tk2_ex_db.id)
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task3'}
        tk3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk3_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(tk3_ac_ex_db)
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))

        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

    def test_with_items_empty_list(self):
        items = []
        num_items = len(items)
        wf_input = {'members': items}

        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'with-items.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
        lv_ac_db, ac_ex_db = action_service.request(lv_ac_db)

        # Wait for the liveaction to complete.
        lv_ac_db = self._wait_on_status(lv_ac_db, action_constants.LIVEACTION_STATUS_SUCCEEDED)

        # Retrieve records from database.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
        t1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        t1_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))

        # Ensure there is no action executions for the task and the task is already completed.
        self.assertEqual(len(t1_ac_ex_dbs), num_items)
        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
        self.assertDictEqual(t1_ex_db.result, {'items': []})

        # Assert the main workflow is completed.
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
        self.assertDictEqual(lv_ac_db.result, {'output': {'items': []}})

    def test_output_on_error(self):
        expected_output = {
            'progress': 25
        }

        expected_errors = [
            {
                'type': 'error',
                'task_id': 'task2',
                'message': 'Execution failed. See result for details.',
                'result': {
                    'failed': True,
                    'return_code': 1,
                    'stderr': '',
                    'stdout': '',
                    'succeeded': False
                }
            }
        ]

        expected_result = {
            'errors': expected_errors,
            'output': expected_output
        }

        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'output-on-error.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]

        # Assert task1 is already completed and workflow execution is still running.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, wf_statuses.RUNNING)

        # Assert task2 is already completed and workflow execution has failed.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task2'}
        tk2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk2_ex_db.id))[0]
        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction['id'])
        self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)

        # Check output and result for expected value(s).
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, wf_statuses.FAILED)
        self.assertDictEqual(wf_ex_db.output, expected_output)

        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(lv_ac_db.result, expected_result)

        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_notify_task_list_bad_item_value(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db.notify = notify_api_models.NotificationsHelper.to_model(MOCK_NOTIFY)

        expected_schema_failure_test_cases = [
            'task1',            # Notify must be type of list.
            [123],              # Item has to be type of string.
            [''],               # String value cannot be empty.
            ['  '],             # String value cannot be just spaces.
            ['      '],         # String value cannot be just tabs.
            ['init task'],      # String value cannot have space.
            ['init-task'],      # String value cannot have dash.
            ['task1', 'task1']  # String values have to be unique.
        ]

        for notify_tasks in expected_schema_failure_test_cases:
            lv_ac_db.parameters = {'notify': notify_tasks}

            try:
                self.assertRaises(
                    jsonschema.ValidationError,
                    action_service.request,
                    lv_ac_db
                )
            except Exception as e:
                raise AssertionError('%s: %s' % (six.text_type(e), notify_tasks))

    def test_fail_start_task_action(self):
        expected_errors = [
            {
                'type': 'error',
                'message': (
                    'YaqlEvaluationException: Unable to evaluate expression '
                    '\'<% ctx().func.value %>\'. NoFunctionRegisteredException: '
                    'Unknown function "#property#value"'
                ),
                'task_id': 'task1',
                'route': 0
            }
        ]

        expected_result = {'output': None, 'errors': expected_errors}

        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'fail-start-task-action.yaml')

        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)

        # Assert action execution for task is not started and workflow failed.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))
        self.assertEqual(len(tk_ex_dbs), 0)
        self.assertEqual(wf_ex_db.status, wf_statuses.FAILED)
        self.assertListEqual(self.sort_wf_runtime_errors(wf_ex_db.errors), expected_errors)

        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(lv_ac_db.result, expected_result)

        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_cancel_workflow_cascade_down_to_subworkflow(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'subworkflow.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        # Identify the records for the subworkflow.
        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
        self.assertEqual(len(wf_ex_dbs), 1)

        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_dbs[0].id))
        self.assertEqual(len(tk_ex_dbs), 1)

        tk_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_dbs[0].id))
        self.assertEqual(len(tk_ac_ex_dbs), 1)

        tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk_ac_ex_dbs[0].liveaction['id'])
        self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        # Cancel the main workflow.
        requester = cfg.CONF.system_user.user
        lv_ac_db, ac_ex_db = ac_svc.request_cancellation(lv_ac_db, requester)
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELING)

        # Assert the subworkflow is canceled.
        tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk_lv_ac_db.id))
        self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELED)

        # Assert the main workflow is canceled.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELED)

    def test_retry_policy_applied_on_workflow_failure(self):
        wf_name = 'sequential'
        wf_ac_ref = TEST_PACK + '.' + wf_name
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_name + '.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        # Ensure there is only one execution recorded.
        self.assertEqual(len(lv_db_access.LiveAction.query(action=wf_ac_ref)), 1)

        # Identify the records for the workflow and task.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        t1_ex_db = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))[0]
        t1_lv_ac_db = lv_db_access.LiveAction.query(task_execution=str(t1_ex_db.id))[0]
        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))[0]

        # Manually set the status to fail.
        ac_svc.update_status(t1_lv_ac_db, ac_const.LIVEACTION_STATUS_FAILED)
        t1_lv_ac_db = lv_db_access.LiveAction.query(task_execution=str(t1_ex_db.id))[0]
        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))[0]
        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        notifier.get_notifier().process(t1_ac_ex_db)
        workflows.get_engine().process(t1_ac_ex_db)

        # Assert the main workflow is completed.
        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        notifier.get_notifier().process(ac_ex_db)

        # Ensure execution is retried.
        self.assertEqual(len(lv_db_access.LiveAction.query(action=wf_ac_ref)), 2)

    def test_run_workflow_action_config_context(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'config-context.yaml')
        wf_input = {}
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)

        # Assert action execution is running.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        self.assertEqual(wf_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        # Assert task1 is already completed.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk1_ac_ex_db))

        # Manually handle action execution completion.
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)

        # Assert workflow is completed.
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

        # Verify config_context works
        self.assertEqual(wf_ex_db.output, {'msg': 'value of config key a'})

    def test_fail_start_task_input_expr_eval(self):
        expected_errors = [
            {
                'message': 'Unknown function "#property#value"',
                'task_id': 'task1'
            }
        ]

        expected_result = {'output': None, 'errors': expected_errors}

        wf_file = 'fail-start-task-input-expr-eval.yaml'
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_file)

        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)

        # Assert action execution for task is not started and workflow failed.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))
        self.assertEqual(len(tk_ex_dbs), 0)
        self.assertEqual(wf_ex_db.status, wf_states.FAILED)
        self.assertListEqual(self.sort_wf_runtime_errors(wf_ex_db.errors), expected_errors)

        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(lv_ac_db.result, expected_result)

        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_pause_workflow_cascade_down_to_subworkflow(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'subworkflow.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        # Identify the records for the main workflow.
        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
        self.assertEqual(len(wf_ex_dbs), 1)

        wf_ex_db = wf_ex_dbs[0]
        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))
        self.assertEqual(len(tk_ex_dbs), 1)

        tk_ex_db = tk_ex_dbs[0]
        tk_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_db.id))
        self.assertEqual(len(tk_ac_ex_dbs), 1)

        tk_ac_ex_db = tk_ac_ex_dbs[0]
        tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk_ac_ex_db.liveaction['id'])
        self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        # Identify the records for the subworkflow.
        sub_wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(tk_ac_ex_db.id))
        self.assertEqual(len(sub_wf_ex_dbs), 1)

        sub_wf_ex_db = sub_wf_ex_dbs[0]
        sub_tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(sub_wf_ex_db.id))
        self.assertEqual(len(sub_tk_ex_dbs), 1)

        sub_tk_ex_db = sub_tk_ex_dbs[0]
        sub_tk_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(sub_tk_ex_db.id))
        self.assertEqual(len(sub_tk_ac_ex_dbs), 1)

        # Pause the main workflow and assert it is pausing because subworkflow is still running.
        lv_ac_db, ac_ex_db = ac_svc.request_pause(lv_ac_db, cfg.CONF.system_user.user)
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_PAUSING)

        # Assert the subworkflow is pausing.
        tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk_lv_ac_db.id))
        self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_PAUSING)

        # Manually handle action execution completion for the task in the subworkflow.
        sub_tk_ac_ex_db = sub_tk_ac_ex_dbs[0]
        self.assertEqual(sub_tk_ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        workflows.get_engine().process(sub_tk_ac_ex_db)

        # Assert the subworkflow is paused and manually process the execution update.
        tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk_lv_ac_db.id))
        self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_PAUSED)
        tk_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(tk_ac_ex_db.id))
        self.assertEqual(tk_ac_ex_db.status, ac_const.LIVEACTION_STATUS_PAUSED)
        workflows.get_engine().process(tk_ac_ex_db)

        # Assert the main workflow is paused.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_PAUSED)

    def test_action_context_source_channel(self):
        wf_name = 'subworkflow-source-channel-from-action-context'
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_name + '.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'],
                                             context={'source_channel': 'general'})
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        # Identify the records for the main workflow.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        t1_ex_db = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))[0]
        t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_ex_db.id))[0]
        t1_wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(t1_ac_ex_db.id))[0]
        self.assertEqual(t1_ex_db.status, wf_statuses.RUNNING)
        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)
        self.assertEqual(t1_wf_ex_db.status, wf_statuses.RUNNING)

        # Complete subworkflow under task1.
        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task1'}
        t1_t1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        t1_t1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t1_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(t1_t1_ac_ex_db)

        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task2'}
        t1_t2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        t1_t2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t2_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(t1_t2_ac_ex_db)

        query_filters = {'workflow_execution': str(t1_wf_ex_db.id), 'task_id': 'task3'}
        t1_t3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        t1_t3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(t1_t3_ex_db.id))[0]
        wf_svc.handle_action_execution_completion(t1_t3_ac_ex_db)

        t1_wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(t1_wf_ex_db.id))
        t1_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(t1_ac_ex_db.id))
        self.assertEqual(t1_wf_ex_db.status, wf_statuses.SUCCEEDED)
        self.assertEqual(t1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

        # Complete task1 and main workflow.
        wf_svc.handle_action_execution_completion(t1_ac_ex_db)
        t1_ex_db = wf_db_access.TaskExecution.get_by_id(str(t1_ex_db.id))
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(t1_ex_db.status, wf_statuses.SUCCEEDED)
        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

        # Check result.
        expected_result = {
            'output': {
                'msg': 'general, All your base are belong to us!'
            }
        }

        self.assertDictEqual(lv_ac_db.result, expected_result)

    def test_pause_with_active_children(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        lv_ac_db, ac_ex_db = ac_svc.request_pause(lv_ac_db, cfg.CONF.system_user.user)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_PAUSING)

    def test_cancel_subworkflow_cascade_up_to_workflow_with_other_subworkflows(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'subworkflows.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        # Identify the records for the subworkflow.
        wf_ex_dbs = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))
        self.assertEqual(len(wf_ex_dbs), 1)

        tk_ex_dbs = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_dbs[0].id))
        self.assertEqual(len(tk_ex_dbs), 2)

        tk1_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_dbs[0].id))
        self.assertEqual(len(tk1_ac_ex_dbs), 1)

        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_dbs[0].liveaction['id'])
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        tk2_ac_ex_dbs = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_dbs[1].id))
        self.assertEqual(len(tk2_ac_ex_dbs), 1)

        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_dbs[0].liveaction['id'])
        self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        # Cancel the subworkflow which should cascade up to the root.
        requester = cfg.CONF.system_user.user
        tk1_lv_ac_db, tk1_ac_ex_db = ac_svc.request_cancellation(tk1_lv_ac_db, requester)
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELING)

        # Assert the main workflow is canceling.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELING)

        # Assert both subworkflows are canceled.
        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk1_lv_ac_db.id))
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELED)
        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(str(tk2_lv_ac_db.id))
        self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELED)

        # Manually handle action execution completion for one of the tasks.
        tk1_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(tk1_ac_ex_db.id))
        self.assertEqual(tk1_ac_ex_db.status, ac_const.LIVEACTION_STATUS_CANCELED)
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)

        # Manually handle action execution completion for the other task.
        tk2_ac_ex_db = tk2_ac_ex_dbs[0]
        tk2_ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(tk2_ac_ex_db.id))
        self.assertEqual(tk2_ac_ex_db.status, ac_const.LIVEACTION_STATUS_CANCELED)
        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)

        # Assert the main workflow is canceling.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELED)

    def test_run_workflow_with_unicode_input(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential.yaml')
        wf_input = {'who': '薩諾斯'}
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]

        # Process task1.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task1'}
        tk1_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)
        tk1_ex_db = wf_db_access.TaskExecution.get_by_id(tk1_ex_db.id)
        self.assertEqual(tk1_ex_db.status, wf_statuses.SUCCEEDED)

        # Process task2.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task2'}
        tk2_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk2_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk2_ex_db.id))[0]
        tk2_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk2_ac_ex_db.liveaction['id'])
        self.assertEqual(tk2_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        wf_svc.handle_action_execution_completion(tk2_ac_ex_db)
        tk2_ex_db = wf_db_access.TaskExecution.get_by_id(tk2_ex_db.id)
        self.assertEqual(tk2_ex_db.status, wf_statuses.SUCCEEDED)

        # Process task3.
        query_filters = {'workflow_execution': str(wf_ex_db.id), 'task_id': 'task3'}
        tk3_ex_db = wf_db_access.TaskExecution.query(**query_filters)[0]
        tk3_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk3_ex_db.id))[0]
        tk3_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk3_ac_ex_db.liveaction['id'])
        self.assertEqual(tk3_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        wf_svc.handle_action_execution_completion(tk3_ac_ex_db)
        tk3_ex_db = wf_db_access.TaskExecution.get_by_id(tk3_ex_db.id)
        self.assertEqual(tk3_ex_db.status, wf_statuses.SUCCEEDED)

        # Assert workflow is completed.
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, wf_statuses.SUCCEEDED)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)

        # Check workflow output.
        wf_input_val = wf_input['who'].decode('utf-8') if six.PY2 else wf_input['who']
        expected_output = {'msg': '%s, All your base are belong to us!' % wf_input_val}
        self.assertDictEqual(wf_ex_db.output, expected_output)

        # Check liveaction and action execution result.
        expected_result = {'output': expected_output}

        self.assertDictEqual(lv_ac_db.result, expected_result)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_cancel(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)

        requester = cfg.CONF.system_user.user
        lv_ac_db, ac_ex_db = ac_svc.request_cancellation(lv_ac_db, requester)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_CANCELING)

    def test_fail_next_task_input_value_type(self):
        if six.PY3:
            msg = 'Value "{\'x\': \'foobar\'}" must either be a string or None. Got "dict".'
        else:
            msg = 'Value "{u\'x\': u\'foobar\'}" must either be a string or None. Got "dict".'

        msg = 'ValueError: ' + msg

        expected_errors = [
            {
                'type': 'error',
                'message': msg,
                'task_id': 'task2',
                'route': 0
            }
        ]

        expected_result = {'output': None, 'errors': expected_errors}

        wf_file = 'fail-task-input-value-type.yaml'
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_file)
        wf_input = {'var1': {'x': 'foobar'}}
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'], parameters=wf_input)
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)

        # Assert task1 is already completed and workflow execution is still running.
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        tk1_ex_db = wf_db_access.TaskExecution.query(workflow_execution=str(wf_ex_db.id))[0]
        tk1_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk1_ex_db.id))[0]
        tk1_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk1_ac_ex_db.liveaction['id'])
        self.assertEqual(tk1_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
        self.assertEqual(wf_ex_db.status, wf_statuses.RUNNING)

        # Manually handle action execution completion for task1 which has an error in publish.
        wf_svc.handle_action_execution_completion(tk1_ac_ex_db)

        # Assert workflow execution and task2 execution failed.
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(str(wf_ex_db.id))
        self.assertEqual(wf_ex_db.status, wf_statuses.FAILED)
        self.assertListEqual(self.sort_wf_runtime_errors(wf_ex_db.errors), expected_errors)

        tk2_ex_db = wf_db_access.TaskExecution.query(task_id='task2')[0]
        self.assertEqual(tk2_ex_db.status, wf_statuses.FAILED)
        self.assertDictEqual(tk2_ex_db.result, {'errors': expected_errors})

        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(lv_ac_db.result, expected_result)

        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_no_notify(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'sequential.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = action_service.request(lv_ac_db)

        # Assert action execution is running.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, action_constants.LIVEACTION_STATUS_RUNNING)
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        self.assertEqual(wf_ex_db.status, action_constants.LIVEACTION_STATUS_RUNNING)

        # Check that notify is setup correctly in the db record.
        self.assertDictEqual(wf_ex_db.notify, {})

    def _execute_workflow(self, wf_name, expected_task_sequence, expected_output,
                          expected_status=wf_statuses.SUCCEEDED, expected_errors=None):
        wf_file = wf_name + '.yaml'
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, wf_file)
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)

        # Assert action execution is running.
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_RUNNING, lv_ac_db.result)
        wf_ex_db = wf_db_access.WorkflowExecution.query(action_execution=str(ac_ex_db.id))[0]
        self.assertEqual(wf_ex_db.status, ac_const.LIVEACTION_STATUS_RUNNING)

        for task_id, route in expected_task_sequence:
            tk_ex_dbs = wf_db_access.TaskExecution.query(
                workflow_execution=str(wf_ex_db.id),
                task_id=task_id,
                task_route=route
            )

            if len(tk_ex_dbs) <= 0:
                break

            tk_ex_db = sorted(tk_ex_dbs, key=lambda x: x.start_timestamp)[len(tk_ex_dbs) - 1]
            tk_ac_ex_db = ex_db_access.ActionExecution.query(task_execution=str(tk_ex_db.id))[0]
            tk_lv_ac_db = lv_db_access.LiveAction.get_by_id(tk_ac_ex_db.liveaction['id'])

            self.assertEqual(tk_lv_ac_db.status, ac_const.LIVEACTION_STATUS_SUCCEEDED)
            self.assertTrue(wf_svc.is_action_execution_under_workflow_context(tk_ac_ex_db))

            wf_svc.handle_action_execution_completion(tk_ac_ex_db)

        # Assert workflow is completed.
        wf_ex_db = wf_db_access.WorkflowExecution.get_by_id(wf_ex_db.id)
        self.assertEqual(wf_ex_db.status, expected_status)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))
        self.assertEqual(lv_ac_db.status, expected_status)
        ac_ex_db = ex_db_access.ActionExecution.get_by_id(str(ac_ex_db.id))
        self.assertEqual(ac_ex_db.status, expected_status)

        # Check workflow output, liveaction result, and action execution result.
        expected_result = {'output': expected_output}

        if expected_errors is not None:
            expected_result['errors'] = expected_errors

        if expected_output is not None:
            self.assertDictEqual(wf_ex_db.output, expected_output)

        self.assertDictEqual(lv_ac_db.result, expected_result)
        self.assertDictEqual(ac_ex_db.result, expected_result)

    def test_fail_inspection(self):
        wf_meta = base.get_wf_fixture_meta_data(TEST_PACK_PATH, 'fail-inspection.yaml')
        lv_ac_db = lv_db_models.LiveActionDB(action=wf_meta['name'])
        lv_ac_db, ac_ex_db = ac_svc.request(lv_ac_db)
        lv_ac_db = lv_db_access.LiveAction.get_by_id(str(lv_ac_db.id))

        self.assertEqual(lv_ac_db.status, ac_const.LIVEACTION_STATUS_FAILED)
        self.assertIn('errors', lv_ac_db.result)
        self.assertIn('expressions', lv_ac_db.result['errors'])
        self.assertGreater(len(lv_ac_db.result['errors']['expressions']), 0)
        self.assertIn('context', lv_ac_db.result['errors'])
        self.assertGreater(len(lv_ac_db.result['errors']['context']), 0)
        self.assertIn('syntax', lv_ac_db.result['errors'])
        se