Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion src/palace/manager/service/google_drive/google_drive.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,8 +17,19 @@ def __init__(self, api_client: DriveResource) -> None:
self.api_client = api_client

def get_file(self, name: str, parent_folder_id: str | None = None) -> File | None:
"""
Return the first non-trashed file or folder with the given name.

query = f"name = '{name}'"
:param name: The exact name to search for.
:param parent_folder_id: If provided, restrict the search to items
whose parent is this folder/drive ID.
:return: The first matching, non-trashed ``File`` object, or ``None``
if no match is found.
"""
# Explicitly exclude trashed items so that a folder moved to the
# Drive trash is not mistaken for a live folder, which would cause
# uploads to land inside a trashed (invisible) directory.
query = f"name = '{name}' and trashed = false"

if parent_folder_id:
query += f" and '{parent_folder_id}' in parents"
Expand Down
35 changes: 35 additions & 0 deletions tests/manager/service/google_drive/test_google_drive.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import json
from io import BytesIO
from typing import TYPE_CHECKING
from urllib.parse import unquote_plus

import pytest
from googleapiclient.discovery import build
Expand Down Expand Up @@ -114,6 +115,40 @@ def test_create_file(self):
assert "Hello world" in body
assert f"Content-Type: {mime_type}" in body

def test_get_file_excludes_trashed_items(self):
"""get_file must include 'trashed = false' in its Drive query."""
file_name = "some-folder"
file_id = "folder-id"
http_mock_sequence = HttpMockSequence(
[
(
{"status": "200"},
json.dumps(
{
"files": [
{
"kind": "drive#file",
"id": file_id,
"name": file_name,
"mimeType": "application/vnd.google-apps.folder",
}
]
}
),
),
]
)
service = drive_service(http=http_mock_sequence)

result = service.get_file(name=file_name, parent_folder_id="parent-id")

assert result is not None
assert result["id"] == file_id

# Verify the query sent to the Drive API contains the trashed filter.
request_uri = unquote_plus(http_mock_sequence.request_sequence[0][0])
assert "trashed = false" in request_uri

def test_create_existing_file_fails(self):
file_name = "file.txt"
file_id = "file-id"
Expand Down
Loading