docstr: Return the non completed tasks.
  :return: number of tasks in non completed state (SUCCESS, FAILED, SKIPPED, UPSTREAM_FAILED)
docstr_tok: ["Return","the","non","completed","tasks",".",":","return",":","number","of","tasks","in","non","completed","state","(","SUCCESS",",","FAILED",",","SKIPPED",",","UPSTREAM_FAILED",")"]
code: def _get_num_tasks_with_non_completed_state(self):
      """
      Return the non completed tasks.
      :return: number of tasks in non completed state (SUCCESS, FAILED, SKIPPED, UPSTREAM_FAILED)
      """
      expected = len(
          self.INITIAL_TASK_STATES.values()
      ) - self._get_num_tasks_with_starting_state(
          State.SUCCESS, inclusion=True
      )
      expected = expected - self._get_num_tasks_with_starting_state(
          State.FAILED, inclusion=True
      )
      expected = expected - self._get_num_tasks_with_starting_state(
          State.SKIPPED, inclusion=True
      )
      expected = expected - self._get_num_tasks_with_starting_state(
          State.UPSTREAM_FAILED, inclusion=True
      )
  
      return expected
  
imports: from __future__ import annotations
  import datetime
  from typing import Callable
  import pytest
  from sqlalchemy.orm import eagerload
  from airflow import models
  from airflow.api.common.mark_tasks import _create_dagruns, _DagRunInfo, set_dag_run_state_to_failed, set_dag_run_state_to_queued, set_dag_run_state_to_running, set_dag_run_state_to_success, set_state
  from airflow.models import DagRun
  from airflow.utils import timezone
  from airflow.utils.session import create_session, provide_session
  from airflow.utils.state import State
  from airflow.utils.types import DagRunType
  from tests.test_utils.db import clear_db_runs
  from tests.test_utils.mapping import expand_mapped_task
label_desc: Use image node_id 5 for calling the TestMarkDAGRun obj's underlying member method code with example usage: obj._get_num_tasks_with_non_completed_state() and returns: expected