docstr: Asserts that a task is deferred and a CloudComposerExecutionTrigger will be fired
  when the CloudComposerEnvironmentSensor is executed.
code: def test_cloud_composer_existence_sensor_async(self):
      """
      Asserts that a task is deferred and a CloudComposerExecutionTrigger will be fired
      when the CloudComposerEnvironmentSensor is executed.
      """
      task = CloudComposerEnvironmentSensor(
          task_id="task_id",
          project_id=TEST_PROJECT_ID,
          region=TEST_REGION,
          operation_name=TEST_OPERATION_NAME,
      )
      with pytest.raises(TaskDeferred) as exc:
          task.execute(context={})
      assert isinstance(
          exc.value.trigger, CloudComposerExecutionTrigger
      ), "Trigger is not a CloudComposerExecutionTrigger"
  
file_path: airflow/tests/providers/google/cloud/sensors/test_cloud_composer.py
imports: from __future__ import annotations
  from unittest import mock
  import pytest
  from airflow.exceptions import AirflowException, AirflowSkipException, TaskDeferred
  from airflow.providers.google.cloud.sensors.cloud_composer import CloudComposerEnvironmentSensor
  from airflow.providers.google.cloud.triggers.cloud_composer import CloudComposerExecutionTrigger
label_desc: Use image node_id 1 for calling the TestCloudComposerEnvironmentSensor obj's underlying member method code with example usage: obj.test_cloud_composer_existence_sensor_async() without return types