code: def test_valid_source_code_union_field(
      self, mock_hook, source_code, project_id
  ):
      mock_hook.return_value.upload_function_zip.return_value = (
          "https://uploadUrl"
      )
      mock_hook.return_value.get_function.side_effect = mock.Mock(
          side_effect=HttpError(
              resp=MOCK_RESP_404, content=b"not found"
          )
      )
      mock_hook.return_value.create_new_function.return_value = True
      body = deepcopy(VALID_BODY)
      body.pop("sourceUploadUrl", None)
      body.pop("sourceArchiveUrl", None)
      body.pop("sourceRepository", None)
      body.pop("sourceRepositoryUrl", None)
      zip_path = source_code.pop("zip_path", None)
      body.update(source_code)
      if project_id:
          op = CloudFunctionDeployFunctionOperator(
              project_id="test_project_id",
              location="test_region",
              body=body,
              task_id="id",
              zip_path=zip_path,
          )
      else:
          op = CloudFunctionDeployFunctionOperator(
              location="test_region",
              body=body,
              task_id="id",
              zip_path=zip_path,
          )
      op.execute(context=mock.MagicMock())
      mock_hook.assert_called_once_with(
          api_version="v1",
          gcp_conn_id="google_cloud_default",
          impersonation_chain=None,
      )
      if zip_path:
          mock_hook.return_value.upload_function_zip.assert_called_once_with(
              project_id=project_id,
              location="test_region",
              zip_path="/path/to/file",
          )
      mock_hook.return_value.get_function.assert_called_once_with(
          "projects/test_project_id/locations/test_region/functions/helloWorld"
      )
      mock_hook.return_value.create_new_function.assert_called_once_with(
          project_id=project_id, location="test_region", body=body
      )
      mock_hook.reset_mock()
  
imports: from __future__ import annotations
  from copy import deepcopy
  from unittest import mock
  import httplib2
  import pytest
  from googleapiclient.errors import HttpError
  from airflow.exceptions import AirflowException
  from airflow.providers.google.cloud.operators.functions import FUNCTION_NAME_PATTERN, CloudFunctionDeleteFunctionOperator, CloudFunctionDeployFunctionOperator, CloudFunctionInvokeFunctionOperator
  from airflow.version import version
label_desc: Use image node_id 14 for calling the TestGcfFunctionDeploy obj's underlying member method code with example usage: obj.test_valid_source_code_union_field(mock_hook, source_code, project_id) without return types