docstr: Pusher-only test pipeline to AI Platform Prediction.
docstr_tok: ["Pusher-only","test","pipeline","to","AI","Platform","Prediction","."]
code: def testAIPlatformPusherPipeline(self):
      """Pusher-only test pipeline to AI Platform Prediction."""
      pipeline_name_base = self._make_unique_pipeline_name(
          "kubeflow-aip-pusher"
      )
      # AI Platform does not accept '-' in the model name.
      model_name = ("%s_model" % pipeline_name_base).replace("-", "_")
      self.addCleanup(
          kubeflow_test_utils.delete_ai_platform_model, model_name
      )
  
      def _pusher(model_importer, model_blessing_importer):
          return Pusher(
              custom_executor_spec=executor_spec.ExecutorClassSpec(
                  ai_platform_pusher_executor.Executor
              ),
              model=model_importer.outputs["result"],
              model_blessing=model_blessing_importer.outputs["result"],
              custom_config={
                  tfx.extensions.google_cloud_ai_platform.experimental.PUSHER_SERVING_ARGS_KEY: {
                      "model_name": model_name,
                      "project_id": self._GCP_PROJECT_ID,
                  }
              },
          )
  
      # Use default service_name / api_version.
      (
          service_name,
          api_version,
      ) = runner.get_service_name_and_api_version({})
      api = discovery.build(
          service_name,
          api_version,
          requestBuilder=telemetry_utils.TFXHttpRequest,
      )
  
      # The model should be NotFound yet.
      with self.assertRaisesRegex(
          googleapiclient_errors.HttpError, "HttpError 404"
      ):
          self._sendDummyRequestToModel(
              api, self._GCP_PROJECT_ID, model_name
          )
  
      # Test creation of multiple versions under the same model_name.
      pipeline_name_1 = "%s-1" % pipeline_name_base
      pipeline_1 = self._create_pipeline(
          pipeline_name_1,
          [
              self.model_1_importer,
              self.model_blessing_1_importer,
              _pusher(
                  self.model_1_importer, self.model_blessing_1_importer
              ),
          ],
      )
      self._compile_and_run_pipeline(pipeline_1)
      self.assertEqual(
          1,
          self._getNumberOfVersionsForModel(
              api, self._GCP_PROJECT_ID, model_name
          ),
      )
      self._sendDummyRequestToModel(
          api, self._GCP_PROJECT_ID, model_name
      )
  
      pipeline_name_2 = "%s-2" % pipeline_name_base
      pipeline_2 = self._create_pipeline(
          pipeline_name_2,
          [
              self.model_2_importer,
              self.model_blessing_2_importer,
              _pusher(
                  self.model_2_importer, self.model_blessing_2_importer
              ),
          ],
      )
      self._compile_and_run_pipeline(pipeline_2)
      self.assertEqual(
          2,
          self._getNumberOfVersionsForModel(
              api, self._GCP_PROJECT_ID, model_name
          ),
      )
      self._sendDummyRequestToModel(
          api, self._GCP_PROJECT_ID, model_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 17 for calling the KubeflowGCPIntegrationTest obj's underlying member method code with example usage: obj.testAIPlatformPusherPipeline() and returns: Pusher