feat(caldav): Add support for tasks
This commit is contained in:
Chris Coutinho
@@ -136,7 +136,17 @@ Each Nextcloud app has a corresponding server module that:
|
||||
### Supported Nextcloud Apps
|
||||
|
||||
- **Notes** - Full CRUD operations and search
|
||||
- **Calendar** - CalDAV integration with events, recurring events, attendees
|
||||
- **Calendar** - CalDAV integration with events, recurring events, attendees, and **tasks (VTODO)**
|
||||
- **Calendar Operations**: List, create, delete calendars
|
||||
- **Event Operations**: Full CRUD, recurring events, attendees, reminders, bulk operations
|
||||
- **Task Operations (VTODO)**: Full CRUD for CalDAV tasks with:
|
||||
- Status tracking (NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED)
|
||||
- Priority levels (0-9, 1=highest, 9=lowest)
|
||||
- Due dates, start dates, completion tracking
|
||||
- Percent complete (0-100%)
|
||||
- Categories and filtering
|
||||
- Search across all calendars
|
||||
- **Note**: Calendar implementation uses caldav library's AsyncDavClient
|
||||
- **Contacts** - CardDAV integration with address book operations
|
||||
- **Tables** - Row-level operations on Nextcloud Tables
|
||||
- **WebDAV** - Complete file system access
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
FROM ghcr.io/astral-sh/uv:0.9.4-python3.11-alpine@sha256:1a51c7710eaf839fa3365329ad993b48d17ddd9ab0f0672efaa9b09f407ebf44
|
||||
|
||||
# Install git (required for caldav dependency from git)
|
||||
RUN apk add --no-cache git
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
COPY . .
|
||||
|
||||
@@ -11,9 +11,12 @@ php /var/www/html/occ app:enable calendar
|
||||
echo "Waiting for calendar app to initialize..."
|
||||
sleep 5
|
||||
|
||||
# Increase limits on calendar creation for integration tests (100 in 60s)
|
||||
php occ config:app:set dav rateLimitCalendarCreation --type=integer --value=100
|
||||
php occ config:app:set dav rateLimitPeriodCalendarCreation --type=integer --value=60
|
||||
# Disable rate limits on calendar creation for integration tests
|
||||
# Set to -1 to completely disable rate limiting
|
||||
# Reference: https://docs.nextcloud.com/server/stable/admin_manual/groupware/calendar.html#rate-limits
|
||||
php occ config:app:set dav rateLimitCalendarCreation --type=integer --value=-1
|
||||
php occ config:app:set dav rateLimitPeriodCalendarCreation --type=integer --value=-1
|
||||
php occ config:app:set dav maximumCalendarsSubscriptions --type=integer --value=-1
|
||||
|
||||
# Ensure maintenance mode is off before calendar operations
|
||||
php /var/www/html/occ maintenance:mode --off
|
||||
|
||||
@@ -14,12 +14,19 @@ services:
|
||||
- MYSQL_DATABASE=nextcloud
|
||||
- MYSQL_USER=nextcloud
|
||||
|
||||
# Note: Redis is an external service. You can find more information about the configuration here:
|
||||
# https://hub.docker.com/_/redis
|
||||
redis:
|
||||
image: docker.io/library/redis:alpine@sha256:59b6e694653476de2c992937ebe1c64182af4728e54bb49e9b7a6c26614d8933
|
||||
restart: always
|
||||
|
||||
app:
|
||||
image: docker.io/library/nextcloud:32.0.0@sha256:3e70e4dfe882ef44738fdc30d9896fb07c12febb27c4a1177e3d63dc0004a0b4
|
||||
restart: always
|
||||
ports:
|
||||
- 0.0.0.0:8080:80
|
||||
depends_on:
|
||||
- redis
|
||||
- db
|
||||
volumes:
|
||||
- nextcloud:/var/www/html
|
||||
@@ -32,6 +39,7 @@ services:
|
||||
- MYSQL_DATABASE=nextcloud
|
||||
- MYSQL_USER=nextcloud
|
||||
- MYSQL_HOST=db
|
||||
- REDIS_HOST=redis
|
||||
|
||||
recipes:
|
||||
image: docker.io/library/nginx:alpine@sha256:61e01287e546aac28a3f56839c136b31f590273f3b41187a36f46f6a03bbfe22
|
||||
|
||||
@@ -72,7 +72,9 @@ class NextcloudClient:
|
||||
self.notes = NotesClient(self._client, username)
|
||||
self.webdav = WebDAVClient(self._client, username)
|
||||
self.tables = TablesClient(self._client, username)
|
||||
self.calendar = CalendarClient(self._client, username)
|
||||
self.calendar = CalendarClient(
|
||||
base_url, username, auth
|
||||
) # Uses AsyncDavClient internally
|
||||
self.contacts = ContactsClient(self._client, username)
|
||||
self.cookbook = CookbookClient(self._client, username)
|
||||
self.deck = DeckClient(self._client, username)
|
||||
@@ -129,5 +131,6 @@ class NextcloudClient:
|
||||
return f"/remote.php/dav/files/{self.username}"
|
||||
|
||||
async def close(self):
|
||||
"""Close the HTTP client."""
|
||||
"""Close the HTTP client and CalDAV client."""
|
||||
await self._client.aclose()
|
||||
await self.calendar.close()
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -180,3 +180,71 @@ class ManageCalendarResponse(BaseResponse):
|
||||
None, description="List of calendars (for list action)"
|
||||
)
|
||||
message: str = Field(description="Success message")
|
||||
|
||||
|
||||
# ============= Todo/Task Models =============
|
||||
|
||||
|
||||
class Todo(BaseModel):
|
||||
"""Model for a CalDAV todo/task (VTODO)."""
|
||||
|
||||
uid: str = Field(description="Todo UID")
|
||||
summary: str = Field(description="Todo summary/title")
|
||||
description: str = Field(default="", description="Todo description")
|
||||
status: str = Field(
|
||||
default="NEEDS-ACTION",
|
||||
description="Todo status: NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED",
|
||||
)
|
||||
priority: int = Field(
|
||||
default=0, description="Todo priority (0=undefined, 1=highest, 9=lowest)"
|
||||
)
|
||||
percent_complete: int = Field(default=0, description="Percentage complete (0-100)")
|
||||
due: Optional[str] = Field(None, description="Due date/time (ISO format)")
|
||||
dtstart: Optional[str] = Field(None, description="Start date/time (ISO format)")
|
||||
completed: Optional[str] = Field(
|
||||
None, description="Completion timestamp (ISO format)"
|
||||
)
|
||||
categories: str = Field(default="", description="Comma-separated categories")
|
||||
href: str = Field(default="", description="CalDAV href")
|
||||
etag: str = Field(default="", description="ETag for versioning")
|
||||
calendar_name: Optional[str] = Field(
|
||||
None, description="Calendar containing this todo"
|
||||
)
|
||||
calendar_display_name: Optional[str] = Field(
|
||||
None, description="Display name of calendar containing this todo"
|
||||
)
|
||||
|
||||
|
||||
class ListTodosResponse(BaseResponse):
|
||||
"""Response model for listing todos."""
|
||||
|
||||
todos: List[Todo] = Field(description="List of todos/tasks")
|
||||
calendar_name: Optional[str] = Field(
|
||||
None, description="Calendar name (if filtered to one calendar)"
|
||||
)
|
||||
total_count: int = Field(description="Total number of todos found")
|
||||
|
||||
|
||||
class CreateTodoResponse(BaseResponse):
|
||||
"""Response model for todo creation."""
|
||||
|
||||
todo: Todo = Field(description="The created todo")
|
||||
calendar_name: str = Field(
|
||||
description="Name of the calendar the todo was created in"
|
||||
)
|
||||
|
||||
|
||||
class UpdateTodoResponse(BaseResponse):
|
||||
"""Response model for todo updates."""
|
||||
|
||||
todo: Todo = Field(description="The updated todo")
|
||||
calendar_name: str = Field(description="Name of the calendar the todo belongs to")
|
||||
|
||||
|
||||
class DeleteTodoResponse(StatusResponse):
|
||||
"""Response model for todo deletion."""
|
||||
|
||||
deleted_uid: str = Field(description="UID of the deleted todo")
|
||||
calendar_name: str = Field(
|
||||
description="Name of the calendar the todo was deleted from"
|
||||
)
|
||||
|
||||
@@ -5,7 +5,12 @@ from typing import Optional
|
||||
from mcp.server.fastmcp import Context, FastMCP
|
||||
|
||||
from nextcloud_mcp_server.context import get_client
|
||||
from nextcloud_mcp_server.models.calendar import Calendar, ListCalendarsResponse
|
||||
from nextcloud_mcp_server.models.calendar import (
|
||||
Calendar,
|
||||
ListCalendarsResponse,
|
||||
ListTodosResponse,
|
||||
Todo,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -796,3 +801,209 @@ def configure_calendar_tools(mcp: FastMCP):
|
||||
|
||||
else:
|
||||
raise ValueError("Action must be 'create', 'delete', 'update', or 'list'")
|
||||
|
||||
# ============= Todo/Task Tools =============
|
||||
|
||||
@mcp.tool()
|
||||
async def nc_calendar_list_todos(
|
||||
calendar_name: str,
|
||||
ctx: Context,
|
||||
status: Optional[str] = None,
|
||||
min_priority: Optional[int] = None,
|
||||
categories: Optional[str] = None,
|
||||
summary_contains: Optional[str] = None,
|
||||
) -> ListTodosResponse:
|
||||
"""List todos/tasks in a calendar with optional filtering.
|
||||
|
||||
Args:
|
||||
calendar_name: Name of the calendar to list todos from
|
||||
ctx: MCP context
|
||||
status: Filter by status (NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED)
|
||||
min_priority: Filter by minimum priority (1=highest, 9=lowest)
|
||||
categories: Filter by categories (comma-separated, e.g., "work,urgent")
|
||||
summary_contains: Filter todos where summary contains this text
|
||||
|
||||
Returns:
|
||||
List of todos matching the filters
|
||||
"""
|
||||
client = get_client(ctx)
|
||||
|
||||
# Build filters dictionary
|
||||
filters = {}
|
||||
if status is not None:
|
||||
filters["status"] = status
|
||||
if min_priority is not None:
|
||||
filters["min_priority"] = min_priority
|
||||
if categories is not None:
|
||||
filters["categories"] = [cat.strip() for cat in categories.split(",")]
|
||||
if summary_contains is not None:
|
||||
filters["summary_contains"] = summary_contains
|
||||
|
||||
todos_data = await client.calendar.list_todos(
|
||||
calendar_name, filters if filters else None
|
||||
)
|
||||
|
||||
todos = [Todo(**todo_data) for todo_data in todos_data]
|
||||
return ListTodosResponse(
|
||||
todos=todos, calendar_name=calendar_name, total_count=len(todos)
|
||||
)
|
||||
|
||||
@mcp.tool()
|
||||
async def nc_calendar_create_todo(
|
||||
calendar_name: str,
|
||||
summary: str,
|
||||
ctx: Context,
|
||||
description: str = "",
|
||||
status: str = "NEEDS-ACTION",
|
||||
priority: int = 0,
|
||||
due: str = "",
|
||||
dtstart: str = "",
|
||||
categories: str = "",
|
||||
):
|
||||
"""Create a new todo/task in a calendar.
|
||||
|
||||
Args:
|
||||
calendar_name: Name of the calendar to create the todo in
|
||||
summary: Todo title/summary
|
||||
ctx: MCP context
|
||||
description: Detailed description of the todo
|
||||
status: Todo status (NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED)
|
||||
priority: Priority (0=undefined, 1=highest, 9=lowest)
|
||||
due: Due date/time (ISO format, e.g., "2025-01-15T14:00:00")
|
||||
dtstart: Start date/time (ISO format)
|
||||
categories: Comma-separated categories (e.g., "work,urgent")
|
||||
|
||||
Returns:
|
||||
Dict with todo creation result
|
||||
"""
|
||||
client = get_client(ctx)
|
||||
|
||||
todo_data = {
|
||||
"summary": summary,
|
||||
"description": description,
|
||||
"status": status,
|
||||
"priority": priority,
|
||||
"due": due,
|
||||
"dtstart": dtstart,
|
||||
"categories": categories,
|
||||
}
|
||||
|
||||
return await client.calendar.create_todo(calendar_name, todo_data)
|
||||
|
||||
@mcp.tool()
|
||||
async def nc_calendar_update_todo(
|
||||
calendar_name: str,
|
||||
todo_uid: str,
|
||||
ctx: Context,
|
||||
summary: Optional[str] = None,
|
||||
description: Optional[str] = None,
|
||||
status: Optional[str] = None,
|
||||
priority: Optional[int] = None,
|
||||
percent_complete: Optional[int] = None,
|
||||
due: Optional[str] = None,
|
||||
dtstart: Optional[str] = None,
|
||||
completed: Optional[str] = None,
|
||||
categories: Optional[str] = None,
|
||||
):
|
||||
"""Update an existing todo/task.
|
||||
|
||||
Args:
|
||||
calendar_name: Name of the calendar containing the todo
|
||||
todo_uid: UID of the todo to update
|
||||
ctx: MCP context
|
||||
summary: New summary/title
|
||||
description: New description
|
||||
status: New status (NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED)
|
||||
priority: New priority (0-9)
|
||||
percent_complete: New completion percentage (0-100)
|
||||
due: New due date/time (ISO format)
|
||||
dtstart: New start date/time (ISO format)
|
||||
completed: Completion timestamp (ISO format)
|
||||
categories: New categories (comma-separated)
|
||||
|
||||
Returns:
|
||||
Dict with todo update result
|
||||
"""
|
||||
client = get_client(ctx)
|
||||
|
||||
# Build update data with only non-None values
|
||||
todo_data = {}
|
||||
if summary is not None:
|
||||
todo_data["summary"] = summary
|
||||
if description is not None:
|
||||
todo_data["description"] = description
|
||||
if status is not None:
|
||||
todo_data["status"] = status
|
||||
if priority is not None:
|
||||
todo_data["priority"] = priority
|
||||
if percent_complete is not None:
|
||||
todo_data["percent_complete"] = percent_complete
|
||||
if due is not None:
|
||||
todo_data["due"] = due
|
||||
if dtstart is not None:
|
||||
todo_data["dtstart"] = dtstart
|
||||
if completed is not None:
|
||||
todo_data["completed"] = completed
|
||||
if categories is not None:
|
||||
todo_data["categories"] = categories
|
||||
|
||||
return await client.calendar.update_todo(calendar_name, todo_uid, todo_data)
|
||||
|
||||
@mcp.tool()
|
||||
async def nc_calendar_delete_todo(
|
||||
calendar_name: str,
|
||||
todo_uid: str,
|
||||
ctx: Context,
|
||||
):
|
||||
"""Delete a todo/task from a calendar.
|
||||
|
||||
Args:
|
||||
calendar_name: Name of the calendar containing the todo
|
||||
todo_uid: UID of the todo to delete
|
||||
ctx: MCP context
|
||||
|
||||
Returns:
|
||||
Dict with deletion status
|
||||
"""
|
||||
client = get_client(ctx)
|
||||
return await client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
|
||||
@mcp.tool()
|
||||
async def nc_calendar_search_todos(
|
||||
ctx: Context,
|
||||
status: Optional[str] = None,
|
||||
min_priority: Optional[int] = None,
|
||||
categories: Optional[str] = None,
|
||||
summary_contains: Optional[str] = None,
|
||||
):
|
||||
"""Search todos across all calendars with optional filtering.
|
||||
|
||||
Args:
|
||||
ctx: MCP context
|
||||
status: Filter by status (NEEDS-ACTION, IN-PROCESS, COMPLETED, CANCELLED)
|
||||
min_priority: Filter by minimum priority (1=highest, 9=lowest)
|
||||
categories: Filter by categories (comma-separated, e.g., "work,urgent")
|
||||
summary_contains: Filter todos where summary contains this text
|
||||
|
||||
Returns:
|
||||
List of todos matching the filters from all calendars
|
||||
"""
|
||||
client = get_client(ctx)
|
||||
|
||||
# Build filters dictionary
|
||||
filters = {}
|
||||
if status is not None:
|
||||
filters["status"] = status
|
||||
if min_priority is not None:
|
||||
filters["min_priority"] = min_priority
|
||||
if categories is not None:
|
||||
filters["categories"] = [cat.strip() for cat in categories.split(",")]
|
||||
if summary_contains is not None:
|
||||
filters["summary_contains"] = summary_contains
|
||||
|
||||
todos_data = await client.calendar.search_todos_across_calendars(
|
||||
filters if filters else None
|
||||
)
|
||||
|
||||
todos = [Todo(**todo_data) for todo_data in todos_data]
|
||||
return ListTodosResponse(todos=todos, total_count=len(todos))
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Shared fixtures for calendar integration tests.
|
||||
|
||||
Note: The temporary_calendar fixture is defined in tests/conftest.py and uses
|
||||
a shared session-scoped calendar to avoid Nextcloud rate limiting issues.
|
||||
This conftest.py exists for any calendar-specific fixtures that might be needed
|
||||
in the future.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -0,0 +1,498 @@
|
||||
"""Integration tests for Calendar VTODO (task) operations."""
|
||||
|
||||
import logging
|
||||
import uuid
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from httpx import HTTPStatusError
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mark all tests in this module as integration tests
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def temporary_todo(nc_client: NextcloudClient, temporary_calendar: str):
|
||||
"""Create a temporary todo for testing and clean up afterward."""
|
||||
todo_uid = None
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
# Create a test todo
|
||||
tomorrow = datetime.now() + timedelta(days=1)
|
||||
todo_data = {
|
||||
"summary": f"Test Task {uuid.uuid4().hex[:8]}",
|
||||
"description": "Test todo created by integration tests",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 5,
|
||||
"due": tomorrow.strftime("%Y-%m-%dT18:00:00"),
|
||||
"categories": "testing",
|
||||
}
|
||||
|
||||
try:
|
||||
logger.info(f"Creating temporary todo in calendar: {calendar_name}")
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uid = result.get("uid")
|
||||
|
||||
if not todo_uid:
|
||||
pytest.fail("Failed to create temporary todo")
|
||||
|
||||
logger.info(f"Created temporary todo with UID: {todo_uid}")
|
||||
yield {"uid": todo_uid, "calendar_name": calendar_name, "data": todo_data}
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
if todo_uid:
|
||||
try:
|
||||
logger.info(f"Cleaning up temporary todo: {todo_uid}")
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
logger.info(f"Successfully deleted temporary todo: {todo_uid}")
|
||||
except HTTPStatusError as e:
|
||||
if e.response.status_code != 404:
|
||||
logger.error(f"Error deleting temporary todo {todo_uid}: {e}")
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Unexpected error deleting temporary todo {todo_uid}: {e}"
|
||||
)
|
||||
|
||||
|
||||
# ============= Basic CRUD Tests =============
|
||||
|
||||
|
||||
async def test_create_and_delete_todo(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test creating and deleting a basic todo."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
# Create todo
|
||||
tomorrow = datetime.now() + timedelta(days=1)
|
||||
todo_data = {
|
||||
"summary": "Integration Test Task",
|
||||
"description": "Test task for integration testing",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 3,
|
||||
"due": tomorrow.strftime("%Y-%m-%dT18:00:00"),
|
||||
"categories": "testing,integration",
|
||||
}
|
||||
|
||||
try:
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
assert "uid" in result
|
||||
assert result["status_code"] in [200, 201, 204]
|
||||
|
||||
todo_uid = result["uid"]
|
||||
logger.info(f"Created todo with UID: {todo_uid}")
|
||||
|
||||
# Verify todo was created by listing todos
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
todo_uids = [todo.get("uid") for todo in todos]
|
||||
assert todo_uid in todo_uids
|
||||
|
||||
# Find our todo in the list
|
||||
our_todo = next((t for t in todos if t.get("uid") == todo_uid), None)
|
||||
assert our_todo is not None
|
||||
assert our_todo["summary"] == "Integration Test Task"
|
||||
assert our_todo["status"] == "NEEDS-ACTION"
|
||||
assert our_todo["priority"] == 3
|
||||
|
||||
# Delete todo
|
||||
delete_result = await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
assert delete_result["status_code"] in [200, 204, 404]
|
||||
|
||||
logger.info(f"Successfully deleted todo: {todo_uid}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Test failed: {e}")
|
||||
raise
|
||||
|
||||
|
||||
async def test_list_todos(nc_client: NextcloudClient, temporary_calendar: str):
|
||||
"""Test listing todos in a calendar."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
# Create multiple todos
|
||||
todo_uids = []
|
||||
for i in range(3):
|
||||
todo_data = {
|
||||
"summary": f"Test Task {i + 1}",
|
||||
"description": f"Task number {i + 1}",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": i + 1,
|
||||
}
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uids.append(result["uid"])
|
||||
|
||||
try:
|
||||
# List todos
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
|
||||
assert isinstance(todos, list)
|
||||
assert len(todos) >= 3 # At least our 3 todos
|
||||
|
||||
# Check structure
|
||||
for todo in todos:
|
||||
assert "uid" in todo
|
||||
assert "summary" in todo
|
||||
assert "status" in todo
|
||||
assert "priority" in todo
|
||||
|
||||
# Verify our todos are in the list
|
||||
listed_uids = [todo["uid"] for todo in todos]
|
||||
for uid in todo_uids:
|
||||
assert uid in listed_uids
|
||||
|
||||
logger.info(f"Found {len(todos)} todos in calendar")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for uid in todo_uids:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_update_todo(nc_client: NextcloudClient, temporary_todo: dict):
|
||||
"""Test updating an existing todo."""
|
||||
calendar_name = temporary_todo["calendar_name"]
|
||||
todo_uid = temporary_todo["uid"]
|
||||
|
||||
# Update todo data
|
||||
updated_data = {
|
||||
"summary": "Updated Test Task Title",
|
||||
"description": "Updated description for test task",
|
||||
"status": "IN-PROCESS",
|
||||
"priority": 1, # High priority
|
||||
"percent_complete": 50,
|
||||
}
|
||||
|
||||
try:
|
||||
result = await nc_client.calendar.update_todo(
|
||||
calendar_name, todo_uid, updated_data
|
||||
)
|
||||
assert result["uid"] == todo_uid
|
||||
|
||||
# Verify updates by listing todos
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
updated_todo = next((t for t in todos if t["uid"] == todo_uid), None)
|
||||
|
||||
assert updated_todo is not None
|
||||
assert updated_todo["summary"] == "Updated Test Task Title"
|
||||
assert updated_todo["description"] == "Updated description for test task"
|
||||
assert updated_todo["status"] == "IN-PROCESS"
|
||||
assert updated_todo["priority"] == 1
|
||||
assert updated_todo["percent_complete"] == 50
|
||||
|
||||
logger.info(f"Successfully updated todo: {todo_uid}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Todo update test failed: {e}")
|
||||
raise
|
||||
|
||||
|
||||
async def test_todo_with_dates(nc_client: NextcloudClient, temporary_calendar: str):
|
||||
"""Test creating a todo with start, due, and completed dates."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
now = datetime.now()
|
||||
start_date = now + timedelta(days=1)
|
||||
due_date = now + timedelta(days=7)
|
||||
|
||||
todo_data = {
|
||||
"summary": "Task with Dates",
|
||||
"description": "Test task with various date fields",
|
||||
"status": "NEEDS-ACTION",
|
||||
"dtstart": start_date.strftime("%Y-%m-%dT09:00:00"),
|
||||
"due": due_date.strftime("%Y-%m-%dT17:00:00"),
|
||||
}
|
||||
|
||||
try:
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uid = result["uid"]
|
||||
logger.info(f"Created todo with dates, UID: {todo_uid}")
|
||||
|
||||
# Verify dates
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
created_todo = next((t for t in todos if t["uid"] == todo_uid), None)
|
||||
|
||||
assert created_todo is not None
|
||||
assert created_todo["summary"] == "Task with Dates"
|
||||
assert "dtstart" in created_todo
|
||||
assert "due" in created_todo
|
||||
|
||||
# Cleanup
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Date handling test failed: {e}")
|
||||
raise
|
||||
|
||||
|
||||
# ============= Advanced Feature Tests =============
|
||||
|
||||
|
||||
async def test_todo_status_transitions(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test transitioning through different todo statuses."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
todo_data = {
|
||||
"summary": "Status Transition Test",
|
||||
"description": "Testing status changes",
|
||||
"status": "NEEDS-ACTION",
|
||||
}
|
||||
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uid = result["uid"]
|
||||
|
||||
try:
|
||||
# Transition: NEEDS-ACTION → IN-PROCESS
|
||||
await nc_client.calendar.update_todo(
|
||||
calendar_name,
|
||||
todo_uid,
|
||||
{"status": "IN-PROCESS", "percent_complete": 25},
|
||||
)
|
||||
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
todo = next((t for t in todos if t["uid"] == todo_uid), None)
|
||||
assert todo["status"] == "IN-PROCESS"
|
||||
assert todo["percent_complete"] == 25
|
||||
|
||||
# Transition: IN-PROCESS → COMPLETED
|
||||
completed_time = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
await nc_client.calendar.update_todo(
|
||||
calendar_name,
|
||||
todo_uid,
|
||||
{
|
||||
"status": "COMPLETED",
|
||||
"percent_complete": 100,
|
||||
"completed": completed_time,
|
||||
},
|
||||
)
|
||||
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
todo = next((t for t in todos if t["uid"] == todo_uid), None)
|
||||
assert todo["status"] == "COMPLETED"
|
||||
assert todo["percent_complete"] == 100
|
||||
assert "completed" in todo
|
||||
|
||||
logger.info(f"Successfully transitioned todo through statuses: {todo_uid}")
|
||||
|
||||
finally:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
|
||||
|
||||
async def test_todo_priority_levels(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test different priority levels (0=undefined, 1=highest, 9=lowest)."""
|
||||
calendar_name = temporary_calendar
|
||||
priorities = [0, 1, 5, 9]
|
||||
priority_labels = {0: "Undefined", 1: "Highest", 5: "Medium", 9: "Lowest"}
|
||||
todo_uids = []
|
||||
|
||||
try:
|
||||
# Create todos with different priorities
|
||||
for priority in priorities:
|
||||
todo_data = {
|
||||
"summary": f"Priority {priority} Task ({priority_labels[priority]})",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": priority,
|
||||
}
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uids.append((result["uid"], priority))
|
||||
|
||||
# Verify all priorities
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
|
||||
for uid, expected_priority in todo_uids:
|
||||
todo = next((t for t in todos if t["uid"] == uid), None)
|
||||
assert todo is not None
|
||||
assert todo["priority"] == expected_priority
|
||||
|
||||
logger.info(f"Successfully tested priority levels: {priorities}")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for uid, _ in todo_uids:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_todo_with_categories(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test creating a todo with multiple categories."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
todo_data = {
|
||||
"summary": "Task with Categories",
|
||||
"description": "Testing category support",
|
||||
"status": "NEEDS-ACTION",
|
||||
"categories": "work,meeting,important,quarterly",
|
||||
}
|
||||
|
||||
try:
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
todo_uid = result["uid"]
|
||||
logger.info(f"Created todo with categories, UID: {todo_uid}")
|
||||
|
||||
# Verify categories
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
created_todo = next((t for t in todos if t["uid"] == todo_uid), None)
|
||||
|
||||
assert created_todo is not None
|
||||
assert "categories" in created_todo
|
||||
categories_str = created_todo["categories"]
|
||||
assert "work" in categories_str
|
||||
assert "meeting" in categories_str
|
||||
assert "important" in categories_str
|
||||
assert "quarterly" in categories_str
|
||||
|
||||
# Cleanup
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Categories test failed: {e}")
|
||||
raise
|
||||
|
||||
|
||||
async def test_search_todos_across_calendars(
|
||||
nc_client: NextcloudClient, temporary_calendar: str, shared_calendar_2: str
|
||||
):
|
||||
"""Test searching for todos across multiple calendars.
|
||||
|
||||
Uses two shared test calendars to avoid rate limiting.
|
||||
"""
|
||||
# Use existing shared calendars to avoid rate limits
|
||||
cal1_name = temporary_calendar # First shared test calendar
|
||||
cal2_name = shared_calendar_2 # Second shared test calendar
|
||||
|
||||
try:
|
||||
# Create todos in both calendars
|
||||
todo1_data = {"summary": "Task in Calendar 1", "status": "NEEDS-ACTION"}
|
||||
todo2_data = {"summary": "Task in Calendar 2", "status": "IN-PROCESS"}
|
||||
|
||||
result1 = await nc_client.calendar.create_todo(cal1_name, todo1_data)
|
||||
result2 = await nc_client.calendar.create_todo(cal2_name, todo2_data)
|
||||
|
||||
# Search across all calendars
|
||||
all_todos = await nc_client.calendar.search_todos_across_calendars()
|
||||
|
||||
assert isinstance(all_todos, list)
|
||||
|
||||
# Find our todos
|
||||
todo1 = next((t for t in all_todos if t["uid"] == result1["uid"]), None)
|
||||
todo2 = next((t for t in all_todos if t["uid"] == result2["uid"]), None)
|
||||
|
||||
assert todo1 is not None
|
||||
assert todo2 is not None
|
||||
assert "calendar_name" in todo1
|
||||
assert "calendar_name" in todo2
|
||||
assert todo1["calendar_name"] == cal1_name
|
||||
assert todo2["calendar_name"] == cal2_name
|
||||
|
||||
logger.info(f"Found {len(all_todos)} todos across all calendars")
|
||||
|
||||
finally:
|
||||
# Cleanup: Delete only the todos we created (calendars are reused/built-in)
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(cal1_name, result1["uid"])
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(cal2_name, result2["uid"])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ============= Edge Case Tests =============
|
||||
|
||||
|
||||
async def test_get_nonexistent_todo(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test attempting to retrieve a non-existent todo."""
|
||||
calendar_name = temporary_calendar
|
||||
fake_uid = f"nonexistent-{uuid.uuid4()}"
|
||||
|
||||
# List todos to ensure it doesn't exist
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
matching_todos = [t for t in todos if t.get("uid") == fake_uid]
|
||||
assert len(matching_todos) == 0
|
||||
|
||||
logger.info(f"Verified nonexistent todo UID: {fake_uid}")
|
||||
|
||||
|
||||
async def test_delete_nonexistent_todo(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test deleting a non-existent todo."""
|
||||
calendar_name = temporary_calendar
|
||||
fake_uid = f"nonexistent-{uuid.uuid4()}"
|
||||
|
||||
result = await nc_client.calendar.delete_todo(calendar_name, fake_uid)
|
||||
assert result["status_code"] == 404
|
||||
logger.info(f"Correctly got 404 for deleting nonexistent todo: {fake_uid}")
|
||||
|
||||
|
||||
async def test_list_todos_with_filters(
|
||||
nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test listing todos with various filters."""
|
||||
calendar_name = temporary_calendar
|
||||
|
||||
# Create todos with different statuses and priorities
|
||||
test_todos = [
|
||||
{
|
||||
"summary": "High Priority Task",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 1,
|
||||
"categories": "urgent",
|
||||
},
|
||||
{
|
||||
"summary": "In Progress Task",
|
||||
"status": "IN-PROCESS",
|
||||
"priority": 5,
|
||||
"categories": "work",
|
||||
},
|
||||
{
|
||||
"summary": "Low Priority Task",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 9,
|
||||
"categories": "someday",
|
||||
},
|
||||
]
|
||||
|
||||
created_uids = []
|
||||
|
||||
try:
|
||||
# Create test todos
|
||||
for todo_data in test_todos:
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
created_uids.append(result["uid"])
|
||||
|
||||
# Test basic list without filters
|
||||
all_todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
assert len(all_todos) >= 3
|
||||
|
||||
# Verify all our todos are in the list
|
||||
our_todo_uids = [t["uid"] for t in all_todos if t["uid"] in created_uids]
|
||||
assert len(our_todo_uids) == 3
|
||||
|
||||
logger.info(f"Successfully created and listed {len(created_uids)} test todos")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for uid in created_uids:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, uid)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -501,6 +501,120 @@ async def temporary_board_with_card(
|
||||
logger.error(f"Unexpected error deleting temporary card {card.id}: {e}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def shared_test_calendar_name():
|
||||
"""Unique calendar name for the entire test session."""
|
||||
return f"test_calendar_shared_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def shared_test_calendar_name_2():
|
||||
"""Second unique calendar name for cross-calendar tests."""
|
||||
return f"test_calendar_shared_2_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def shared_calendar(nc_client: NextcloudClient, shared_test_calendar_name: str):
|
||||
"""Create a shared calendar for all tests in the session. Reuses the calendar to avoid rate limiting."""
|
||||
calendar_name = shared_test_calendar_name
|
||||
|
||||
try:
|
||||
# Create a test calendar
|
||||
logger.info(f"Creating shared test calendar: {calendar_name}")
|
||||
result = await nc_client.calendar.create_calendar(
|
||||
calendar_name=calendar_name,
|
||||
display_name=f"Shared Test Calendar {calendar_name}",
|
||||
description="Shared calendar for integration testing (reused across tests)",
|
||||
color="#FF5722",
|
||||
)
|
||||
|
||||
if result["status_code"] not in [200, 201]:
|
||||
pytest.skip(f"Failed to create shared test calendar: {result}")
|
||||
|
||||
logger.info(f"Created shared test calendar: {calendar_name}")
|
||||
yield calendar_name
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error setting up shared test calendar: {e}")
|
||||
pytest.skip(f"Shared calendar setup failed: {e}")
|
||||
|
||||
finally:
|
||||
# Cleanup: Delete the shared calendar at end of session
|
||||
try:
|
||||
logger.info(f"Cleaning up shared test calendar: {calendar_name}")
|
||||
await nc_client.calendar.delete_calendar(calendar_name)
|
||||
logger.info(f"Successfully deleted shared test calendar: {calendar_name}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting shared test calendar {calendar_name}: {e}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def shared_calendar_2(
|
||||
nc_client: NextcloudClient, shared_test_calendar_name_2: str
|
||||
):
|
||||
"""Create a second shared calendar for cross-calendar tests."""
|
||||
calendar_name = shared_test_calendar_name_2
|
||||
|
||||
try:
|
||||
# Create a test calendar
|
||||
logger.info(f"Creating second shared test calendar: {calendar_name}")
|
||||
result = await nc_client.calendar.create_calendar(
|
||||
calendar_name=calendar_name,
|
||||
display_name=f"Shared Test Calendar 2 {calendar_name}",
|
||||
description="Second shared calendar for cross-calendar testing",
|
||||
color="#4CAF50",
|
||||
)
|
||||
|
||||
if result["status_code"] not in [200, 201]:
|
||||
pytest.skip(f"Failed to create second shared test calendar: {result}")
|
||||
|
||||
logger.info(f"Created second shared test calendar: {calendar_name}")
|
||||
yield calendar_name
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error setting up second shared test calendar: {e}")
|
||||
pytest.skip(f"Second shared calendar setup failed: {e}")
|
||||
|
||||
finally:
|
||||
# Cleanup: Delete the second shared calendar at end of session
|
||||
try:
|
||||
logger.info(f"Cleaning up second shared test calendar: {calendar_name}")
|
||||
await nc_client.calendar.delete_calendar(calendar_name)
|
||||
logger.info(
|
||||
f"Successfully deleted second shared test calendar: {calendar_name}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error deleting second shared test calendar {calendar_name}: {e}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def temporary_calendar(shared_calendar: str, nc_client: NextcloudClient):
|
||||
"""Provide the shared calendar and clean up todos after each test.
|
||||
|
||||
This fixture reuses a session-scoped calendar to avoid Nextcloud rate limiting
|
||||
on calendar creation. Each test gets the same calendar but todos are cleaned up
|
||||
between tests.
|
||||
"""
|
||||
calendar_name = shared_calendar
|
||||
|
||||
yield calendar_name
|
||||
|
||||
# Cleanup: Delete all todos from this calendar
|
||||
try:
|
||||
logger.info(f"Cleaning up todos from shared calendar: {calendar_name}")
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
for todo in todos:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo["uid"])
|
||||
except Exception as e:
|
||||
logger.warning(f"Error deleting todo {todo['uid']}: {e}")
|
||||
logger.info(f"Cleaned up {len(todos)} todos from shared calendar")
|
||||
except Exception as e:
|
||||
logger.error(f"Error cleaning up todos from calendar {calendar_name}: {e}")
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def nc_oauth_client(
|
||||
anyio_backend,
|
||||
|
||||
@@ -0,0 +1,476 @@
|
||||
"""Integration tests for Calendar VTODO (task) MCP tools."""
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import pytest
|
||||
from mcp import ClientSession
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
async def test_mcp_todo_complete_workflow(
|
||||
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test complete todo workflow via MCP tools with verification via NextcloudClient."""
|
||||
|
||||
calendar_name = temporary_calendar
|
||||
todo_uid = None
|
||||
|
||||
try:
|
||||
# 1. Create todo via MCP
|
||||
logger.info(f"Creating todo in {calendar_name} via MCP")
|
||||
tomorrow = datetime.now() + timedelta(days=1)
|
||||
|
||||
create_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_create_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"summary": "MCP Test Task",
|
||||
"description": "Test task created via MCP tools",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 3,
|
||||
"due": tomorrow.strftime("%Y-%m-%dT18:00:00"),
|
||||
"categories": "testing,mcp",
|
||||
},
|
||||
)
|
||||
assert create_result.isError is False
|
||||
|
||||
# Extract UID from the result
|
||||
result_data = create_result.content[0].text
|
||||
import json
|
||||
|
||||
result_json = json.loads(result_data)
|
||||
todo_uid = result_json["uid"]
|
||||
logger.info(f"Created todo with UID: {todo_uid}")
|
||||
|
||||
# 2. Verify todo creation via client
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
assert any(t["uid"] == todo_uid for t in todos)
|
||||
created_todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert created_todo["summary"] == "MCP Test Task"
|
||||
assert created_todo["status"] == "NEEDS-ACTION"
|
||||
assert created_todo["priority"] == 3
|
||||
|
||||
# 3. List todos via MCP
|
||||
logger.info(f"Listing todos in {calendar_name} via MCP")
|
||||
list_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_list_todos",
|
||||
{"calendar_name": calendar_name},
|
||||
)
|
||||
assert list_result.isError is False
|
||||
|
||||
list_data = json.loads(list_result.content[0].text)
|
||||
assert "todos" in list_data
|
||||
assert any(t["uid"] == todo_uid for t in list_data["todos"])
|
||||
|
||||
# 4. Update todo via MCP
|
||||
logger.info(f"Updating todo {todo_uid} via MCP")
|
||||
update_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_update_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"todo_uid": todo_uid,
|
||||
"summary": "MCP Test Task Updated",
|
||||
"status": "IN-PROCESS",
|
||||
"priority": 1,
|
||||
"percent_complete": 50,
|
||||
},
|
||||
)
|
||||
assert update_result.isError is False
|
||||
|
||||
# 5. Verify update via client
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
updated_todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert updated_todo["summary"] == "MCP Test Task Updated"
|
||||
assert updated_todo["status"] == "IN-PROCESS"
|
||||
assert updated_todo["priority"] == 1
|
||||
assert updated_todo["percent_complete"] == 50
|
||||
|
||||
# 6. Delete todo via MCP
|
||||
logger.info(f"Deleting todo {todo_uid} via MCP")
|
||||
delete_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_delete_todo",
|
||||
{"calendar_name": calendar_name, "todo_uid": todo_uid},
|
||||
)
|
||||
assert delete_result.isError is False
|
||||
|
||||
# 7. Verify deletion via client
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
assert not any(t["uid"] == todo_uid for t in todos)
|
||||
|
||||
logger.info("Complete todo workflow test passed")
|
||||
|
||||
finally:
|
||||
# Cleanup in case of failure
|
||||
if todo_uid:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_mcp_list_todos_with_filters(
|
||||
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test listing todos with various filters via MCP tools."""
|
||||
|
||||
calendar_name = temporary_calendar
|
||||
created_uids = []
|
||||
|
||||
try:
|
||||
# Create test todos with different properties
|
||||
test_todos = [
|
||||
{
|
||||
"summary": "High Priority Task",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 1,
|
||||
"categories": "urgent,work",
|
||||
},
|
||||
{
|
||||
"summary": "In Progress Task",
|
||||
"status": "IN-PROCESS",
|
||||
"priority": 5,
|
||||
"categories": "work",
|
||||
},
|
||||
{
|
||||
"summary": "Low Priority Task",
|
||||
"status": "NEEDS-ACTION",
|
||||
"priority": 9,
|
||||
"categories": "someday",
|
||||
},
|
||||
]
|
||||
|
||||
# Create todos via client
|
||||
for todo_data in test_todos:
|
||||
result = await nc_client.calendar.create_todo(calendar_name, todo_data)
|
||||
created_uids.append(result["uid"])
|
||||
|
||||
# Test 1: Filter by status
|
||||
logger.info("Testing filter by status")
|
||||
result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_list_todos",
|
||||
{"calendar_name": calendar_name, "status": "NEEDS-ACTION"},
|
||||
)
|
||||
assert result.isError is False
|
||||
import json
|
||||
|
||||
data = json.loads(result.content[0].text)
|
||||
needs_action_todos = [t for t in data["todos"] if t["uid"] in created_uids]
|
||||
assert len(needs_action_todos) == 2 # Two NEEDS-ACTION todos
|
||||
|
||||
# Test 2: Filter by priority
|
||||
logger.info("Testing filter by minimum priority")
|
||||
result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_list_todos",
|
||||
{"calendar_name": calendar_name, "min_priority": 1},
|
||||
)
|
||||
assert result.isError is False
|
||||
data = json.loads(result.content[0].text)
|
||||
high_priority_todos = [t for t in data["todos"] if t["uid"] in created_uids]
|
||||
assert len(high_priority_todos) >= 1 # At least the priority 1 todo
|
||||
|
||||
# Test 3: Filter by categories
|
||||
logger.info("Testing filter by categories")
|
||||
result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_list_todos",
|
||||
{"calendar_name": calendar_name, "categories": "work"},
|
||||
)
|
||||
assert result.isError is False
|
||||
data = json.loads(result.content[0].text)
|
||||
work_todos = [t for t in data["todos"] if t["uid"] in created_uids]
|
||||
assert len(work_todos) >= 2 # Two todos with "work" category
|
||||
|
||||
# Test 4: Filter by summary text
|
||||
logger.info("Testing filter by summary text")
|
||||
result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_list_todos",
|
||||
{"calendar_name": calendar_name, "summary_contains": "Priority"},
|
||||
)
|
||||
assert result.isError is False
|
||||
data = json.loads(result.content[0].text)
|
||||
priority_todos = [t for t in data["todos"] if t["uid"] in created_uids]
|
||||
assert len(priority_todos) == 2 # Two have "Priority" in summary (High, Low)
|
||||
|
||||
logger.info("List todos with filters test passed")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
for uid in created_uids:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_mcp_search_todos_across_calendars(
|
||||
nc_mcp_client: ClientSession,
|
||||
nc_client: NextcloudClient,
|
||||
temporary_calendar: str,
|
||||
shared_calendar_2: str,
|
||||
):
|
||||
"""Test searching todos across multiple calendars via MCP tools.
|
||||
|
||||
Note: Uses two shared test calendars to avoid rate limiting.
|
||||
"""
|
||||
|
||||
cal1_name = temporary_calendar # First shared test calendar
|
||||
cal2_name = shared_calendar_2 # Second shared test calendar
|
||||
created_uids = []
|
||||
|
||||
try:
|
||||
# Use existing shared calendars (no creation needed, avoiding rate limits)
|
||||
|
||||
# Create todos in both calendars
|
||||
result1 = await nc_client.calendar.create_todo(
|
||||
cal1_name,
|
||||
{
|
||||
"summary": "Task in Calendar 1",
|
||||
"status": "NEEDS-ACTION",
|
||||
"categories": "cal1",
|
||||
},
|
||||
)
|
||||
created_uids.append((cal1_name, result1["uid"]))
|
||||
|
||||
result2 = await nc_client.calendar.create_todo(
|
||||
cal2_name,
|
||||
{
|
||||
"summary": "Task in Calendar 2",
|
||||
"status": "IN-PROCESS",
|
||||
"categories": "cal2",
|
||||
},
|
||||
)
|
||||
created_uids.append((cal2_name, result2["uid"]))
|
||||
|
||||
# Search across all calendars via MCP
|
||||
logger.info("Searching todos across all calendars via MCP")
|
||||
search_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_search_todos",
|
||||
{},
|
||||
)
|
||||
assert search_result.isError is False
|
||||
|
||||
import json
|
||||
|
||||
data = json.loads(search_result.content[0].text)
|
||||
assert "todos" in data
|
||||
|
||||
# Verify both todos are in the results
|
||||
found_uids = {t["uid"] for t in data["todos"]}
|
||||
assert result1["uid"] in found_uids
|
||||
assert result2["uid"] in found_uids
|
||||
|
||||
# Verify calendar_name is included
|
||||
our_todos = [
|
||||
t for t in data["todos"] if t["uid"] in [result1["uid"], result2["uid"]]
|
||||
]
|
||||
for todo in our_todos:
|
||||
assert "calendar_name" in todo
|
||||
assert todo["calendar_name"] in [cal1_name, cal2_name]
|
||||
|
||||
# Test search with status filter
|
||||
logger.info("Searching with status filter via MCP")
|
||||
search_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_search_todos",
|
||||
{"status": "IN-PROCESS"},
|
||||
)
|
||||
assert search_result.isError is False
|
||||
data = json.loads(search_result.content[0].text)
|
||||
in_process_todos = [
|
||||
t for t in data["todos"] if t["uid"] in [uid for _, uid in created_uids]
|
||||
]
|
||||
assert len(in_process_todos) >= 1
|
||||
|
||||
logger.info("Search todos across calendars test passed")
|
||||
|
||||
finally:
|
||||
# Cleanup: Only delete todos, not calendars (they're reused/built-in)
|
||||
for cal_name, uid in created_uids:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(cal_name, uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_mcp_todo_status_transitions(
|
||||
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test transitioning through different todo statuses via MCP tools."""
|
||||
|
||||
calendar_name = temporary_calendar
|
||||
todo_uid = None
|
||||
|
||||
try:
|
||||
# Create todo
|
||||
result = await nc_client.calendar.create_todo(
|
||||
calendar_name,
|
||||
{"summary": "Status Transition Test", "status": "NEEDS-ACTION"},
|
||||
)
|
||||
todo_uid = result["uid"]
|
||||
|
||||
# Transition: NEEDS-ACTION → IN-PROCESS
|
||||
logger.info("Transitioning todo to IN-PROCESS via MCP")
|
||||
update_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_update_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"todo_uid": todo_uid,
|
||||
"status": "IN-PROCESS",
|
||||
"percent_complete": 25,
|
||||
},
|
||||
)
|
||||
assert update_result.isError is False
|
||||
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert todo["status"] == "IN-PROCESS"
|
||||
assert todo["percent_complete"] == 25
|
||||
|
||||
# Transition: IN-PROCESS → COMPLETED
|
||||
logger.info("Transitioning todo to COMPLETED via MCP")
|
||||
completed_time = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
|
||||
update_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_update_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"todo_uid": todo_uid,
|
||||
"status": "COMPLETED",
|
||||
"percent_complete": 100,
|
||||
"completed": completed_time,
|
||||
},
|
||||
)
|
||||
assert update_result.isError is False
|
||||
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert todo["status"] == "COMPLETED"
|
||||
assert todo["percent_complete"] == 100
|
||||
assert "completed" in todo
|
||||
|
||||
logger.info("Todo status transitions test passed")
|
||||
|
||||
finally:
|
||||
if todo_uid:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_mcp_todo_with_dates(
|
||||
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test creating and managing todos with date fields via MCP tools."""
|
||||
|
||||
calendar_name = temporary_calendar
|
||||
todo_uid = None
|
||||
|
||||
try:
|
||||
now = datetime.now()
|
||||
start_date = (now + timedelta(days=1)).strftime("%Y-%m-%dT09:00:00")
|
||||
due_date = (now + timedelta(days=7)).strftime("%Y-%m-%dT17:00:00")
|
||||
|
||||
# Create todo with dates via MCP
|
||||
logger.info("Creating todo with dates via MCP")
|
||||
create_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_create_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"summary": "Task with Dates",
|
||||
"description": "Test task with various date fields",
|
||||
"status": "NEEDS-ACTION",
|
||||
"dtstart": start_date,
|
||||
"due": due_date,
|
||||
},
|
||||
)
|
||||
assert create_result.isError is False
|
||||
|
||||
import json
|
||||
|
||||
result_data = json.loads(create_result.content[0].text)
|
||||
todo_uid = result_data["uid"]
|
||||
|
||||
# Verify dates via client
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
created_todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert created_todo["summary"] == "Task with Dates"
|
||||
assert "dtstart" in created_todo
|
||||
assert "due" in created_todo
|
||||
|
||||
logger.info("Todo with dates test passed")
|
||||
|
||||
finally:
|
||||
if todo_uid:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_mcp_todo_categories(
|
||||
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_calendar: str
|
||||
):
|
||||
"""Test creating and managing todos with categories via MCP tools."""
|
||||
|
||||
calendar_name = temporary_calendar
|
||||
todo_uid = None
|
||||
|
||||
try:
|
||||
# Create todo with multiple categories via MCP
|
||||
logger.info("Creating todo with categories via MCP")
|
||||
create_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_create_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"summary": "Task with Categories",
|
||||
"status": "NEEDS-ACTION",
|
||||
"categories": "work,meeting,important,quarterly",
|
||||
},
|
||||
)
|
||||
assert create_result.isError is False
|
||||
|
||||
import json
|
||||
|
||||
result_data = json.loads(create_result.content[0].text)
|
||||
todo_uid = result_data["uid"]
|
||||
|
||||
# Verify categories via client
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
created_todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
assert "categories" in created_todo
|
||||
categories_str = created_todo["categories"]
|
||||
assert "work" in categories_str
|
||||
assert "meeting" in categories_str
|
||||
assert "important" in categories_str
|
||||
assert "quarterly" in categories_str
|
||||
|
||||
# Update categories via MCP
|
||||
logger.info("Updating todo categories via MCP")
|
||||
update_result = await nc_mcp_client.call_tool(
|
||||
"nc_calendar_update_todo",
|
||||
{
|
||||
"calendar_name": calendar_name,
|
||||
"todo_uid": todo_uid,
|
||||
"categories": "updated,new-category",
|
||||
},
|
||||
)
|
||||
assert update_result.isError is False
|
||||
|
||||
# Verify updated categories
|
||||
todos = await nc_client.calendar.list_todos(calendar_name)
|
||||
updated_todo = next(t for t in todos if t["uid"] == todo_uid)
|
||||
categories_str = updated_todo["categories"]
|
||||
assert "updated" in categories_str
|
||||
assert "new-category" in categories_str
|
||||
|
||||
logger.info("Todo categories test passed")
|
||||
|
||||
finally:
|
||||
if todo_uid:
|
||||
try:
|
||||
await nc_client.calendar.delete_todo(calendar_name, todo_uid)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -57,6 +57,11 @@ async def test_mcp_connectivity(nc_mcp_client: ClientSession):
|
||||
"nc_calendar_find_availability",
|
||||
"nc_calendar_bulk_operations",
|
||||
"nc_calendar_manage_calendar",
|
||||
"nc_calendar_list_todos",
|
||||
"nc_calendar_create_todo",
|
||||
"nc_calendar_update_todo",
|
||||
"nc_calendar_delete_todo",
|
||||
"nc_calendar_search_todos",
|
||||
"deck_create_board",
|
||||
"nc_cookbook_import_recipe",
|
||||
"nc_cookbook_list_recipes",
|
||||
|
||||
@@ -54,8 +54,8 @@ wheels = [
|
||||
|
||||
[[package]]
|
||||
name = "caldav"
|
||||
version = "2.0.2.dev22+gaa8322dc7"
|
||||
source = { git = "https://github.com/cbcoutinho/caldav?branch=feature%2Fhttpx#aa8322dc7c4d0bf99593e1f46e577bb0aa5073c8" }
|
||||
version = "2.0.2.dev33+g4877e4688"
|
||||
source = { git = "https://github.com/cbcoutinho/caldav?branch=feature%2Fhttpx#4877e46884dbd2bc54f8fb61ee5d056342605e9c" }
|
||||
dependencies = [
|
||||
{ name = "httpx", extra = ["http2"] },
|
||||
{ name = "icalendar" },
|
||||
|
||||
Reference in New Issue
Block a user