'Yields a dependency status that indicate whether the given task instance\'s trigger
rule was met.
:param ti: the task instance to evaluate the trigger rule of
:type ti: TaskInstance
:param successes: Number of successful upstream tasks
:type successes: boolean
:param skipped: Number of skipped upstream tasks
:type skipped: boolean
:param failed: Number of failed upstream tasks
:type failed: boolean
:param upstream_failed: Number of upstream_failed upstream tasks
:type upstream_failed: boolean
:param done: Number of completed upstream tasks
:type done: boolean
:param flag_upstream_failed: This is a hack to generate
the upstream_failed state creation while checking to see
whether the task instance is runnable. It was the shortest
path to add the feature
:type flag_upstream_failed: boolean
:param session: database session
:type session: Session'

@provide_session
def _evaluate_trigger_rule(self, ti, successes, skipped, failed, upstream_failed, done, flag_upstream_failed, session):

    TR = airflow.models.TriggerRule
    task = ti.task
    upstream = len(task.upstream_task_ids)
    tr = task.trigger_rule
    upstream_done = (done >= upstream)
    upstream_tasks_state = {'successes': successes, 'skipped': skipped, 'failed': failed, 'upstream_failed': upstream_failed, 'done': done}
    if flag_upstream_failed:
        if (tr == TR.ALL_SUCCESS):
            if (upstream_failed or failed):
                ti.set_state(State.UPSTREAM_FAILED, session)
            elif skipped:
                ti.set_state(State.SKIPPED, session)
        elif (tr == TR.ALL_FAILED):
            if (successes or skipped):
                ti.set_state(State.SKIPPED, session)
        elif (tr == TR.ONE_SUCCESS):
            if (upstream_done and (not successes)):
                ti.set_state(State.SKIPPED, session)
        elif (tr == TR.ONE_FAILED):
            if (upstream_done and (not (failed or upstream_failed))):
                ti.set_state(State.SKIPPED, session)
    if (tr == TR.ONE_SUCCESS):
        if (successes <= 0):
            (yield self._failing_status(reason="Task's    trigger    rule    '{0}'    requires    one    upstream    task    success,    but    none    were    found.    upstream_tasks_state={1},    upstream_task_ids={2}".format(tr, upstream_tasks_state, task.upstream_task_ids)))
    elif (tr == TR.ONE_FAILED):
        if ((not failed) and (not upstream_failed)):
            (yield self._failing_status(reason="Task's    trigger    rule    '{0}'    requires    one    upstream    task    failure,    but    none    were    found.    upstream_tasks_state={1},    upstream_task_ids={2}".format(tr, upstream_tasks_state, task.upstream_task_ids)))
    elif (tr == TR.ALL_SUCCESS):
        num_failures = (upstream - successes)
        if (num_failures > 0):
            (yield self._failing_status(reason="Task's    trigger    rule    '{0}'    requires    all    upstream    tasks    to    have    succeeded,    but    found    {1}    non-success(es).    upstream_tasks_state={2},    upstream_task_ids={3}".format(tr, num_failures, upstream_tasks_state, task.upstream_task_ids)))
    elif (tr == TR.ALL_FAILED):
        num_successes = ((upstream - failed) - upstream_failed)
        if (num_successes > 0):
            (yield self._failing_status(reason="Task's    trigger    rule    '{0}'    requires    all    upstream    tasks    to    have    failed,    but    found    {1}    non-failure(s).    upstream_tasks_state={2},    upstream_task_ids={3}".format(tr, num_successes, upstream_tasks_state, task.upstream_task_ids)))
    elif (tr == TR.ALL_DONE):
        if (not upstream_done):
            (yield self._failing_status(reason="Task's    trigger    rule    '{0}'    requires    all    upstream    tasks    to    have    completed,    but    found    {1}    task(s)    that    weren't    done.    upstream_tasks_state={2},    upstream_task_ids={3}".format(tr, (upstream - done), upstream_tasks_state, task.upstream_task_ids)))
    else:
        (yield self._failing_status(reason="No    strategy    to    evaluate    trigger    rule    '{0}'.".format(tr)))
