docstr: Asserts that a task is deferred and a GCSUploadSessionTrigger will be fired
  when the GCSUploadSessionCompleteAsyncSensor is executed.
code: def test_gcs_upload_session_complete_async_sensor(self, mock_hook):
      """
      Asserts that a task is deferred and a GCSUploadSessionTrigger will be fired
      when the GCSUploadSessionCompleteAsyncSensor is executed.
      """
      mock_hook.return_value.is_bucket_updated.return_value = False
      with pytest.raises(TaskDeferred) as exc:
          self.OPERATOR.execute(mock.MagicMock())
      assert isinstance(
          exc.value.trigger, GCSUploadSessionTrigger
      ), "Trigger is not a GCSUploadSessionTrigger"
  
imports: from __future__ import annotations
  from datetime import datetime, timedelta, timezone
  from unittest import mock
  import pendulum
  import pytest
  from google.cloud.storage.retry import DEFAULT_RETRY
  from airflow.exceptions import AirflowProviderDeprecationWarning, AirflowSensorTimeout, AirflowSkipException, TaskDeferred
  from airflow.models.dag import DAG, AirflowException
  from airflow.providers.google.cloud.sensors.gcs import GCSObjectExistenceAsyncSensor, GCSObjectExistenceSensor, GCSObjectsWithPrefixExistenceSensor, GCSObjectUpdateSensor, GCSUploadSessionCompleteSensor, ts_function
  from airflow.providers.google.cloud.triggers.gcs import GCSBlobTrigger, GCSCheckBlobUpdateTimeTrigger, GCSPrefixBlobTrigger, GCSUploadSessionTrigger
label_desc: Use image node_id 1 for calling the TestGCSUploadSessionCompleteAsyncSensor obj's underlying member method code with example usage: obj.test_gcs_upload_session_complete_async_sensor(mock_hook) without return types