class_docstr: An end to end test using fully orchestrated Airflow.
class_docstr_tok: ["An","end","to","end","test","using","fully","orchestrated","Airflow","."]
code: def testSimplePipeline(self):
      # We will use subprocess to start the DAG instead of webserver, so only
      # need to start a scheduler on the background.
      with airflow_test_utils.AirflowScheduler():
          subprocess.run(
              ["airflow", "dags", "unpause", self._dag_id], check=True
          )
          subprocess.run(
              [
                  "airflow",
                  "dags",
                  "trigger",
                  self._dag_id,
                  "-r",
                  self._run_id,
                  "-e",
                  self._execution_date,
              ],
              check=True,
          )
          absl.logging.info("Dag triggered: %s", self._dag_id)
  
          pending_tasks = set(self._all_tasks)
          attempts = (
              int(
                  _MAX_TASK_STATE_CHANGE_SEC
                  / _TASK_POLLING_INTERVAL_SEC
              )
              + 1
          )
          while True:
              if not pending_tasks:
                  absl.logging.info("No pending task left anymore")
                  return
              for _ in range(attempts):
                  absl.logging.debug("Polling task state")
                  still_pending = self._CheckPendingTasks(pending_tasks)
                  if len(still_pending) != len(pending_tasks):
                      pending_tasks = still_pending
                      break
                  absl.logging.info(
                      "Polling task state after %d secs",
                      _TASK_POLLING_INTERVAL_SEC,
                  )
                  time.sleep(_TASK_POLLING_INTERVAL_SEC)
              else:
                  self.fail(
                      "No pending tasks in %s finished within %d secs"
                      % (pending_tasks, _MAX_TASK_STATE_CHANGE_SEC)
                  )
  
imports: import os
  import platform
  import subprocess
  import time
  from typing import Sequence, Set
  import unittest
  import absl
  import tensorflow
  from tfx.dsl.io import fileio
  from tfx.orchestration.airflow import test_utils
  from tfx.tools.cli.e2e import test_utils
  from tfx.utils import io_utils
  from tfx.utils import test_case_utils
label_desc: Use image node_id 6 for calling the AirflowEndToEndTest obj's underlying member method code with example usage: obj.testSimplePipeline() without return types