Fix GoogleDriveHook writing files to trashed folders on upload v2 (#29119)

Author: vchiapaikeo

airflow/providers/google/suite/hooks/drive.py

@@ -81,7 +81,8 @@ def _ensure_folders_exists(self, path: str) -> str:
         for current_folder in folders:
             self.log.debug("Looking for %s directory with %s parent", current_folder, current_parent)
             conditions = [
-                "mimeType = 'application/vnd.google-apps.folder'",
+                "trashed=false",
+                "mimeType='application/vnd.google-apps.folder'",
                 f"name='{current_folder}'",
                 f"'{current_parent}' in parents",
             ]

tests/providers/google/suite/hooks/test_drive.py

@@ -57,6 +57,16 @@ def test_ensure_folders_exists_when_no_folder_exists(self, mock_get_conn):
 
         mock_get_conn.assert_has_calls(
             [
+                mock.call()
+                .files()
+                .list(
+                    q=(
+                        "trashed=false and mimeType='application/vnd.google-apps.folder' "
+                        "and name='AAA' and 'root' in parents"
+                    ),
+                    spaces="drive",
+                    fields="files(id, name)",
+                ),
                 mock.call()
                 .files()
                 .create(
@@ -119,6 +129,19 @@ def test_ensure_folders_exists_when_some_folders_exists(self, mock_get_conn):
 
         mock_get_conn.assert_has_calls(
             [
+                *[
+                    mock.call()
+                    .files()
+                    .list(
+                        q=(
+                            "trashed=false and mimeType='application/vnd.google-apps.folder' "
+                            f"and name='{d}' and '{key}' in parents"
+                        ),
+                        spaces="drive",
+                        fields="files(id, name)",
+                    )
+                    for d, key in [("AAA", "root"), ("BBB", "ID_1"), ("CCC", "ID_2")]
+                ],
                 mock.call()
                 .files()
                 .create(
@@ -156,6 +179,25 @@ def test_ensure_folders_exists_when_all_folders_exists(self, mock_get_conn):
 
         result_value = self.gdrive_hook._ensure_folders_exists("AAA/BBB/CCC/DDD")
 
+        mock_get_conn.assert_has_calls(
+            [
+                *[
+                    mock.call()
+                    .files()
+                    .list(
+                        q=(
+                            "trashed=false and mimeType='application/vnd.google-apps.folder' "
+                            f"and name='{d}' and '{key}' in parents"
+                        ),
+                        spaces="drive",
+                        fields="files(id, name)",
+                    )
+                    for d, key in [("AAA", "root"), ("BBB", "ID_1"), ("CCC", "ID_2"), ("DDD", "ID_3")]
+                ],
+            ],
+            any_order=True,
+        )
+
         mock_get_conn.return_value.files.return_value.create.assert_not_called()
         assert "ID_4" == result_value