Raise exception when GCP credential doesn't support account impersona…

…tion (#8213)

airflow/providers/google/cloud/utils/credentials_provider.py

@@ -241,7 +241,14 @@ def get_credentials_and_project_id(
         project_id = credentials.project_id
 
     if delegate_to:
-        credentials = credentials.with_subject(delegate_to)
+        if hasattr(credentials, 'with_subject'):
+            credentials = credentials.with_subject(delegate_to)
+        else:
+            raise AirflowException(
+                "The `delegate_to` parameter cannot be used here as the current "
+                "authentication method does not support account impersonate. "
+                "Please use service-account for authorization."
+            )
 
     return credentials, project_id
 

tests/providers/google/common/hooks/test_base_google.py

@@ -421,6 +421,20 @@ def test_get_credentials_and_project_id_with_default_auth_and_delegate(
         )
         self.assertEqual((mock_credentials, "PROJECT_ID"), result)
 
+    @mock.patch('google.auth.default')
+    def test_get_credentials_and_project_id_with_default_auth_and_unsupported_delegate(
+        self, mock_auth_default
+    ):
+        self.instance.delegate_to = "TEST_DELLEGATE_TO"
+        mock_credentials = mock.MagicMock(spec=google.auth.compute_engine.Credentials)
+        mock_auth_default.return_value = (mock_credentials, "PROJECT_ID")
+
+        with self.assertRaisesRegex(AirflowException, re.escape(
+            "The `delegate_to` parameter cannot be used here as the current authentication method does not "
+            "support account impersonate. Please use service-account for authorization."
+        )):
+            self.instance._get_credentials_and_project_id()
+
     @mock.patch(  # type: ignore
         MODULE_NAME + '.get_credentials_and_project_id',
         return_value=("CREDENTIALS", "PROJECT_ID")