def test_subworkflow_environment_inheritance(self):
        env = {'key1': 'abc'}

        wf2_ex = self.engine.start_workflow('wb1.wf2', env=env)

        # Execution of 'wf2'.
        self.assertIsNotNone(wf2_ex)
        self.assertDictEqual({}, wf2_ex.input)
        self.assertDictEqual(
            {'env': env, 'namespace': ''},
            wf2_ex.params
        )

        self._await(lambda: len(db_api.get_workflow_executions()) == 2, 0.5, 5)

        wf_execs = db_api.get_workflow_executions()

        self.assertEqual(2, len(wf_execs))

        # Execution of 'wf1'.
        wf1_ex = self._assert_single_item(wf_execs, name='wb1.wf1')
        wf2_ex = self._assert_single_item(wf_execs, name='wb1.wf2')

        self.assertIsNotNone(wf1_ex.task_execution_id)
        self.assertDictContainsSubset({}, wf1_ex.params)

        # Wait till workflow 'wf1' is completed.
        self.await_workflow_success(wf1_ex.id)

        # Wait till workflow 'wf2' is completed.
        self.await_workflow_success(wf2_ex.id)

    def test_rerun_sub_workflow(self):
        wf_service.create_workflows("""---
        version: '2.0'
        wf1:
          tasks:
            task1:
              workflow: wf2
        wf2:
          tasks:
            task2:
              workflow: wf3
        wf3:
          tasks:
            task3:
              action: std.noop""")

        # Run workflow and fail task.
        wf1_ex = self.engine.start_workflow('wf1')

        self.await_workflow_error(wf1_ex.id)

        with db_api.transaction():
            wf_exs = db_api.get_workflow_executions()
            task_exs = db_api.get_task_executions()

            self.assertEqual(3, len(wf_exs),
                             'The number of workflow executions')
            self.assertEqual(3, len(task_exs),
                             'The number of task executions')

            for wf_ex in wf_exs:
                self.assertEqual(states.ERROR, wf_ex.state,
                                 'The executions must fail the first time')
            for task_ex in task_exs:
                self.assertEqual(states.ERROR, task_ex.state,
                                 'The tasks must fail the first time')

            wf3_ex = self._assert_single_item(wf_exs, name='wf3')
            task3_ex = self._assert_single_item(wf3_ex.task_executions,
                                                name="task3")

        self.engine.rerun_workflow(task3_ex.id)

        self.await_workflow_success(wf1_ex.id)

        with db_api.transaction():
            wf_exs = db_api.get_workflow_executions()
            task_exs = db_api.get_task_executions()

            self.assertEqual(3, len(wf_exs),
                             'The number of workflow executions')
            self.assertEqual(3, len(task_exs),
                             'The number of task executions')

            for wf_ex in wf_exs:
                self.assertEqual(states.SUCCESS, wf_ex.state,
                                 'The executions must success the second time')
            for task_ex in task_exs:
                self.assertEqual(states.SUCCESS, task_ex.state,
                                 'The tasks must success the second time')

    def test_subworkflow_environment_inheritance(self):
        env = {'key1': 'abc'}

        wf2_ex = self.engine.start_workflow('my_wb.wf2', None, env=env)

        # Execution of 'wf2'.
        self.assertIsNotNone(wf2_ex)
        self.assertDictEqual({}, wf2_ex.input)
        self.assertDictEqual({'env': env}, wf2_ex.params)

        self._await(lambda: len(db_api.get_workflow_executions()) == 2, 0.5, 5)

        wf_execs = db_api.get_workflow_executions()

        self.assertEqual(2, len(wf_execs))

        # Execution of 'wf1'.
        wf1_ex = self._assert_single_item(wf_execs, name='my_wb.wf1')
        wf2_ex = self._assert_single_item(wf_execs, name='my_wb.wf2')

        expected_start_params = {
            'task_name': 'task2',
            'task_execution_id': wf1_ex.task_execution_id,
            'env': env
        }

        self.assertIsNotNone(wf1_ex.task_execution_id)
        self.assertDictContainsSubset(expected_start_params, wf1_ex.params)

        # Wait till workflow 'wf1' is completed.
        self._await(lambda: self.is_execution_success(wf1_ex.id))

        # Wait till workflow 'wf2' is completed.
        self._await(lambda: self.is_execution_success(wf2_ex.id))

    def test_subworkflow_root_execution_id(self):
        self.engine.start_workflow('wb6.wf1')

        self._await(lambda: len(db_api.get_workflow_executions()) == 3, 0.5, 5)

        wf_execs = db_api.get_workflow_executions()

        wf1_ex = self._assert_single_item(wf_execs, name='wb6.wf1')
        wf2_ex = self._assert_single_item(wf_execs, name='wb6.wf2')
        wf3_ex = self._assert_single_item(wf_execs, name='wb6.wf3')

        self.assertEqual(3, len(wf_execs))

        # Wait till workflow 'wf1' is completed (and all the sub-workflows
        # will be completed also).
        self.await_workflow_success(wf1_ex.id)

        with db_api.transaction():
            wf1_ex = db_api.get_workflow_execution(wf1_ex.id)
            wf2_ex = db_api.get_workflow_execution(wf2_ex.id)
            wf3_ex = db_api.get_workflow_execution(wf3_ex.id)

            self.assertIsNone(wf1_ex.root_execution_id, None)
            self.assertEqual(wf2_ex.root_execution_id, wf1_ex.id)
            self.assertEqual(wf2_ex.root_execution, wf1_ex)
            self.assertEqual(wf3_ex.root_execution_id, wf1_ex.id)
            self.assertEqual(wf3_ex.root_execution, wf1_ex)

    def test_cancel_child_workflow(self):
        workbook = """
        version: '2.0'

        name: wb

        workflows:
            wf:
              type: direct
              tasks:
                taskx:
                  workflow: subwf

            subwf:
              type: direct
              tasks:
                task1:
                  action: std.echo output="Echo"
                  on-complete:
                    - task2

                task2:
                  action: std.echo output="foo"
                  wait-before: 3
        """

        wb_service.create_workbook_v2(workbook)

        self.engine.start_workflow('wb.wf', {})

        wf_execs = db_api.get_workflow_executions()

        wf_ex = self._assert_single_item(wf_execs, name='wb.wf')
        task_ex = self._assert_single_item(wf_ex.task_executions, name='taskx')
        subwf_ex = self._assert_single_item(wf_execs, name='wb.subwf')

        self.engine.stop_workflow(
            subwf_ex.id,
            states.CANCELLED,
            "Cancelled by user."
        )

        self.await_workflow_cancelled(subwf_ex.id)
        self.await_task_cancelled(task_ex.id)
        self.await_workflow_cancelled(wf_ex.id)

        wf_execs = db_api.get_workflow_executions()

        wf_ex = self._assert_single_item(wf_execs, name='wb.wf')
        task_ex = self._assert_single_item(wf_ex.task_executions, name='taskx')
        subwf_ex = self._assert_single_item(wf_execs, name='wb.subwf')

        self.assertEqual(states.CANCELLED, subwf_ex.state)
        self.assertEqual("Cancelled by user.", subwf_ex.state_info)
        self.assertEqual(states.CANCELLED, task_ex.state)
        self.assertIn("Cancelled by user.", task_ex.state_info)
        self.assertEqual(states.CANCELLED, wf_ex.state)
        self.assertEqual("Cancelled tasks: taskx", wf_ex.state_info)

    def test_invalid_workflow_input(self):
        # Check that in case of invalid input workflow objects aren't even
        # created.
        wf_text = """
        version: '2.0'

        wf:
          input:
            - param1
            - param2

          tasks:
            task1:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

        self.assertRaises(
            exc.InputException,
            self.engine.start_workflow,
            'wf',
            '',
            {'wrong_param': 'some_value'}
        )

        self.assertEqual(0, len(db_api.get_workflow_executions()))
        self.assertEqual(0, len(db_api.get_task_executions()))
        self.assertEqual(0, len(db_api.get_action_executions()))

    def test_start_workflow_with_ex_id(self):
        wf_input = {'param1': 'Hey1', 'param2': 'Hi1'}
        the_ex_id = 'theId'

        # Start workflow.
        wf_ex = self.engine.start_workflow(
            'wb.wf',
            wf_input=wf_input,
            description='my execution',
            task_name='task2',
            wf_ex_id=the_ex_id
        )

        self.assertEqual(the_ex_id, wf_ex.id)

        wf_ex_2 = self.engine.start_workflow(
            'wb.wf',
            wf_input={'param1': 'Hey2', 'param2': 'Hi2'},
            wf_ex_id=the_ex_id
        )

        self.assertDictEqual(dict(wf_ex), dict(wf_ex_2))

        wf_executions = db_api.get_workflow_executions()

        self.assertEqual(1, len(wf_executions))

    def test_cancel_with_items_concurrency(self):
        wb_def = """
            version: '2.0'

            name: wb1

            workflows:
              wf1:
                tasks:
                  t1:
                    with-items: i in <% list(range(0, 4)) %>
                    action: std.async_noop
                    concurrency: 2
                    on-success:
                      - t2
                  t2:
                    action: std.echo output="Task 2"
        """

        wb_service.create_workbook_v2(wb_def)

        wf1_ex = self.engine.start_workflow('wb1.wf1')

        self.await_workflow_state(wf1_ex.id, states.RUNNING)

        with db_api.transaction():
            wf1_execs = db_api.get_workflow_executions()

            wf1_ex = self._assert_single_item(wf1_execs, name='wb1.wf1')
            wf1_t1_ex = self._assert_single_item(
                wf1_ex.task_executions,
                name='t1'
            )

        wf1_t1_action_exs = db_api.get_action_executions(
            task_execution_id=wf1_t1_ex.id
        )

        self.assertEqual(2, len(wf1_t1_action_exs))
        self.assertEqual(states.RUNNING, wf1_t1_action_exs[0].state)
        self.assertEqual(states.RUNNING, wf1_t1_action_exs[1].state)

        # Cancel action execution for task.
        for wf1_t1_action_ex in wf1_t1_action_exs:
            self.engine.on_action_complete(
                wf1_t1_action_ex.id,
                ml_actions.Result(cancel=True)
            )

        self.await_task_cancelled(wf1_t1_ex.id)
        self.await_workflow_cancelled(wf1_ex.id)

        wf1_t1_action_exs = db_api.get_action_executions(
            task_execution_id=wf1_t1_ex.id
        )

        self.assertEqual(2, len(wf1_t1_action_exs))
        self.assertEqual(states.CANCELLED, wf1_t1_action_exs[0].state)
        self.assertEqual(states.CANCELLED, wf1_t1_action_exs[1].state)

    def test_subworkflow_error(self):
        self.engine.start_workflow('wb1.wf2')

        self._await(lambda: len(db_api.get_workflow_executions()) == 2, 0.5, 5)

        wf_execs = db_api.get_workflow_executions()

        self.assertEqual(2, len(wf_execs))

        wf1_ex = self._assert_single_item(wf_execs, name='wb1.wf1')
        wf2_ex = self._assert_single_item(wf_execs, name='wb1.wf2')

        # Wait till workflow 'wf1' is completed.
        self.await_workflow_error(wf1_ex.id)

        # Wait till workflow 'wf2' is completed, its state must be ERROR.
        self.await_workflow_error(wf2_ex.id)

    def print_executions(exc_info=None):
        if exc_info:
            print("\nEngine test case exception occurred: %s" % exc_info[1])
            print("Exception type: %s" % exc_info[0])

        print("\nPrinting workflow executions...")

        with db_api.transaction():
            wf_execs = db_api.get_workflow_executions()

            for w in wf_execs:
                print(
                    "\n%s (%s) [state=%s, state_info=%s, output=%s]" %
                    (w.name, w.id, w.state, w.state_info, w.output)
                )

                for t in w.task_executions:
                    print(
                        "\t%s [id=%s, state=%s, state_info=%s, processed=%s,"
                        " published=%s, runtime_context=%s]" %
                        (t.name,
                         t.id,
                         t.state,
                         t.state_info,
                         t.processed,
                         t.published,
                         t.runtime_context)
                    )

                    child_execs = t.executions

                    for a in child_execs:
                        print(
                            "\t\t%s [id=%s, state=%s, state_info=%s,"
                            " accepted=%s, output=%s]" %
                            (a.name,
                             a.id,
                             a.state,
                             a.state_info,
                             a.accepted,
                             a.output)
                        )

            print("\nPrinting standalone action executions...")

            child_execs = db_api.get_action_executions(task_execution_id=None)

            for a in child_execs:
                print(
                    "\t\t%s [id=%s, state=%s, state_info=%s, accepted=%s,"
                    " output=%s]" %
                    (a.name,
                     a.id,
                     a.state,
                     a.state_info,
                     a.accepted,
                     a.output)
                )

    def test_read_only_transactions(self):
        with db_api.transaction():
            db_api.create_workflow_execution(WF_EXECS[0])

            wf_execs = db_api.get_workflow_executions()
            self.assertEqual(1, len(wf_execs))

        wf_execs = db_api.get_workflow_executions()
        self.assertEqual(1, len(wf_execs))

        with db_api.transaction(read_only=True):
            db_api.create_workflow_execution(WF_EXECS[1])

            wf_execs = db_api.get_workflow_executions()
            self.assertEqual(2, len(wf_execs))

        wf_execs = db_api.get_workflow_executions()
        self.assertEqual(1, len(wf_execs))

    def test_subworkflow_env_no_duplicate(self):
        wf_text = """---
        version: '2.0'

        parent_wf:
          tasks:
            task1:
              workflow: sub_wf

        sub_wf:
          output:
            result: <% $.result %>

          tasks:
            task1:
              action: std.noop
              publish:
                result: <% env().param1 %>
        """

        wf_service.create_workflows(wf_text)

        env = {
            'param1': 'val1',
            'param2': 'val2',
            'param3': 'val3'
        }

        parent_wf_ex = self.engine.start_workflow('parent_wf', env=env)

        self.await_workflow_success(parent_wf_ex.id)

        with db_api.transaction():
            parent_wf_ex = db_api.get_workflow_execution(parent_wf_ex.id)

            t = self._assert_single_item(
                parent_wf_ex.task_executions,
                name='task1'
            )

            sub_wf_ex = db_api.get_workflow_executions(
                task_execution_id=t.id
            )[0]

            self.assertDictEqual(
                {
                    "result": "val1"
                },
                sub_wf_ex.output
            )

        # The environment of the subworkflow must be empty.
        # To evaluate expressions it should be taken from the
        # parent workflow execution.
        self.assertDictEqual({}, sub_wf_ex.params['env'])
        self.assertNotIn('__env', sub_wf_ex.context)

    def get_all(self):
        """Return all Executions."""
        LOG.info("Fetch executions")

        wf_executions = [
            Execution.from_dict(db_model.to_dict())
            for db_model in db_api.get_workflow_executions()
        ]

        return Executions(executions=wf_executions)

    def test_cache_workflow_spec_no_duplicates(self):
        wfs_text = """
        version: '2.0'

        wf:
          tasks:
            task1:
              action: std.noop
              on-success:
                - task2
                - task3

            task2:
              workflow: sub_wf my_param="val1"

            task3:
              workflow: sub_wf my_param="val2"

        sub_wf:
          input:
            - my_param

          tasks:
            task1:
              action: std.echo output="Param value is <% $.my_param %>"
        """

        wfs = wf_service.create_workflows(wfs_text)

        self.assertEqual(2, len(wfs))

        self.assertEqual(0, spec_parser.get_wf_execution_spec_cache_size())
        self.assertEqual(0, spec_parser.get_wf_definition_spec_cache_size())

        wf_ex = self.engine.start_workflow('wf')

        self.await_workflow_success(wf_ex.id)

        # We expect to have a cache entry for every workflow execution
        # but two of them should refer to the same object.
        self.assertEqual(3, spec_parser.get_wf_execution_spec_cache_size())
        self.assertEqual(2, spec_parser.get_wf_definition_spec_cache_size())

        sub_wf_execs = db_api.get_workflow_executions(name='sub_wf')

        self.assertEqual(2, len(sub_wf_execs))

        spec1 = spec_parser.get_workflow_spec_by_execution_id(
            sub_wf_execs[0].id
        )
        spec2 = spec_parser.get_workflow_spec_by_execution_id(
            sub_wf_execs[1].id
        )

        self.assertIs(spec1, spec2)

    def test_cascade_delete(self):
        wf_text = """
        version: 2.0

        wf:
          tasks:
            task1:
              workflow: sub_wf1

            task2:
              workflow: sub_wf2

        sub_wf1:
          tasks:
            task1:
              action: std.noop

        sub_wf2:
          tasks:
            task1:
              action: std.noop
        """

        wf_service.create_workflows(wf_text)

        wf_ex = self.engine.start_workflow('wf')

        self.await_workflow_success(wf_ex.id)

        self.assertEqual(3, len(db_api.get_workflow_executions()))
        self.assertEqual(4, len(db_api.get_task_executions()))
        self.assertEqual(2, len(db_api.get_action_executions()))

        # Now delete the root workflow execution and make sure that
        # all dependent objects are deleted as well.
        db_api.delete_workflow_execution(wf_ex.id)

        self.assertEqual(0, len(db_api.get_workflow_executions()))
        self.assertEqual(0, len(db_api.get_task_executions()))
        self.assertEqual(0, len(db_api.get_action_executions()))

def _get_tasks_from_db(workflow_execution_id=None, recursive=False, state=None,
                       flat=False):
    task_execs = []
    nested_task_exs = []

    kwargs = {}

    if workflow_execution_id:
        kwargs['workflow_execution_id'] = workflow_execution_id

    # We can't add state to query if we want to filter by workflow_execution_id
    # recursively. There might be a workflow_execution in one state with a
    # nested workflow execution that has a task in the desired state until we
    # have an optimization for queering all workflow executions under a given
    # top level workflow execution, this is the way to go.
    if state and not (workflow_execution_id and recursive):
        kwargs['state'] = state

    task_execs.extend(db_api.get_task_executions(**kwargs))

    # If it is not recursive no need to check nested workflows.
    # If there is no workflow execution id, we already have all we need, and
    # doing more queries will just create duplication in the results.
    if recursive and workflow_execution_id:
        for t in task_execs:
            if t.type == utils.WORKFLOW_TASK_TYPE:
                # Get nested workflow execution that matches the task.
                nested_workflow_executions = db_api.get_workflow_executions(
                    task_execution_id=t.id
                )

                # There might be zero nested executions.
                for nested_workflow_execution in nested_workflow_executions:
                    nested_task_exs.extend(
                        _get_tasks_from_db(
                            nested_workflow_execution.id,
                            recursive,
                            state,
                            flat
                        )
                    )

    if state or flat:
        # Filter by state and flat.
        task_execs = [
            t for t in task_execs if _should_pass_filter(t, state, flat)
        ]

    # The nested tasks were already filtered, since this is a recursion.
    task_execs.extend(nested_task_exs)

    return task_execs

def executions_(context,
                id=None,
                root_execution_id=None,
                state=None,
                from_time=None,
                to_time=None
                ):

    filter = {}

    if id is not None:
        filter = filter_utils.create_or_update_filter(
            'id',
            id,
            "eq",
            filter
        )

    if root_execution_id is not None:
        filter = filter_utils.create_or_update_filter(
            'root_execution_id',
            root_execution_id,
            "eq",
            filter
        )

    if state is not None:
        filter = filter_utils.create_or_update_filter(
            'state',
            state,
            "eq",
            filter
        )

    if from_time is not None:
        filter = filter_utils.create_or_update_filter(
            'created_at',
            from_time,
            "gte",
            filter
        )

    if to_time is not None:
        filter = filter_utils.create_or_update_filter(
            'created_at',
            to_time,
            "lt",
            filter
        )

    return db_api.get_workflow_executions(**filter)

def pause_workflow(wf_ex, msg=None):
    # Pause subworkflows first.
    for task_ex in wf_ex.task_executions:
        sub_wf_exs = db_api.get_workflow_executions(
            task_execution_id=task_ex.id
        )

        for sub_wf_ex in sub_wf_exs:
            if not states.is_completed(sub_wf_ex.state):
                pause_workflow(sub_wf_ex, msg=msg)

    # If all subworkflows paused successfully, pause the main workflow.
    # If any subworkflows failed to pause for temporary reason, this
    # allows pause to be executed again on the main workflow.
    wf = workflows.Workflow(wf_ex=wf_ex)
    wf.pause(msg=msg)

    def get_all(self, marker=None, limit=None, sort_keys='created_at',
                sort_dirs='asc'):
        """Return all Executions.

        :param marker: Optional. Pagination marker for large data sets.
        :param limit: Optional. Maximum number of resources to return in a
                      single result. Default value is None for backward
                      compatibility.
        :param sort_keys: Optional. Columns to sort results by.
                          Default: created_at, which is backward compatible.
        :param sort_dirs: Optional. Directions to sort corresponding to
                          sort_keys, "asc" or "desc" can be chosen.
                          Default: desc. The length of sort_dirs can be equal
                          or less than that of sort_keys.
        """
        LOG.info(
            "Fetch executions. marker=%s, limit=%s, sort_keys=%s, "
            "sort_dirs=%s", marker, limit, sort_keys, sort_dirs
        )

        rest_utils.validate_query_params(limit, sort_keys, sort_dirs)

        marker_obj = None

        if marker:
            marker_obj = db_api.get_workflow_execution(marker)

        db_workflow_exs = db_api.get_workflow_executions(
            limit=limit,
            marker=marker_obj,
            sort_keys=sort_keys,
            sort_dirs=sort_dirs
        )

        wf_executions = [
            Execution.from_dict(db_model.to_dict())
            for db_model in db_workflow_exs
        ]

        return Executions.convert_with_links(
            wf_executions,
            limit,
            pecan.request.host_url,
            sort_keys=','.join(sort_keys),
            sort_dirs=','.join(sort_dirs)
        )

    def test_subworkflow_yaql_error(self):
        wf_ex = self.engine.start_workflow('wb2.wf1', None)

        self.await_execution_error(wf_ex.id)

        wf_execs = db_api.get_workflow_executions()

        self.assertEqual(2, len(wf_execs))

        wf2_ex = self._assert_single_item(wf_execs, name='wb2.wf2')
        self.assertEqual(states.ERROR, wf2_ex.state)
        self.assertIn('Can not evaluate YAQL expression', wf2_ex.state_info)

        # Ensure error message is bubbled up to the main workflow.
        wf1_ex = self._assert_single_item(wf_execs, name='wb2.wf1')
        self.assertEqual(states.ERROR, wf1_ex.state)
        self.assertIn('Can not evaluate YAQL expression', wf1_ex.state_info)