nextcloud-mcp-server/tests/server/test_calendar_todos_mcp.py

"""Integration tests for Calendar VTODO (task) MCP tools."""

import json
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

        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

        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

        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

        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

        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