docstr: BigQuery ML Pusher pipeline on CAIP.
docstr_tok: ["BigQuery","ML","Pusher","pipeline","on","CAIP","."]
code: def testBigQueryMlPusherPipeline(self):
      """BigQuery ML Pusher pipeline on CAIP."""
      pipeline_name = self._make_unique_pipeline_name(
          "kubeflow-aip-bqml-pusher"
      )
      # Big Query does not accept '-' in the dataset name.
      dataset_name = ("%s_model" % pipeline_name).replace("-", "_")
      self.addCleanup(
          _delete_bigquery_dataset, dataset_name, self._GCP_PROJECT_ID
      )
  
      api = discovery.build("bigquery", "v2")
      api.datasets().insert(
          projectId=self._GCP_PROJECT_ID,
          body={
              "location": "US",
              "projectId": self._GCP_PROJECT_ID,
              "datasetReference": {
                  "datasetId": dataset_name,
                  "projectId": self._GCP_PROJECT_ID,
              },
          },
      ).execute()
  
      def _pusher(
          model_importer, model_blessing_importer, bigquery_dataset_id
      ):
          return Pusher(
              custom_executor_spec=executor_spec.ExecutorClassSpec(
                  bigquery_pusher_executor.Executor
              ),
              model=model_importer.outputs["result"],
              model_blessing=model_blessing_importer.outputs["result"],
              custom_config={
                  bigquery_pusher_executor.SERVING_ARGS_KEY: {
                      "bq_dataset_id": bigquery_dataset_id,
                      "model_name": pipeline_name,
                      "project_id": self._GCP_PROJECT_ID,
                  }
              },
          )
  
      # The model list should be empty
      self.assertEmpty(
          self._get_list_bigqueryml_models(api, dataset_name)
      )
  
      # Test creation of multiple versions under the same model_name.
      pipeline = self._create_pipeline(
          pipeline_name,
          [
              self.model_1_importer,
              self.model_blessing_1_importer,
              _pusher(
                  self.model_1_importer,
                  self.model_blessing_1_importer,
                  dataset_name,
              ),
          ],
      )
      self._compile_and_run_pipeline(pipeline)
      self.assertIn(
          pipeline_name,
          self._get_list_bigqueryml_models(api, dataset_name),
      )
  
imports: import os
  import absl
  from googleapiclient import discovery
  from googleapiclient import errors
  import tensorflow
  from tfx import v1
  from tfx.components.pusher.component import Pusher
  from tfx.components.trainer.component import Trainer
  from tfx.dsl.components.base import executor_spec
  from tfx.dsl.components.common import importer
  from tfx.dsl.io import fileio
  from tfx.extensions.google_cloud_ai_platform import constants
  from tfx.extensions.google_cloud_ai_platform import runner
  from tfx.extensions.google_cloud_ai_platform.pusher import executor
  from tfx.extensions.google_cloud_ai_platform.trainer import executor
  from tfx.extensions.google_cloud_ai_platform.tuner import component
  from tfx.extensions.google_cloud_ai_platform.tuner import executor
  from tfx.extensions.google_cloud_big_query.pusher import executor
  from tfx.orchestration import test_utils
  from tfx.orchestration.kubeflow import test_utils
  from tfx.proto import trainer_pb2
  from tfx.proto import tuner_pb2
  from tfx.types import standard_artifacts
  from tfx.utils import path_utils
  from tfx.utils import telemetry_utils
label_desc: Use image node_id 14 for calling the KubeflowGCPIntegrationTest obj's underlying member method code with example usage: obj.testBigQueryMlPusherPipeline() and returns: Pusher