mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-04-14 23:12:37 +00:00
Compare commits
1 Commits
1.14.2a4
...
feat/googl
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0000239b3c |
@@ -86,6 +86,22 @@ CREWAI_PLATFORM_INTEGRATION_TOKEN=your_enterprise_token
|
||||
|
||||
</Accordion>
|
||||
|
||||
<Accordion title="google_drive/upload_from_file">
|
||||
**Description:** Upload a file from a local path to Google Drive. The file is read directly from disk — its content never passes through the LLM context window, making this efficient for large or binary files.
|
||||
|
||||
<Note>
|
||||
Unlike `google_drive/upload_file` which requires passing file content as a parameter (consuming LLM tokens), this action takes a file path and reads the file directly. Use this for binary files, large documents, or any case where you want to avoid token overhead.
|
||||
</Note>
|
||||
|
||||
**Parameters:**
|
||||
- `file_path` (string, required): Path to the local file to upload.
|
||||
- `name` (string, optional): Name for the file in Google Drive (defaults to the local filename).
|
||||
- `mime_type` (string, optional): MIME type of the file (auto-detected from file extension if not provided).
|
||||
- `parent_folder_id` (string, optional): ID of the parent folder where the file should be created.
|
||||
- `description` (string, optional): Description of the file.
|
||||
|
||||
</Accordion>
|
||||
|
||||
<Accordion title="google_drive/download_file">
|
||||
**Description:** Download a file from Google Drive.
|
||||
|
||||
@@ -205,6 +221,42 @@ crew = Crew(
|
||||
crew.kickoff()
|
||||
```
|
||||
|
||||
### Uploading Files from Disk (Token-Efficient)
|
||||
|
||||
```python
|
||||
from crewai import Agent, Task, Crew
|
||||
|
||||
# Use upload_from_file to avoid passing file content through the LLM context
|
||||
upload_agent = Agent(
|
||||
role="File Uploader",
|
||||
goal="Upload local files to Google Drive efficiently",
|
||||
backstory="An AI assistant that uploads files without wasting context tokens.",
|
||||
apps=[
|
||||
'google_drive/upload_from_file',
|
||||
'google_drive/list_files'
|
||||
]
|
||||
)
|
||||
|
||||
# The agent only needs to specify the file path — the file is read directly
|
||||
# from disk and never passes through the LLM context window
|
||||
upload_task = Task(
|
||||
description="Upload the file at /data/reports/quarterly_report.pdf to Google Drive in the Reports folder",
|
||||
agent=upload_agent,
|
||||
expected_output="File uploaded successfully with file ID"
|
||||
)
|
||||
|
||||
crew = Crew(
|
||||
agents=[upload_agent],
|
||||
tasks=[upload_task]
|
||||
)
|
||||
|
||||
crew.kickoff()
|
||||
```
|
||||
|
||||
<Tip>
|
||||
Use `google_drive/upload_from_file` instead of `google_drive/upload_file` when uploading existing files from disk. This is especially important for binary files (PDFs, images, etc.) or large files that would otherwise consume significant LLM context tokens or exceed context limits.
|
||||
</Tip>
|
||||
|
||||
### Advanced File Management
|
||||
|
||||
```python
|
||||
|
||||
@@ -7,6 +7,9 @@ through the CrewAI platform API.
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import (
|
||||
CrewAIPlatformActionTool,
|
||||
)
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
|
||||
CrewAIPlatformFileUploadTool,
|
||||
)
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_tool_builder import (
|
||||
CrewaiPlatformToolBuilder,
|
||||
)
|
||||
@@ -17,6 +20,7 @@ from crewai_tools.tools.crewai_platform_tools.crewai_platform_tools import (
|
||||
|
||||
__all__ = [
|
||||
"CrewAIPlatformActionTool",
|
||||
"CrewAIPlatformFileUploadTool",
|
||||
"CrewaiPlatformToolBuilder",
|
||||
"CrewaiPlatformTools",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,165 @@
|
||||
"""CrewAI Platform File Upload Tool.
|
||||
|
||||
Uploads a file from disk to Google Drive without passing file content
|
||||
through the LLM context window. This avoids token waste and context
|
||||
limit issues with large or binary files.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from crewai.tools import BaseTool
|
||||
from pydantic import Field, create_model
|
||||
|
||||
from crewai_tools.tools.crewai_platform_tools.misc import (
|
||||
get_platform_api_base_url,
|
||||
get_platform_integration_token,
|
||||
)
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Google Drive simple upload limit
|
||||
MAX_FILE_SIZE_BYTES = 50 * 1024 * 1024 # 50 MB
|
||||
|
||||
|
||||
_UploadFromFileSchema = create_model(
|
||||
"UploadFromFileSchema",
|
||||
file_path=(str, Field(description="Path to the local file to upload")),
|
||||
name=(
|
||||
Optional[str],
|
||||
Field(
|
||||
default=None,
|
||||
description="Name for the file in Google Drive (defaults to the local filename)",
|
||||
),
|
||||
),
|
||||
mime_type=(
|
||||
Optional[str],
|
||||
Field(
|
||||
default=None,
|
||||
description="MIME type of the file (auto-detected if not provided)",
|
||||
),
|
||||
),
|
||||
parent_folder_id=(
|
||||
Optional[str],
|
||||
Field(
|
||||
default=None,
|
||||
description="ID of the parent folder where the file should be created",
|
||||
),
|
||||
),
|
||||
description=(
|
||||
Optional[str],
|
||||
Field(default=None, description="Description of the file"),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class CrewAIPlatformFileUploadTool(BaseTool):
|
||||
"""Upload a file from disk to Google Drive.
|
||||
|
||||
Reads the file locally and sends it to the platform API, bypassing
|
||||
the LLM context window entirely. Supports auto-detection of MIME type
|
||||
and optional file naming.
|
||||
"""
|
||||
|
||||
name: str = "google_drive_upload_from_file"
|
||||
description: str = (
|
||||
"Upload a file from a local path to Google Drive. "
|
||||
"The file is read directly from disk — its content never passes "
|
||||
"through the LLM context, making this efficient for large or binary files."
|
||||
)
|
||||
args_schema: type = _UploadFromFileSchema
|
||||
|
||||
def _run(
|
||||
self,
|
||||
file_path: str,
|
||||
name: str | None = None,
|
||||
mime_type: str | None = None,
|
||||
parent_folder_id: str | None = None,
|
||||
description: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> str:
|
||||
try:
|
||||
path = Path(file_path).expanduser().resolve()
|
||||
|
||||
if not path.exists():
|
||||
return f"Error: File not found: {file_path}"
|
||||
|
||||
if not path.is_file():
|
||||
return f"Error: Path is not a file: {file_path}"
|
||||
|
||||
file_size = path.stat().st_size
|
||||
if file_size > MAX_FILE_SIZE_BYTES:
|
||||
return (
|
||||
f"Error: File size ({file_size / (1024 * 1024):.1f} MB) exceeds "
|
||||
f"the 50 MB limit for simple uploads. Consider splitting the file "
|
||||
f"or using a resumable upload method."
|
||||
)
|
||||
|
||||
# Read and encode file content
|
||||
content_bytes = path.read_bytes()
|
||||
content_b64 = base64.b64encode(content_bytes).decode("utf-8")
|
||||
|
||||
# Auto-detect MIME type if not provided
|
||||
if mime_type is None:
|
||||
guessed_type, _ = mimetypes.guess_type(str(path))
|
||||
mime_type = guessed_type or "application/octet-stream"
|
||||
|
||||
# Use filename if name not provided
|
||||
upload_name = name or path.name
|
||||
|
||||
# Build payload matching the existing upload_file action format
|
||||
payload_data: dict[str, Any] = {
|
||||
"name": upload_name,
|
||||
"content": content_b64,
|
||||
}
|
||||
if mime_type:
|
||||
payload_data["mime_type"] = mime_type
|
||||
if parent_folder_id:
|
||||
payload_data["parent_folder_id"] = parent_folder_id
|
||||
if description:
|
||||
payload_data["description"] = description
|
||||
|
||||
api_url = (
|
||||
f"{get_platform_api_base_url()}/actions/GOOGLE_DRIVE_SAVE_FILE/execute"
|
||||
)
|
||||
token = get_platform_integration_token()
|
||||
headers = {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
payload = {"integration": payload_data}
|
||||
|
||||
response = requests.post(
|
||||
url=api_url,
|
||||
headers=headers,
|
||||
json=payload,
|
||||
timeout=120, # Longer timeout for file uploads
|
||||
verify=os.environ.get("CREWAI_FACTORY", "false").lower() != "true",
|
||||
)
|
||||
|
||||
data = response.json()
|
||||
if not response.ok:
|
||||
if isinstance(data, dict):
|
||||
error_info = data.get("error", {})
|
||||
if isinstance(error_info, dict):
|
||||
error_message = error_info.get("message", json.dumps(data))
|
||||
else:
|
||||
error_message = str(error_info)
|
||||
else:
|
||||
error_message = str(data)
|
||||
return f"API request failed: {error_message}"
|
||||
|
||||
return json.dumps(data, indent=2)
|
||||
|
||||
except PermissionError:
|
||||
return f"Error: Permission denied reading file: {file_path}"
|
||||
except Exception as e:
|
||||
return f"Error uploading file: {e!s}"
|
||||
@@ -11,11 +11,22 @@ import requests
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_action_tool import (
|
||||
CrewAIPlatformActionTool,
|
||||
)
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
|
||||
CrewAIPlatformFileUploadTool,
|
||||
)
|
||||
from crewai_tools.tools.crewai_platform_tools.misc import (
|
||||
get_platform_api_base_url,
|
||||
get_platform_integration_token,
|
||||
)
|
||||
|
||||
# Apps that have client-side local tools (e.g. file-path-based upload)
|
||||
_LOCAL_TOOL_APPS = {
|
||||
"google_drive": [CrewAIPlatformFileUploadTool],
|
||||
}
|
||||
_LOCAL_TOOL_ACTIONS = {
|
||||
"google_drive/upload_from_file": CrewAIPlatformFileUploadTool,
|
||||
}
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -95,6 +106,23 @@ class CrewaiPlatformToolBuilder:
|
||||
|
||||
tools.append(tool)
|
||||
|
||||
# Inject client-side local tools based on requested apps
|
||||
added_local_tools: set[type] = set()
|
||||
for app in self._apps:
|
||||
# Check for specific action (e.g. "google_drive/upload_from_file")
|
||||
if app in _LOCAL_TOOL_ACTIONS:
|
||||
tool_cls = _LOCAL_TOOL_ACTIONS[app]
|
||||
if tool_cls not in added_local_tools:
|
||||
tools.append(tool_cls())
|
||||
added_local_tools.add(tool_cls)
|
||||
# Check for full app (e.g. "google_drive") — inject all local tools
|
||||
app_base = app.split("/")[0]
|
||||
if app_base in _LOCAL_TOOL_APPS:
|
||||
for tool_cls in _LOCAL_TOOL_APPS[app_base]:
|
||||
if tool_cls not in added_local_tools:
|
||||
tools.append(tool_cls())
|
||||
added_local_tools.add(tool_cls)
|
||||
|
||||
self._tools = tools
|
||||
|
||||
def __enter__(self) -> list[BaseTool]:
|
||||
|
||||
@@ -0,0 +1,363 @@
|
||||
"""Tests for CrewAIPlatformFileUploadTool."""
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool import (
|
||||
CrewAIPlatformFileUploadTool,
|
||||
MAX_FILE_SIZE_BYTES,
|
||||
)
|
||||
|
||||
|
||||
class TestCrewAIPlatformFileUploadTool:
|
||||
"""Test suite for the file upload tool."""
|
||||
|
||||
def setup_method(self):
|
||||
self.tool = CrewAIPlatformFileUploadTool()
|
||||
|
||||
def test_tool_name_and_description(self):
|
||||
assert self.tool.name == "google_drive_upload_from_file"
|
||||
assert "local path" in self.tool.description.lower()
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_successful_upload(self, mock_post):
|
||||
"""Test uploading a file successfully."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {
|
||||
"id": "file123",
|
||||
"name": "test.txt",
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("Hello, Google Drive!")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
result = self.tool._run(file_path=temp_path)
|
||||
|
||||
assert "file123" in result
|
||||
mock_post.assert_called_once()
|
||||
|
||||
call_kwargs = mock_post.call_args
|
||||
payload = call_kwargs.kwargs["json"]["integration"]
|
||||
assert payload["name"] == Path(temp_path).name
|
||||
assert payload["mime_type"] == "text/plain"
|
||||
# Verify content is base64-encoded
|
||||
decoded = base64.b64decode(payload["content"]).decode("utf-8")
|
||||
assert decoded == "Hello, Google Drive!"
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_custom_name_override(self, mock_post):
|
||||
"""Test that a custom name overrides the filename."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path, name="custom_report.txt")
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert payload["name"] == "custom_report.txt"
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_mime_type_auto_detection(self, mock_post):
|
||||
"""Test MIME type is auto-detected from file extension."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".json", delete=False
|
||||
) as f:
|
||||
f.write("{}")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path)
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert payload["mime_type"] == "application/json"
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_custom_mime_type(self, mock_post):
|
||||
"""Test that a custom MIME type overrides auto-detection."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path, mime_type="application/pdf")
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert payload["mime_type"] == "application/pdf"
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
def test_file_not_found(self):
|
||||
"""Test error when file does not exist."""
|
||||
result = self.tool._run(file_path="/nonexistent/path/file.txt")
|
||||
assert "Error: File not found" in result
|
||||
|
||||
def test_path_is_directory(self):
|
||||
"""Test error when path is a directory."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
result = self.tool._run(file_path=tmpdir)
|
||||
assert "Error: Path is not a file" in result
|
||||
|
||||
def test_file_too_large(self):
|
||||
"""Test error when file exceeds 50 MB limit."""
|
||||
with tempfile.NamedTemporaryFile(delete=False) as f:
|
||||
temp_path = f.name
|
||||
# Create a sparse file that reports as > 50 MB
|
||||
f.seek(MAX_FILE_SIZE_BYTES + 1)
|
||||
f.write(b"\0")
|
||||
|
||||
try:
|
||||
result = self.tool._run(file_path=temp_path)
|
||||
assert "exceeds" in result
|
||||
assert "50 MB" in result
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_optional_params_included(self, mock_post):
|
||||
"""Test that optional params are included in payload when provided."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(
|
||||
file_path=temp_path,
|
||||
parent_folder_id="folder456",
|
||||
description="My report",
|
||||
)
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert payload["parent_folder_id"] == "folder456"
|
||||
assert payload["description"] == "My report"
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_optional_params_excluded_when_none(self, mock_post):
|
||||
"""Test that optional params are NOT in payload when not provided."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path)
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert "parent_folder_id" not in payload
|
||||
assert "description" not in payload
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_api_error_response(self, mock_post):
|
||||
"""Test handling of API error responses."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = False
|
||||
mock_response.json.return_value = {
|
||||
"error": {"message": "Unauthorized"}
|
||||
}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
result = self.tool._run(file_path=temp_path)
|
||||
assert "API request failed" in result
|
||||
assert "Unauthorized" in result
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{
|
||||
"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token",
|
||||
"CREWAI_FACTORY": "true",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_ssl_verification_disabled_for_factory(self, mock_post):
|
||||
"""Test that SSL verification is disabled when CREWAI_FACTORY=true."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path)
|
||||
assert mock_post.call_args.kwargs["verify"] is False
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_ssl_verification_enabled_by_default(self, mock_post):
|
||||
"""Test that SSL verification is enabled by default."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".txt", delete=False
|
||||
) as f:
|
||||
f.write("content")
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path)
|
||||
assert mock_post.call_args.kwargs["verify"] is True
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
|
||||
@patch.dict(
|
||||
"os.environ",
|
||||
{"CREWAI_PLATFORM_INTEGRATION_TOKEN": "test_token"},
|
||||
clear=True,
|
||||
)
|
||||
@patch(
|
||||
"crewai_tools.tools.crewai_platform_tools.crewai_platform_file_upload_tool.requests.post"
|
||||
)
|
||||
def test_binary_file_upload(self, mock_post):
|
||||
"""Test uploading a binary file (e.g. PNG)."""
|
||||
mock_response = Mock()
|
||||
mock_response.ok = True
|
||||
mock_response.json.return_value = {"id": "file123"}
|
||||
mock_post.return_value = mock_response
|
||||
|
||||
binary_content = bytes(range(256))
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
suffix=".png", delete=False
|
||||
) as f:
|
||||
f.write(binary_content)
|
||||
temp_path = f.name
|
||||
|
||||
try:
|
||||
self.tool._run(file_path=temp_path)
|
||||
|
||||
payload = mock_post.call_args.kwargs["json"]["integration"]
|
||||
assert payload["mime_type"] == "image/png"
|
||||
decoded = base64.b64decode(payload["content"])
|
||||
assert decoded == binary_content
|
||||
finally:
|
||||
os.unlink(temp_path)
|
||||
Reference in New Issue
Block a user