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 setUp(self):
      super().setUp()
      # setup airflow_home in a temp directory, config and init db.
      self._airflow_home = self.tmp_dir
      self.enter_context(
          test_case_utils.override_env_var(
              "AIRFLOW_HOME", self._airflow_home
          )
      )
      self.enter_context(
          test_case_utils.override_env_var("HOME", self._airflow_home)
      )
      absl.logging.info(
          "Using %s as AIRFLOW_HOME and HOME in this e2e test",
          self._airflow_home,
      )
  
      self._mysql_container_name = (
          "airflow_" + test_utils.generate_random_id()
      )
      ip_address, db_port = airflow_test_utils.create_mysql_container(
          self._mysql_container_name
      )
      self.addCleanup(
          airflow_test_utils.delete_mysql_container,
          self._mysql_container_name,
      )
      os.environ[
          "AIRFLOW__CORE__SQL_ALCHEMY_CONN"
      ] = "mysql://tfx@%s:%d/airflow" % (ip_address, db_port)
  
      # Set a couple of important environment variables. See
      # https://airflow.apache.org/howto/set-config.html for details.
      os.environ["AIRFLOW__CORE__DAGS_FOLDER"] = os.path.join(
          self._airflow_home, "dags"
      )
      os.environ["AIRFLOW__CORE__BASE_LOG_FOLDER"] = os.path.join(
          self._airflow_home, "logs"
      )
      os.environ["AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT"] = "300"
      # Do not load examples to make this a bit faster.
      os.environ["AIRFLOW__CORE__LOAD_EXAMPLES"] = "False"
      # Following environment variables make scheduler process dags faster.
      os.environ["AIRFLOW__SCHEDULER__JOB_HEARTBEAT_SEC"] = "1"
      os.environ["AIRFLOW__SCHEDULER__SCHEDULER_HEARTBEAT_SEC"] = "1"
      os.environ["AIRFLOW__SCHEDULER__RUN_DURATION"] = "-1"
      os.environ["AIRFLOW__SCHEDULER__MIN_FILE_PROCESS_INTERVAL"] = "1"
      os.environ["AIRFLOW__SCHEDULER__PRINT_STATS_INTERVAL"] = "30"
  
      # Following fields are specific to the chicago_taxi_simple example.
      self._dag_id = "chicago_taxi_simple"
      self._run_id = "manual_run_id_1"
      # This execution date must be after the start_date in chicago_taxi_simple
      # but before current execution date.
      self._execution_date = "2019-02-01T01:01:01"
      self._all_tasks = [
          "CsvExampleGen",
          "Evaluator",
          "ExampleValidator",
          "Pusher",
          "SchemaGen",
          "StatisticsGen",
          "Trainer",
          "Transform",
          "latest_blessed_model_resolver",
      ]
      # Copy dag file and data.
      chicago_taxi_pipeline_dir = os.path.dirname(__file__)
      simple_pipeline_file = os.path.join(
          chicago_taxi_pipeline_dir, "taxi_pipeline_simple.py"
      )
  
      io_utils.copy_file(
          simple_pipeline_file,
          os.path.join(
              self._airflow_home, "dags", "taxi_pipeline_simple.py"
          ),
      )
  
      data_dir = os.path.join(
          chicago_taxi_pipeline_dir, "data", "simple"
      )
      content = fileio.listdir(data_dir)
      assert content, "content in {} is empty".format(data_dir)
      target_data_dir = os.path.join(
          self._airflow_home, "taxi", "data", "simple"
      )
      io_utils.copy_dir(data_dir, target_data_dir)
      assert fileio.isdir(target_data_dir)
      content = fileio.listdir(target_data_dir)
      assert content, "content in {} is {}".format(
          target_data_dir, content
      )
      io_utils.copy_file(
          os.path.join(chicago_taxi_pipeline_dir, "taxi_utils.py"),
          os.path.join(self._airflow_home, "taxi", "taxi_utils.py"),
      )
  
      # Initialize database.
      subprocess.run(["airflow", "db", "init"], check=True)
  
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 5 for calling the AirflowEndToEndTest obj's underlying member method code with example usage: obj.setUp() without return types