mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-23 15:55:11 +00:00
feat: add google_drive/upload_from_file action for token-efficient file uploads
Add a new CrewAIPlatformFileUploadTool that reads files directly from disk and uploads to Google Drive via the platform API, bypassing the LLM context window entirely. This solves two problems: 1. File content no longer consumes LLM tokens 2. Binary/large files no longer risk exceeding the 128k context limit The new action accepts a file_path parameter instead of content. It handles: - Auto-detection of MIME type from file extension - Optional custom file name (defaults to local filename) - File size validation (50 MB limit for simple uploads) - Base64 encoding of file content before sending to API The existing google_drive/upload_file action is unchanged — full backwards compatibility with the 10k+ existing executions. Changes: - New tool: CrewAIPlatformFileUploadTool - Builder auto-injects the tool when apps include google_drive or google_drive/upload_from_file - 14 unit tests covering upload, error handling, MIME detection, SSL - Updated docs with new action reference and usage examples
This commit is contained in:
@@ -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