Compare commits
4 Commits
v1.2.0-alp
...
v1.3.0-alp
| Author | SHA1 | Date | |
|---|---|---|---|
| 848cfd684f | |||
| ea175d55a2 | |||
| db12fca615 | |||
| f79ae5546f |
2
.github/workflows/build.yaml
vendored
2
.github/workflows/build.yaml
vendored
@@ -18,7 +18,7 @@ jobs:
|
|||||||
|
|
||||||
steps:
|
steps:
|
||||||
- name: Checkout repository
|
- name: Checkout repository
|
||||||
uses: actions/checkout@v4
|
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6
|
||||||
|
|
||||||
- name: Log in to Container Registry
|
- name: Log in to Container Registry
|
||||||
uses: docker/login-action@v3
|
uses: docker/login-action@v3
|
||||||
|
|||||||
26
CHANGELOG.md
26
CHANGELOG.md
@@ -5,6 +5,32 @@ All notable changes to this project will be documented in this file.
|
|||||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
|
||||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||||
|
|
||||||
|
## [1.3.0] - 2026-01-03
|
||||||
|
|
||||||
|
### Added
|
||||||
|
|
||||||
|
#### Attachment Upload
|
||||||
|
- **`upload_attachment` MCP tool**: Upload files to Grist documents
|
||||||
|
- Base64-encoded content input (required for JSON-based MCP protocol)
|
||||||
|
- Automatic MIME type detection from filename
|
||||||
|
- Returns attachment ID for linking to records via `update_records`
|
||||||
|
|
||||||
|
#### Usage
|
||||||
|
```python
|
||||||
|
# 1. Upload attachment
|
||||||
|
result = upload_attachment(
|
||||||
|
document="accounting",
|
||||||
|
filename="invoice.pdf",
|
||||||
|
content_base64="JVBERi0xLjQK..."
|
||||||
|
)
|
||||||
|
# Returns: {"attachment_id": 42, "filename": "invoice.pdf", "size_bytes": 31395}
|
||||||
|
|
||||||
|
# 2. Link to record
|
||||||
|
update_records(document="accounting", table="Bills", records=[
|
||||||
|
{"id": 1, "fields": {"Attachment": [42]}}
|
||||||
|
])
|
||||||
|
```
|
||||||
|
|
||||||
## [1.2.0] - 2026-01-02
|
## [1.2.0] - 2026-01-02
|
||||||
|
|
||||||
### Added
|
### Added
|
||||||
|
|||||||
187
docs/plans/2026-01-03-attachment-upload-design.md
Normal file
187
docs/plans/2026-01-03-attachment-upload-design.md
Normal file
@@ -0,0 +1,187 @@
|
|||||||
|
# Attachment Upload Feature Design
|
||||||
|
|
||||||
|
**Date:** 2026-01-03
|
||||||
|
**Status:** Approved
|
||||||
|
|
||||||
|
## Summary
|
||||||
|
|
||||||
|
Add an `upload_attachment` MCP tool to upload files to Grist documents and receive an attachment ID for linking to records.
|
||||||
|
|
||||||
|
## Design Decisions
|
||||||
|
|
||||||
|
| Decision | Choice | Rationale |
|
||||||
|
|----------|--------|-----------|
|
||||||
|
| Content encoding | Base64 string | MCP tools use JSON; binary must be encoded |
|
||||||
|
| Batch support | Single file only | YAGNI; caller can loop if needed |
|
||||||
|
| Linking behavior | Upload only, return ID | Single responsibility; use existing `update_records` to link |
|
||||||
|
| Download support | Not included | YAGNI; can add later if needed |
|
||||||
|
| Permission level | Write | Attachments are data, not schema |
|
||||||
|
| Proxy support | MCP tool only | Reduces scope; scripts can use Grist API directly |
|
||||||
|
|
||||||
|
## Tool Interface
|
||||||
|
|
||||||
|
### Input Schema
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"document": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "Document name"
|
||||||
|
},
|
||||||
|
"filename": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "Filename with extension (e.g., 'invoice.pdf')"
|
||||||
|
},
|
||||||
|
"content_base64": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "File content as base64-encoded string"
|
||||||
|
},
|
||||||
|
"content_type": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "MIME type (optional, auto-detected from filename if omitted)"
|
||||||
|
}
|
||||||
|
},
|
||||||
|
"required": ["document", "filename", "content_base64"]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Response
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"attachment_id": 42,
|
||||||
|
"filename": "invoice.pdf",
|
||||||
|
"size_bytes": 30720
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Usage Example
|
||||||
|
|
||||||
|
```python
|
||||||
|
# 1. Upload attachment
|
||||||
|
result = upload_attachment(
|
||||||
|
document="accounting",
|
||||||
|
filename="Invoice-001.pdf",
|
||||||
|
content_base64="JVBERi0xLjQK..."
|
||||||
|
)
|
||||||
|
|
||||||
|
# 2. Link to record via existing update_records tool
|
||||||
|
update_records("Bills", [{
|
||||||
|
"id": 1,
|
||||||
|
"fields": {"Attachment": [result["attachment_id"]]}
|
||||||
|
}])
|
||||||
|
```
|
||||||
|
|
||||||
|
## Implementation
|
||||||
|
|
||||||
|
### Files to Modify
|
||||||
|
|
||||||
|
1. **`src/grist_mcp/grist_client.py`** - Add `upload_attachment()` method
|
||||||
|
2. **`src/grist_mcp/tools/write.py`** - Add tool function
|
||||||
|
3. **`src/grist_mcp/server.py`** - Register tool
|
||||||
|
|
||||||
|
### GristClient Method
|
||||||
|
|
||||||
|
```python
|
||||||
|
async def upload_attachment(
|
||||||
|
self,
|
||||||
|
filename: str,
|
||||||
|
content: bytes,
|
||||||
|
content_type: str | None = None
|
||||||
|
) -> dict:
|
||||||
|
"""Upload a file attachment. Returns attachment metadata."""
|
||||||
|
if content_type is None:
|
||||||
|
content_type = "application/octet-stream"
|
||||||
|
|
||||||
|
files = {"upload": (filename, content, content_type)}
|
||||||
|
|
||||||
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
||||||
|
response = await client.post(
|
||||||
|
f"{self._base_url}/attachments",
|
||||||
|
headers=self._headers,
|
||||||
|
files=files,
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
# Grist returns list of attachment IDs
|
||||||
|
attachment_ids = response.json()
|
||||||
|
return {
|
||||||
|
"attachment_id": attachment_ids[0],
|
||||||
|
"filename": filename,
|
||||||
|
"size_bytes": len(content),
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Tool Function
|
||||||
|
|
||||||
|
```python
|
||||||
|
import base64
|
||||||
|
import mimetypes
|
||||||
|
|
||||||
|
async def upload_attachment(
|
||||||
|
agent: Agent,
|
||||||
|
auth: Authenticator,
|
||||||
|
document: str,
|
||||||
|
filename: str,
|
||||||
|
content_base64: str,
|
||||||
|
content_type: str | None = None,
|
||||||
|
client: GristClient | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Upload a file attachment to a document."""
|
||||||
|
auth.authorize(agent, document, Permission.WRITE)
|
||||||
|
|
||||||
|
# Decode base64
|
||||||
|
try:
|
||||||
|
content = base64.b64decode(content_base64)
|
||||||
|
except Exception:
|
||||||
|
raise ValueError("Invalid base64 encoding")
|
||||||
|
|
||||||
|
# Auto-detect MIME type if not provided
|
||||||
|
if content_type is None:
|
||||||
|
content_type, _ = mimetypes.guess_type(filename)
|
||||||
|
if content_type is None:
|
||||||
|
content_type = "application/octet-stream"
|
||||||
|
|
||||||
|
if client is None:
|
||||||
|
doc = auth.get_document(document)
|
||||||
|
client = GristClient(doc)
|
||||||
|
|
||||||
|
return await client.upload_attachment(filename, content, content_type)
|
||||||
|
```
|
||||||
|
|
||||||
|
## Error Handling
|
||||||
|
|
||||||
|
| Error | Cause | Response |
|
||||||
|
|-------|-------|----------|
|
||||||
|
| Invalid base64 | Malformed content_base64 | `ValueError: Invalid base64 encoding` |
|
||||||
|
| Authorization | Agent lacks write permission | `AuthError` (existing pattern) |
|
||||||
|
| Grist API error | Upload fails | `httpx.HTTPStatusError` (existing pattern) |
|
||||||
|
|
||||||
|
## Testing
|
||||||
|
|
||||||
|
### Unit Tests
|
||||||
|
|
||||||
|
**`tests/unit/test_tools_write.py`:**
|
||||||
|
- `test_upload_attachment_success` - Valid base64, returns attachment_id
|
||||||
|
- `test_upload_attachment_invalid_base64` - Raises ValueError
|
||||||
|
- `test_upload_attachment_auth_required` - Verifies write permission check
|
||||||
|
- `test_upload_attachment_mime_detection` - Auto-detects type from filename
|
||||||
|
|
||||||
|
**`tests/unit/test_grist_client.py`:**
|
||||||
|
- `test_upload_attachment_api_call` - Correct multipart request format
|
||||||
|
- `test_upload_attachment_with_explicit_content_type` - Passes through MIME type
|
||||||
|
|
||||||
|
### Mock Approach
|
||||||
|
|
||||||
|
Mock `httpx.AsyncClient` responses; no Grist server needed for unit tests.
|
||||||
|
|
||||||
|
## Future Considerations
|
||||||
|
|
||||||
|
Not included in this implementation (YAGNI):
|
||||||
|
- Batch upload (multiple files)
|
||||||
|
- Download attachment
|
||||||
|
- Proxy API support
|
||||||
|
- Size limit validation (rely on Grist's limits)
|
||||||
|
|
||||||
|
These can be added if real use cases emerge.
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[project]
|
[project]
|
||||||
name = "grist-mcp"
|
name = "grist-mcp"
|
||||||
version = "1.2.0"
|
version = "1.3.0"
|
||||||
description = "MCP server for AI agents to interact with Grist documents"
|
description = "MCP server for AI agents to interact with Grist documents"
|
||||||
requires-python = ">=3.14"
|
requires-python = ">=3.14"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -116,6 +116,39 @@ class GristClient:
|
|||||||
"""Delete records by ID."""
|
"""Delete records by ID."""
|
||||||
await self._request("POST", f"/tables/{table}/data/delete", json=record_ids)
|
await self._request("POST", f"/tables/{table}/data/delete", json=record_ids)
|
||||||
|
|
||||||
|
async def upload_attachment(
|
||||||
|
self,
|
||||||
|
filename: str,
|
||||||
|
content: bytes,
|
||||||
|
content_type: str = "application/octet-stream",
|
||||||
|
) -> dict:
|
||||||
|
"""Upload a file attachment. Returns attachment metadata.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
filename: Name for the uploaded file.
|
||||||
|
content: File content as bytes.
|
||||||
|
content_type: MIME type of the file.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict with attachment_id, filename, and size_bytes.
|
||||||
|
"""
|
||||||
|
files = {"upload": (filename, content, content_type)}
|
||||||
|
|
||||||
|
async with httpx.AsyncClient(timeout=self._timeout) as client:
|
||||||
|
response = await client.post(
|
||||||
|
f"{self._base_url}/attachments",
|
||||||
|
headers=self._headers,
|
||||||
|
files=files,
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
# Grist returns list of attachment IDs
|
||||||
|
attachment_ids = response.json()
|
||||||
|
return {
|
||||||
|
"attachment_id": attachment_ids[0],
|
||||||
|
"filename": filename,
|
||||||
|
"size_bytes": len(content),
|
||||||
|
}
|
||||||
|
|
||||||
# Schema operations
|
# Schema operations
|
||||||
|
|
||||||
async def create_table(self, table_id: str, columns: list[dict]) -> str:
|
async def create_table(self, table_id: str, columns: list[dict]) -> str:
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ from grist_mcp.tools.read import sql_query as _sql_query
|
|||||||
from grist_mcp.tools.write import add_records as _add_records
|
from grist_mcp.tools.write import add_records as _add_records
|
||||||
from grist_mcp.tools.write import update_records as _update_records
|
from grist_mcp.tools.write import update_records as _update_records
|
||||||
from grist_mcp.tools.write import delete_records as _delete_records
|
from grist_mcp.tools.write import delete_records as _delete_records
|
||||||
|
from grist_mcp.tools.write import upload_attachment as _upload_attachment
|
||||||
from grist_mcp.tools.schema import create_table as _create_table
|
from grist_mcp.tools.schema import create_table as _create_table
|
||||||
from grist_mcp.tools.schema import add_column as _add_column
|
from grist_mcp.tools.schema import add_column as _add_column
|
||||||
from grist_mcp.tools.schema import modify_column as _modify_column
|
from grist_mcp.tools.schema import modify_column as _modify_column
|
||||||
@@ -218,6 +219,32 @@ def create_server(
|
|||||||
"required": ["document", "table", "column_id"],
|
"required": ["document", "table", "column_id"],
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
|
Tool(
|
||||||
|
name="upload_attachment",
|
||||||
|
description="Upload a file attachment to a Grist document. Returns attachment ID for linking to records via update_records.",
|
||||||
|
inputSchema={
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"document": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "Document name",
|
||||||
|
},
|
||||||
|
"filename": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "Filename with extension (e.g., 'invoice.pdf')",
|
||||||
|
},
|
||||||
|
"content_base64": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "File content as base64-encoded string",
|
||||||
|
},
|
||||||
|
"content_type": {
|
||||||
|
"type": "string",
|
||||||
|
"description": "MIME type (optional, auto-detected from filename)",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"required": ["document", "filename", "content_base64"],
|
||||||
|
},
|
||||||
|
),
|
||||||
Tool(
|
Tool(
|
||||||
name="get_proxy_documentation",
|
name="get_proxy_documentation",
|
||||||
description="Get complete documentation for the HTTP proxy API",
|
description="Get complete documentation for the HTTP proxy API",
|
||||||
@@ -324,6 +351,12 @@ def create_server(
|
|||||||
_current_agent, auth, arguments["document"], arguments["table"],
|
_current_agent, auth, arguments["document"], arguments["table"],
|
||||||
arguments["column_id"],
|
arguments["column_id"],
|
||||||
)
|
)
|
||||||
|
elif name == "upload_attachment":
|
||||||
|
result = await _upload_attachment(
|
||||||
|
_current_agent, auth, arguments["document"],
|
||||||
|
arguments["filename"], arguments["content_base64"],
|
||||||
|
content_type=arguments.get("content_type"),
|
||||||
|
)
|
||||||
elif name == "get_proxy_documentation":
|
elif name == "get_proxy_documentation":
|
||||||
result = await _get_proxy_documentation()
|
result = await _get_proxy_documentation()
|
||||||
elif name == "request_session_token":
|
elif name == "request_session_token":
|
||||||
|
|||||||
@@ -1,4 +1,7 @@
|
|||||||
"""Write tools - create, update, delete records."""
|
"""Write tools - create, update, delete records, upload attachments."""
|
||||||
|
|
||||||
|
import base64
|
||||||
|
import mimetypes
|
||||||
|
|
||||||
from grist_mcp.auth import Agent, Authenticator, Permission
|
from grist_mcp.auth import Agent, Authenticator, Permission
|
||||||
from grist_mcp.grist_client import GristClient
|
from grist_mcp.grist_client import GristClient
|
||||||
@@ -59,3 +62,50 @@ async def delete_records(
|
|||||||
|
|
||||||
await client.delete_records(table, record_ids)
|
await client.delete_records(table, record_ids)
|
||||||
return {"deleted": True}
|
return {"deleted": True}
|
||||||
|
|
||||||
|
|
||||||
|
async def upload_attachment(
|
||||||
|
agent: Agent,
|
||||||
|
auth: Authenticator,
|
||||||
|
document: str,
|
||||||
|
filename: str,
|
||||||
|
content_base64: str,
|
||||||
|
content_type: str | None = None,
|
||||||
|
client: GristClient | None = None,
|
||||||
|
) -> dict:
|
||||||
|
"""Upload a file attachment to a document.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
agent: The authenticated agent.
|
||||||
|
auth: Authenticator for permission checks.
|
||||||
|
document: Document name.
|
||||||
|
filename: Filename with extension.
|
||||||
|
content_base64: File content as base64-encoded string.
|
||||||
|
content_type: MIME type (auto-detected from filename if omitted).
|
||||||
|
client: Optional GristClient instance.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict with attachment_id, filename, and size_bytes.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: If content_base64 is not valid base64.
|
||||||
|
"""
|
||||||
|
auth.authorize(agent, document, Permission.WRITE)
|
||||||
|
|
||||||
|
# Decode base64 content
|
||||||
|
try:
|
||||||
|
content = base64.b64decode(content_base64)
|
||||||
|
except Exception:
|
||||||
|
raise ValueError("Invalid base64 encoding")
|
||||||
|
|
||||||
|
# Auto-detect MIME type if not provided
|
||||||
|
if content_type is None:
|
||||||
|
content_type, _ = mimetypes.guess_type(filename)
|
||||||
|
if content_type is None:
|
||||||
|
content_type = "application/octet-stream"
|
||||||
|
|
||||||
|
if client is None:
|
||||||
|
doc = auth.get_document(document)
|
||||||
|
client = GristClient(doc)
|
||||||
|
|
||||||
|
return await client.upload_attachment(filename, content, content_type)
|
||||||
|
|||||||
@@ -196,3 +196,43 @@ def test_sql_validation_rejects_multiple_statements(client):
|
|||||||
def test_sql_validation_allows_trailing_semicolon(client):
|
def test_sql_validation_allows_trailing_semicolon(client):
|
||||||
# Should not raise
|
# Should not raise
|
||||||
client._validate_sql_query("SELECT * FROM users;")
|
client._validate_sql_query("SELECT * FROM users;")
|
||||||
|
|
||||||
|
|
||||||
|
# Attachment tests
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment(client, httpx_mock: HTTPXMock):
|
||||||
|
httpx_mock.add_response(
|
||||||
|
url="https://grist.example.com/api/docs/abc123/attachments",
|
||||||
|
method="POST",
|
||||||
|
json=[42],
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await client.upload_attachment(
|
||||||
|
filename="invoice.pdf",
|
||||||
|
content=b"PDF content here",
|
||||||
|
content_type="application/pdf",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result == {
|
||||||
|
"attachment_id": 42,
|
||||||
|
"filename": "invoice.pdf",
|
||||||
|
"size_bytes": 16,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_default_content_type(client, httpx_mock: HTTPXMock):
|
||||||
|
httpx_mock.add_response(
|
||||||
|
url="https://grist.example.com/api/docs/abc123/attachments",
|
||||||
|
method="POST",
|
||||||
|
json=[99],
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await client.upload_attachment(
|
||||||
|
filename="data.bin",
|
||||||
|
content=b"\x00\x01\x02",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result["attachment_id"] == 99
|
||||||
|
assert result["size_bytes"] == 3
|
||||||
|
|||||||
@@ -52,13 +52,14 @@ tokens:
|
|||||||
assert "add_column" in tool_names
|
assert "add_column" in tool_names
|
||||||
assert "modify_column" in tool_names
|
assert "modify_column" in tool_names
|
||||||
assert "delete_column" in tool_names
|
assert "delete_column" in tool_names
|
||||||
|
assert "upload_attachment" in tool_names
|
||||||
|
|
||||||
# Session tools (always registered)
|
# Session tools (always registered)
|
||||||
assert "get_proxy_documentation" in tool_names
|
assert "get_proxy_documentation" in tool_names
|
||||||
assert "request_session_token" in tool_names
|
assert "request_session_token" in tool_names
|
||||||
|
|
||||||
# Should have all 14 tools
|
# Should have all 15 tools
|
||||||
assert len(result.root.tools) == 14
|
assert len(result.root.tools) == 15
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
|
import base64
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from unittest.mock import AsyncMock
|
from unittest.mock import AsyncMock
|
||||||
|
|
||||||
from grist_mcp.tools.write import add_records, update_records, delete_records
|
from grist_mcp.tools.write import add_records, update_records, delete_records, upload_attachment
|
||||||
from grist_mcp.auth import Authenticator, AuthError
|
from grist_mcp.auth import Authenticator, AuthError
|
||||||
from grist_mcp.config import Config, Document, Token, TokenScope
|
from grist_mcp.config import Config, Document, Token, TokenScope
|
||||||
|
|
||||||
@@ -94,3 +96,105 @@ async def test_delete_records(auth, mock_client):
|
|||||||
)
|
)
|
||||||
|
|
||||||
assert result == {"deleted": True}
|
assert result == {"deleted": True}
|
||||||
|
|
||||||
|
|
||||||
|
# Upload attachment tests
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_client_with_attachment():
|
||||||
|
client = AsyncMock()
|
||||||
|
client.upload_attachment.return_value = {
|
||||||
|
"attachment_id": 42,
|
||||||
|
"filename": "invoice.pdf",
|
||||||
|
"size_bytes": 1024,
|
||||||
|
}
|
||||||
|
return client
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_success(auth, mock_client_with_attachment):
|
||||||
|
agent = auth.authenticate("write-token")
|
||||||
|
content = b"PDF content"
|
||||||
|
content_base64 = base64.b64encode(content).decode()
|
||||||
|
|
||||||
|
result = await upload_attachment(
|
||||||
|
agent, auth, "budget",
|
||||||
|
filename="invoice.pdf",
|
||||||
|
content_base64=content_base64,
|
||||||
|
client=mock_client_with_attachment,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result == {
|
||||||
|
"attachment_id": 42,
|
||||||
|
"filename": "invoice.pdf",
|
||||||
|
"size_bytes": 1024,
|
||||||
|
}
|
||||||
|
mock_client_with_attachment.upload_attachment.assert_called_once_with(
|
||||||
|
"invoice.pdf", content, "application/pdf"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_invalid_base64(auth, mock_client_with_attachment):
|
||||||
|
agent = auth.authenticate("write-token")
|
||||||
|
|
||||||
|
with pytest.raises(ValueError, match="Invalid base64 encoding"):
|
||||||
|
await upload_attachment(
|
||||||
|
agent, auth, "budget",
|
||||||
|
filename="test.txt",
|
||||||
|
content_base64="not-valid-base64!!!",
|
||||||
|
client=mock_client_with_attachment,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_auth_required(auth, mock_client_with_attachment):
|
||||||
|
agent = auth.authenticate("read-token")
|
||||||
|
content_base64 = base64.b64encode(b"test").decode()
|
||||||
|
|
||||||
|
with pytest.raises(AuthError, match="Permission denied"):
|
||||||
|
await upload_attachment(
|
||||||
|
agent, auth, "budget",
|
||||||
|
filename="test.txt",
|
||||||
|
content_base64=content_base64,
|
||||||
|
client=mock_client_with_attachment,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_mime_detection(auth, mock_client_with_attachment):
|
||||||
|
agent = auth.authenticate("write-token")
|
||||||
|
content = b"PNG content"
|
||||||
|
content_base64 = base64.b64encode(content).decode()
|
||||||
|
|
||||||
|
await upload_attachment(
|
||||||
|
agent, auth, "budget",
|
||||||
|
filename="image.png",
|
||||||
|
content_base64=content_base64,
|
||||||
|
client=mock_client_with_attachment,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Should auto-detect image/png from filename
|
||||||
|
mock_client_with_attachment.upload_attachment.assert_called_once_with(
|
||||||
|
"image.png", content, "image/png"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_upload_attachment_explicit_content_type(auth, mock_client_with_attachment):
|
||||||
|
agent = auth.authenticate("write-token")
|
||||||
|
content = b"custom content"
|
||||||
|
content_base64 = base64.b64encode(content).decode()
|
||||||
|
|
||||||
|
await upload_attachment(
|
||||||
|
agent, auth, "budget",
|
||||||
|
filename="file.dat",
|
||||||
|
content_base64=content_base64,
|
||||||
|
content_type="application/custom",
|
||||||
|
client=mock_client_with_attachment,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Should use explicit content type
|
||||||
|
mock_client_with_attachment.upload_attachment.assert_called_once_with(
|
||||||
|
"file.dat", content, "application/custom"
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user