From 32f5a0fe52e6a4d1c4fd645d556ce9125dc94546 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 25 Mar 2026 07:53:50 +0100 Subject: [PATCH] feat: add Nextcloud Collectives app support (#621) Implement MCP tools for the Collectives wiki/documentation app, enabling agentic workflows for team knowledge base management. 16 tools covering collectives, pages, tags, search, and trash: - Read: list collectives, list/get pages (with WebDAV content), search, list tags, list trashed pages - Write: create/update collective, create/move/trash/restore pages, set emoji, create/assign/remove tags Includes Docker hook for app installation, OCS API client with envelope unwrapping, Pydantic models, unit tests (16), and integration tests (10). Closes #621 Co-Authored-By: Claude Opus 4.6 (1M context) --- .../10-install-collectives-app.sh | 35 ++ nextcloud_mcp_server/app.py | 2 + nextcloud_mcp_server/client/__init__.py | 2 + nextcloud_mcp_server/client/collectives.py | 233 ++++++++++ nextcloud_mcp_server/models/auth.py | 2 + nextcloud_mcp_server/models/collectives.py | 135 ++++++ nextcloud_mcp_server/server/__init__.py | 2 + nextcloud_mcp_server/server/collectives.py | 440 ++++++++++++++++++ tests/client/collectives/__init__.py | 0 .../collectives/test_collectives_api.py | 372 +++++++++++++++ tests/server/test_collectives_mcp.py | 412 ++++++++++++++++ 11 files changed, 1635 insertions(+) create mode 100755 app-hooks/post-installation/10-install-collectives-app.sh create mode 100644 nextcloud_mcp_server/client/collectives.py create mode 100644 nextcloud_mcp_server/models/collectives.py create mode 100644 nextcloud_mcp_server/server/collectives.py create mode 100644 tests/client/collectives/__init__.py create mode 100644 tests/client/collectives/test_collectives_api.py create mode 100644 tests/server/test_collectives_mcp.py diff --git a/app-hooks/post-installation/10-install-collectives-app.sh b/app-hooks/post-installation/10-install-collectives-app.sh new file mode 100755 index 00000000..7d96fab3 --- /dev/null +++ b/app-hooks/post-installation/10-install-collectives-app.sh @@ -0,0 +1,35 @@ +#!/bin/bash + +set -euox pipefail + +echo "Installing and configuring collectives app for testing..." + +# Collectives depends on Circles (teams) - ensure it's enabled +# Circles is bundled with Nextcloud, so just enable it +php /var/www/html/occ app:enable circles + +# Check if development collectives app is mounted at /opt/apps/collectives +if [ -d /opt/apps/collectives ]; then + echo "Development collectives app found at /opt/apps/collectives" + + # Remove any existing collectives app in apps (from app store or old symlink) + if [ -e /var/www/html/custom_apps/collectives ]; then + echo "Removing existing collectives in apps..." + rm -rf /var/www/html/custom_apps/collectives + fi + + # Create symlink from apps to the mounted development version + # Per Nextcloud docs: apps outside server root need symlinks in server root + echo "Creating symlink: custom_apps/collectives -> /opt/apps/collectives" + ln -sf /opt/apps/collectives /var/www/html/custom_apps/collectives + + echo "Enabling collectives app from /opt/apps (development mode via symlink)" + php /var/www/html/occ app:enable collectives +elif [ -d /var/www/html/custom_apps/collectives ]; then + echo "collectives app directory found in apps (already installed)" + php /var/www/html/occ app:enable collectives +else + echo "collectives app not found, installing from app store..." + php /var/www/html/occ app:install collectives + php /var/www/html/occ app:enable collectives +fi diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 733de031..e7f2f0c3 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -117,6 +117,7 @@ from nextcloud_mcp_server.observability.metrics import ( ) from nextcloud_mcp_server.server import ( configure_calendar_tools, + configure_collectives_tools, configure_contacts_tools, configure_cookbook_tools, configure_deck_tools, @@ -1274,6 +1275,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = "webdav": configure_webdav_tools, "sharing": configure_sharing_tools, "calendar": configure_calendar_tools, + "collectives": configure_collectives_tools, "contacts": configure_contacts_tools, "cookbook": configure_cookbook_tools, "deck": configure_deck_tools, diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index cd0aec4c..4b567efb 100644 --- a/nextcloud_mcp_server/client/__init__.py +++ b/nextcloud_mcp_server/client/__init__.py @@ -14,6 +14,7 @@ from httpx import ( from ..controllers.notes_search import NotesSearchController from ..http import nextcloud_httpx_transport from .calendar import CalendarClient +from .collectives import CollectivesClient from .contacts import ContactsClient from .cookbook import CookbookClient from .deck import DeckClient @@ -81,6 +82,7 @@ class NextcloudClient: ) # Uses AsyncDavClient internally self.contacts = ContactsClient(self._client, username) self.cookbook = CookbookClient(self._client, username) + self.collectives = CollectivesClient(self._client, username) self.deck = DeckClient(self._client, username) self.news = NewsClient(self._client, username) self.users = UsersClient(self._client, username) diff --git a/nextcloud_mcp_server/client/collectives.py b/nextcloud_mcp_server/client/collectives.py new file mode 100644 index 00000000..2366cbaf --- /dev/null +++ b/nextcloud_mcp_server/client/collectives.py @@ -0,0 +1,233 @@ +"""Client for Nextcloud Collectives app API (OCS).""" + +from typing import Any + +from nextcloud_mcp_server.client.base import BaseNextcloudClient + +API_BASE = "/ocs/v2.php/apps/collectives/api/v1.0" + + +class CollectivesClient(BaseNextcloudClient): + """Client for Nextcloud Collectives app operations.""" + + app_name = "collectives" + + def _get_ocs_headers( + self, additional_headers: dict[str, str] | None = None + ) -> dict[str, str]: + """Get standard headers required for OCS API calls.""" + headers = { + "OCS-APIRequest": "true", + "Content-Type": "application/json", + "Accept": "application/json", + } + if additional_headers: + headers.update(additional_headers) + return headers + + def _unwrap_ocs(self, response_json: dict[str, Any]) -> Any: + """Unwrap OCS envelope, returning the data payload.""" + return response_json["ocs"]["data"] + + # Collectives + + async def get_collectives(self) -> list[dict[str, Any]]: + """List all collectives the user has access to.""" + response = await self._make_request( + "GET", f"{API_BASE}/collectives", headers=self._get_ocs_headers() + ) + data = self._unwrap_ocs(response.json()) + return data["collectives"] + + async def create_collective( + self, name: str, emoji: str | None = None + ) -> dict[str, Any]: + """Create a new collective.""" + json_data: dict[str, Any] = {"name": name} + if emoji is not None: + json_data["emoji"] = emoji + response = await self._make_request( + "POST", + f"{API_BASE}/collectives", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["collective"] + + async def update_collective( + self, collective_id: int, emoji: str | None = None + ) -> dict[str, Any]: + """Update a collective (emoji).""" + json_data: dict[str, Any] = {} + if emoji is not None: + json_data["emoji"] = emoji + response = await self._make_request( + "PUT", + f"{API_BASE}/collectives/{collective_id}", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["collective"] + + # Pages + + async def get_pages(self, collective_id: int) -> list[dict[str, Any]]: + """List all pages in a collective.""" + response = await self._make_request( + "GET", + f"{API_BASE}/collectives/{collective_id}/pages", + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["pages"] + + async def get_page(self, collective_id: int, page_id: int) -> dict[str, Any]: + """Get a single page's metadata.""" + response = await self._make_request( + "GET", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}", + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["page"] + + async def create_page( + self, collective_id: int, parent_id: int, title: str + ) -> dict[str, Any]: + """Create a new page under a parent page.""" + json_data = {"title": title} + response = await self._make_request( + "POST", + f"{API_BASE}/collectives/{collective_id}/pages/{parent_id}", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["page"] + + async def move_page( + self, + collective_id: int, + page_id: int, + parent_id: int | None = None, + title: str | None = None, + index: int = 0, + copy: bool = False, + ) -> dict[str, Any]: + """Move or copy a page within a collective.""" + json_data: dict[str, Any] = {"index": index, "copy": copy} + if parent_id is not None: + json_data["parentId"] = parent_id + if title is not None: + json_data["title"] = title + response = await self._make_request( + "PUT", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["page"] + + async def trash_page(self, collective_id: int, page_id: int) -> None: + """Move a page to trash (soft delete).""" + await self._make_request( + "DELETE", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}", + headers=self._get_ocs_headers(), + ) + + async def set_page_emoji( + self, collective_id: int, page_id: int, emoji: str | None + ) -> dict[str, Any]: + """Set or clear the emoji on a page.""" + json_data = {"emoji": emoji} + response = await self._make_request( + "PUT", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}/emoji", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["page"] + + # Search + + async def search_pages( + self, collective_id: int, query: str + ) -> list[dict[str, Any]]: + """Full-text search within a collective.""" + response = await self._make_request( + "GET", + f"{API_BASE}/collectives/{collective_id}/search", + params={"searchString": query}, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["pages"] + + # Tags + + async def get_tags(self, collective_id: int) -> list[dict[str, Any]]: + """List all tags in a collective.""" + response = await self._make_request( + "GET", + f"{API_BASE}/collectives/{collective_id}/tags", + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["tags"] + + async def create_tag( + self, collective_id: int, name: str, color: str + ) -> dict[str, Any]: + """Create a new tag in a collective.""" + json_data = {"name": name, "color": color} + response = await self._make_request( + "POST", + f"{API_BASE}/collectives/{collective_id}/tags", + json=json_data, + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["tag"] + + async def assign_tag(self, collective_id: int, page_id: int, tag_id: int) -> None: + """Assign a tag to a page.""" + await self._make_request( + "PUT", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}/tags/{tag_id}", + headers=self._get_ocs_headers(), + ) + + async def remove_tag(self, collective_id: int, page_id: int, tag_id: int) -> None: + """Remove a tag from a page.""" + await self._make_request( + "DELETE", + f"{API_BASE}/collectives/{collective_id}/pages/{page_id}/tags/{tag_id}", + headers=self._get_ocs_headers(), + ) + + # Trash + + async def get_trashed_pages(self, collective_id: int) -> list[dict[str, Any]]: + """List trashed pages in a collective.""" + response = await self._make_request( + "GET", + f"{API_BASE}/collectives/{collective_id}/pages/trash", + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["pages"] + + async def restore_page(self, collective_id: int, page_id: int) -> dict[str, Any]: + """Restore a page from trash.""" + response = await self._make_request( + "PATCH", + f"{API_BASE}/collectives/{collective_id}/pages/trash/{page_id}", + headers=self._get_ocs_headers(), + ) + data = self._unwrap_ocs(response.json()) + return data["page"] diff --git a/nextcloud_mcp_server/models/auth.py b/nextcloud_mcp_server/models/auth.py index f249229c..e8f1c707 100644 --- a/nextcloud_mcp_server/models/auth.py +++ b/nextcloud_mcp_server/models/auth.py @@ -74,5 +74,7 @@ ALL_SUPPORTED_SCOPES: frozenset[str] = frozenset( "sharing:write", "news:read", "news:write", + "collectives:read", + "collectives:write", } ) diff --git a/nextcloud_mcp_server/models/collectives.py b/nextcloud_mcp_server/models/collectives.py new file mode 100644 index 00000000..f6dcb180 --- /dev/null +++ b/nextcloud_mcp_server/models/collectives.py @@ -0,0 +1,135 @@ +"""Pydantic models for Nextcloud Collectives app.""" + +from pydantic import BaseModel, Field + +from .base import BaseResponse, StatusResponse + +# Domain Models + + +class Collective(BaseModel): + """A Nextcloud Collective (wiki/knowledge base).""" + + id: int = Field(description="Collective ID") + circleId: str = Field(description="Linked Circle/Team ID") + emoji: str | None = Field(default=None, description="Collective emoji") + name: str = Field(description="Collective name") + level: int = Field(description="User's membership level") + canEdit: bool = Field(description="Whether the user can edit") + canShare: bool = Field(description="Whether the user can share") + pageMode: int = Field(description="Default page mode: 0=view, 1=edit") + + +class PageInfo(BaseModel): + """A page within a Collective.""" + + id: int = Field(description="Page ID") + title: str = Field(description="Page title") + emoji: str | None = Field(default=None, description="Page emoji") + fileName: str = Field(description="Markdown file name") + filePath: str = Field(description="File path within the collective") + collectivePath: str | None = Field( + default=None, description="Collective folder path in user's files" + ) + parentId: int = Field(description="Parent page ID (0 for root)") + timestamp: int = Field(description="Last modification Unix timestamp") + size: int = Field(description="Content size in bytes") + lastUserId: str | None = Field(default=None, description="Last editor user ID") + lastUserDisplayName: str | None = Field( + default=None, description="Last editor display name" + ) + subpageOrder: list[int] = Field( + default_factory=list, description="Ordered subpage IDs" + ) + isFullWidth: bool | None = Field(default=None, description="Full-width page layout") + + +class CollectiveTag(BaseModel): + """A tag within a Collective.""" + + id: int = Field(description="Tag ID") + collectiveId: int = Field(description="Parent collective ID") + name: str = Field(description="Tag name") + color: str = Field(description="Hex color code") + + +# Response Models + + +class ListCollectivesResponse(BaseResponse): + """Response for listing collectives.""" + + collectives: list[Collective] = Field(description="List of collectives") + total: int = Field(description="Total number of collectives") + + +class CreateCollectiveResponse(BaseResponse): + """Response for creating a collective.""" + + id: int = Field(description="Created collective ID") + name: str = Field(description="Created collective name") + emoji: str | None = Field(default=None, description="Collective emoji") + + +class CollectiveOperationResponse(StatusResponse): + """Response for collective update operations.""" + + collective_id: int = Field(description="ID of the affected collective") + + +class ListPagesResponse(BaseResponse): + """Response for listing pages in a collective.""" + + pages: list[PageInfo] = Field(description="List of pages") + total: int = Field(description="Total number of pages") + collective_id: int = Field(description="Collective ID") + + +class GetPageResponse(BaseResponse): + """Response for getting a single page with content.""" + + page: PageInfo = Field(description="Page metadata") + content: str | None = Field( + default=None, + description="Page markdown content (fetched via WebDAV)", + ) + + +class CreatePageResponse(BaseResponse): + """Response for creating a page.""" + + id: int = Field(description="Created page ID") + title: str = Field(description="Created page title") + collective_id: int = Field(description="Collective ID") + parent_id: int = Field(description="Parent page ID") + + +class PageOperationResponse(StatusResponse): + """Response for page operations (update, trash, emoji, tag).""" + + page_id: int = Field(description="ID of the affected page") + collective_id: int = Field(description="Collective ID") + + +class SearchPagesResponse(BaseResponse): + """Response for full-text search within a collective.""" + + results: list[PageInfo] = Field(description="Matching pages") + total: int = Field(description="Total number of results") + query: str = Field(description="Search query") + collective_id: int = Field(description="Collective ID") + + +class ListTagsResponse(BaseResponse): + """Response for listing tags in a collective.""" + + tags: list[CollectiveTag] = Field(description="List of tags") + total: int = Field(description="Total number of tags") + + +class CreateTagResponse(BaseResponse): + """Response for creating a tag.""" + + id: int = Field(description="Created tag ID") + name: str = Field(description="Tag name") + color: str = Field(description="Tag color") diff --git a/nextcloud_mcp_server/server/__init__.py b/nextcloud_mcp_server/server/__init__.py index 0be6bbaa..c7c053a9 100644 --- a/nextcloud_mcp_server/server/__init__.py +++ b/nextcloud_mcp_server/server/__init__.py @@ -1,4 +1,5 @@ from .calendar import configure_calendar_tools +from .collectives import configure_collectives_tools from .contacts import configure_contacts_tools from .cookbook import configure_cookbook_tools from .deck import configure_deck_tools @@ -11,6 +12,7 @@ from .webdav import configure_webdav_tools __all__ = [ "configure_calendar_tools", + "configure_collectives_tools", "configure_contacts_tools", "configure_cookbook_tools", "configure_deck_tools", diff --git a/nextcloud_mcp_server/server/collectives.py b/nextcloud_mcp_server/server/collectives.py new file mode 100644 index 00000000..0c40b49a --- /dev/null +++ b/nextcloud_mcp_server/server/collectives.py @@ -0,0 +1,440 @@ +"""MCP tool definitions for Nextcloud Collectives app.""" + +import logging + +from mcp.server.fastmcp import Context, FastMCP +from mcp.types import ToolAnnotations + +from nextcloud_mcp_server.auth import require_scopes +from nextcloud_mcp_server.context import get_client +from nextcloud_mcp_server.models.collectives import ( + Collective, + CollectiveOperationResponse, + CollectiveTag, + CreateCollectiveResponse, + CreatePageResponse, + CreateTagResponse, + GetPageResponse, + ListCollectivesResponse, + ListPagesResponse, + ListTagsResponse, + PageInfo, + PageOperationResponse, + SearchPagesResponse, +) +from nextcloud_mcp_server.observability.metrics import instrument_tool + +logger = logging.getLogger(__name__) + + +def configure_collectives_tools(mcp: FastMCP): + """Configure Nextcloud Collectives tools for the MCP server.""" + + # --- Read Tools --- + + @mcp.tool( + title="List Collectives", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_get_collectives( + ctx: Context, + ) -> ListCollectivesResponse: + """List all Nextcloud Collectives the user has access to""" + client = await get_client(ctx) + raw_collectives = await client.collectives.get_collectives() + collectives = [Collective(**c) for c in raw_collectives] + return ListCollectivesResponse(collectives=collectives, total=len(collectives)) + + @mcp.tool( + title="List Collective Pages", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_get_pages( + ctx: Context, collective_id: int + ) -> ListPagesResponse: + """List all pages in a Nextcloud Collective + + Args: + collective_id: ID of the collective + """ + client = await get_client(ctx) + raw_pages = await client.collectives.get_pages(collective_id) + pages = [PageInfo(**p) for p in raw_pages] + return ListPagesResponse( + pages=pages, total=len(pages), collective_id=collective_id + ) + + @mcp.tool( + title="Get Collective Page", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_get_page( + ctx: Context, collective_id: int, page_id: int + ) -> GetPageResponse: + """Get a page's metadata and markdown content from a Nextcloud Collective. + + Content is fetched via WebDAV using the page's file path. To update + page content, use the nc_webdav_write_file tool with the path from + the page's collectivePath/filePath fields. + + Args: + collective_id: ID of the collective + page_id: ID of the page + """ + client = await get_client(ctx) + raw_page = await client.collectives.get_page(collective_id, page_id) + page = PageInfo(**raw_page) + + # Fetch content via WebDAV + # Path structure: collectivePath/filePath/fileName + # filePath is empty for root-level pages, contains subdirectory for nested pages + content = None + if page.collectivePath and page.fileName: + parts = [page.collectivePath] + if page.filePath: + parts.append(page.filePath) + parts.append(page.fileName) + webdav_path = "/".join(parts) + try: + file_bytes, _ = await client.webdav.read_file(webdav_path) + content = file_bytes.decode("utf-8") + except Exception: + logger.warning( + "Failed to read page content via WebDAV: %s", + webdav_path, + ) + + return GetPageResponse(page=page, content=content) + + @mcp.tool( + title="Search Collective Pages", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_search_pages( + ctx: Context, collective_id: int, query: str + ) -> SearchPagesResponse: + """Full-text search within a Nextcloud Collective + + Args: + collective_id: ID of the collective + query: Search query string + """ + client = await get_client(ctx) + raw_pages = await client.collectives.search_pages(collective_id, query) + pages = [PageInfo(**p) for p in raw_pages] + return SearchPagesResponse( + results=pages, + total=len(pages), + query=query, + collective_id=collective_id, + ) + + @mcp.tool( + title="List Collective Tags", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_get_tags( + ctx: Context, collective_id: int + ) -> ListTagsResponse: + """List all tags in a Nextcloud Collective + + Args: + collective_id: ID of the collective + """ + client = await get_client(ctx) + raw_tags = await client.collectives.get_tags(collective_id) + tags = [CollectiveTag(**t) for t in raw_tags] + return ListTagsResponse(tags=tags, total=len(tags)) + + @mcp.tool( + title="List Trashed Collective Pages", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("collectives:read") + @instrument_tool + async def collectives_get_trashed_pages( + ctx: Context, collective_id: int + ) -> ListPagesResponse: + """List trashed pages in a Nextcloud Collective + + Args: + collective_id: ID of the collective + """ + client = await get_client(ctx) + raw_pages = await client.collectives.get_trashed_pages(collective_id) + pages = [PageInfo(**p) for p in raw_pages] + return ListPagesResponse( + pages=pages, total=len(pages), collective_id=collective_id + ) + + # --- Write Tools --- + + @mcp.tool( + title="Create Collective", + annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_create_collective( + ctx: Context, name: str, emoji: str | None = None + ) -> CreateCollectiveResponse: + """Create a new Nextcloud Collective + + Args: + name: Name of the collective + emoji: Optional emoji for the collective + """ + client = await get_client(ctx) + raw = await client.collectives.create_collective(name, emoji) + collective = Collective(**raw) + return CreateCollectiveResponse( + id=collective.id, name=collective.name, emoji=collective.emoji + ) + + @mcp.tool( + title="Update Collective", + annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_update_collective( + ctx: Context, collective_id: int, emoji: str | None = None + ) -> CollectiveOperationResponse: + """Update a Nextcloud Collective (emoji) + + Args: + collective_id: ID of the collective + emoji: New emoji for the collective + """ + client = await get_client(ctx) + await client.collectives.update_collective(collective_id, emoji) + return CollectiveOperationResponse( + collective_id=collective_id, + status_code=200, + message="Collective updated", + ) + + @mcp.tool( + title="Create Collective Page", + annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_create_page( + ctx: Context, collective_id: int, parent_id: int, title: str + ) -> CreatePageResponse: + """Create a new page in a Nextcloud Collective. + + Pages are created as empty markdown files. Use nc_webdav_write_file + with the page's collectivePath/filePath to add content after creation. + + Args: + collective_id: ID of the collective + parent_id: ID of the parent page (use 0 for top-level pages) + title: Title of the new page + """ + client = await get_client(ctx) + raw = await client.collectives.create_page(collective_id, parent_id, title) + page = PageInfo(**raw) + return CreatePageResponse( + id=page.id, + title=page.title, + collective_id=collective_id, + parent_id=page.parentId, + ) + + @mcp.tool( + title="Move Collective Page", + annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_move_page( + ctx: Context, + collective_id: int, + page_id: int, + parent_id: int | None = None, + title: str | None = None, + index: int = 0, + copy: bool = False, + ) -> PageOperationResponse: + """Move or copy a page within a Nextcloud Collective + + Args: + collective_id: ID of the collective + page_id: ID of the page to move/copy + parent_id: Target parent page ID + title: New title (optional) + index: Position in subpage order (default 0) + copy: If true, copy instead of move + """ + client = await get_client(ctx) + await client.collectives.move_page( + collective_id, page_id, parent_id, title, index, copy + ) + action = "copied" if copy else "moved" + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message=f"Page {action}", + ) + + @mcp.tool( + title="Trash Collective Page", + annotations=ToolAnnotations( + destructiveHint=True, idempotentHint=True, openWorldHint=True + ), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_trash_page( + ctx: Context, collective_id: int, page_id: int + ) -> PageOperationResponse: + """Move a page to trash in a Nextcloud Collective (soft delete) + + Args: + collective_id: ID of the collective + page_id: ID of the page to trash + """ + client = await get_client(ctx) + await client.collectives.trash_page(collective_id, page_id) + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message="Page moved to trash", + ) + + @mcp.tool( + title="Restore Collective Page", + annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_restore_page( + ctx: Context, collective_id: int, page_id: int + ) -> PageOperationResponse: + """Restore a page from trash in a Nextcloud Collective + + Args: + collective_id: ID of the collective + page_id: ID of the page to restore + """ + client = await get_client(ctx) + await client.collectives.restore_page(collective_id, page_id) + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message="Page restored from trash", + ) + + @mcp.tool( + title="Set Collective Page Emoji", + annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_set_page_emoji( + ctx: Context, + collective_id: int, + page_id: int, + emoji: str | None = None, + ) -> PageOperationResponse: + """Set or clear the emoji on a Nextcloud Collective page + + Args: + collective_id: ID of the collective + page_id: ID of the page + emoji: Emoji to set, or null to clear + """ + client = await get_client(ctx) + await client.collectives.set_page_emoji(collective_id, page_id, emoji) + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message="Page emoji updated", + ) + + @mcp.tool( + title="Create Collective Tag", + annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_create_tag( + ctx: Context, collective_id: int, name: str, color: str + ) -> CreateTagResponse: + """Create a new tag in a Nextcloud Collective + + Args: + collective_id: ID of the collective + name: Tag name + color: Hex color code (e.g. "FF0000") + """ + client = await get_client(ctx) + raw = await client.collectives.create_tag(collective_id, name, color) + tag = CollectiveTag(**raw) + return CreateTagResponse(id=tag.id, name=tag.name, color=tag.color) + + @mcp.tool( + title="Assign Tag to Collective Page", + annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_assign_tag( + ctx: Context, collective_id: int, page_id: int, tag_id: int + ) -> PageOperationResponse: + """Assign a tag to a page in a Nextcloud Collective + + Args: + collective_id: ID of the collective + page_id: ID of the page + tag_id: ID of the tag to assign + """ + client = await get_client(ctx) + await client.collectives.assign_tag(collective_id, page_id, tag_id) + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message=f"Tag {tag_id} assigned to page", + ) + + @mcp.tool( + title="Remove Tag from Collective Page", + annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), + ) + @require_scopes("collectives:write") + @instrument_tool + async def collectives_remove_tag( + ctx: Context, collective_id: int, page_id: int, tag_id: int + ) -> PageOperationResponse: + """Remove a tag from a page in a Nextcloud Collective + + Args: + collective_id: ID of the collective + page_id: ID of the page + tag_id: ID of the tag to remove + """ + client = await get_client(ctx) + await client.collectives.remove_tag(collective_id, page_id, tag_id) + return PageOperationResponse( + page_id=page_id, + collective_id=collective_id, + status_code=200, + message=f"Tag {tag_id} removed from page", + ) diff --git a/tests/client/collectives/__init__.py b/tests/client/collectives/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/client/collectives/test_collectives_api.py b/tests/client/collectives/test_collectives_api.py new file mode 100644 index 00000000..56adea5d --- /dev/null +++ b/tests/client/collectives/test_collectives_api.py @@ -0,0 +1,372 @@ +"""Unit tests for CollectivesClient API methods.""" + +import httpx +import pytest + +from nextcloud_mcp_server.client.collectives import CollectivesClient +from tests.client.conftest import create_mock_response + +pytestmark = pytest.mark.unit + + +# --- OCS mock helpers --- + + +def _ocs_response(data: dict | list) -> httpx.Response: + """Wrap data in an OCS envelope and return as mock response.""" + return create_mock_response( + status_code=200, + json_data={"ocs": {"meta": {"status": "ok", "statuscode": 200}, "data": data}}, + ) + + +def _sample_collective( + collective_id: int = 1, name: str = "Test Wiki", emoji: str | None = None +) -> dict: + return { + "id": collective_id, + "circleId": "circle-abc", + "emoji": emoji, + "name": name, + "level": 9, + "canEdit": True, + "canShare": True, + "pageMode": 0, + } + + +def _sample_page( + page_id: int = 10, + title: str = "Test Page", + parent_id: int = 0, + collective_path: str = "Collectives/Test Wiki", +) -> dict: + return { + "id": page_id, + "title": title, + "emoji": None, + "fileName": f"{title}.md", + "filePath": f"{title}.md", + "collectivePath": collective_path, + "parentId": parent_id, + "timestamp": 1700000000, + "size": 42, + "lastUserId": "testuser", + "lastUserDisplayName": "Test User", + "subpageOrder": [], + "isFullWidth": False, + } + + +def _sample_tag( + tag_id: int = 1, name: str = "important", color: str = "FF0000" +) -> dict: + return { + "id": tag_id, + "collectiveId": 1, + "name": name, + "color": color, + } + + +# --- Collectives --- + + +async def test_get_collectives(mocker): + """Test listing collectives unwraps OCS envelope correctly.""" + mock_response = _ocs_response( + { + "collectives": [ + _sample_collective(1, "Wiki A"), + _sample_collective(2, "Wiki B"), + ] + } + ) + mocker.patch.object(CollectivesClient, "_make_request", return_value=mock_response) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.get_collectives() + + assert isinstance(result, list) + assert len(result) == 2 + assert result[0]["id"] == 1 + assert result[1]["name"] == "Wiki B" + + +async def test_create_collective(mocker): + """Test creating a collective sends name and emoji.""" + mock_response = _ocs_response( + {"collective": _sample_collective(5, "New Wiki", "๐Ÿ“š")} + ) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.create_collective("New Wiki", emoji="๐Ÿ“š") + + assert result["id"] == 5 + assert result["name"] == "New Wiki" + assert result["emoji"] == "๐Ÿ“š" + + call_args = mock_request.call_args + assert call_args[0][0] == "POST" + assert call_args[1]["json"]["name"] == "New Wiki" + assert call_args[1]["json"]["emoji"] == "๐Ÿ“š" + + +# --- Pages --- + + +async def test_get_pages(mocker): + """Test listing pages in a collective.""" + mock_response = _ocs_response( + {"pages": [_sample_page(10, "Page A"), _sample_page(20, "Page B")]} + ) + mocker.patch.object(CollectivesClient, "_make_request", return_value=mock_response) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.get_pages(collective_id=1) + + assert len(result) == 2 + assert result[0]["title"] == "Page A" + assert result[1]["id"] == 20 + + +async def test_get_page(mocker): + """Test getting a single page metadata.""" + mock_response = _ocs_response({"page": _sample_page(10, "My Page")}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.get_page(collective_id=1, page_id=10) + + assert result["id"] == 10 + assert result["title"] == "My Page" + assert result["collectivePath"] == "Collectives/Test Wiki" + + call_args = mock_request.call_args + assert "/collectives/1/pages/10" in call_args[0][1] + + +async def test_create_page(mocker): + """Test creating a page under a parent.""" + mock_response = _ocs_response({"page": _sample_page(30, "New Page", parent_id=10)}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.create_page(collective_id=1, parent_id=10, title="New Page") + + assert result["id"] == 30 + assert result["parentId"] == 10 + + call_args = mock_request.call_args + assert call_args[0][0] == "POST" + assert "/collectives/1/pages/10" in call_args[0][1] + assert call_args[1]["json"]["title"] == "New Page" + + +async def test_trash_page(mocker): + """Test trashing a page sends DELETE.""" + mock_response = create_mock_response(status_code=200, json_data={}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + await client.trash_page(collective_id=1, page_id=10) + + call_args = mock_request.call_args + assert call_args[0][0] == "DELETE" + assert "/collectives/1/pages/10" in call_args[0][1] + + +async def test_move_page(mocker): + """Test moving a page sends PUT with correct params.""" + mock_response = _ocs_response( + {"page": _sample_page(10, "Moved Page", parent_id=20)} + ) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.move_page( + collective_id=1, page_id=10, parent_id=20, title="Moved Page" + ) + + assert result["parentId"] == 20 + call_args = mock_request.call_args + assert call_args[0][0] == "PUT" + assert call_args[1]["json"]["parentId"] == 20 + + +# --- Search --- + + +async def test_search_pages(mocker): + """Test full-text search sends query parameter.""" + mock_response = _ocs_response({"pages": [_sample_page(10, "Result Page")]}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.search_pages(collective_id=1, query="test query") + + assert len(result) == 1 + assert result[0]["title"] == "Result Page" + + call_args = mock_request.call_args + assert call_args[1]["params"]["searchString"] == "test query" + + +# --- Tags --- + + +async def test_get_tags(mocker): + """Test listing tags.""" + mock_response = _ocs_response( + {"tags": [_sample_tag(1, "important"), _sample_tag(2, "draft", "00FF00")]} + ) + mocker.patch.object(CollectivesClient, "_make_request", return_value=mock_response) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.get_tags(collective_id=1) + + assert len(result) == 2 + assert result[0]["name"] == "important" + assert result[1]["color"] == "00FF00" + + +async def test_create_tag(mocker): + """Test creating a tag.""" + mock_response = _ocs_response({"tag": _sample_tag(3, "review", "0000FF")}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.create_tag(collective_id=1, name="review", color="0000FF") + + assert result["id"] == 3 + assert result["name"] == "review" + + call_args = mock_request.call_args + assert call_args[0][0] == "POST" + assert call_args[1]["json"]["name"] == "review" + + +async def test_assign_tag(mocker): + """Test assigning a tag to a page.""" + mock_response = create_mock_response(status_code=200, json_data={}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + await client.assign_tag(collective_id=1, page_id=10, tag_id=3) + + call_args = mock_request.call_args + assert call_args[0][0] == "PUT" + assert "/pages/10/tags/3" in call_args[0][1] + + +async def test_remove_tag(mocker): + """Test removing a tag from a page.""" + mock_response = create_mock_response(status_code=200, json_data={}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + await client.remove_tag(collective_id=1, page_id=10, tag_id=3) + + call_args = mock_request.call_args + assert call_args[0][0] == "DELETE" + assert "/pages/10/tags/3" in call_args[0][1] + + +# --- Trash --- + + +async def test_get_trashed_pages(mocker): + """Test listing trashed pages.""" + trashed = _sample_page(10, "Trashed Page") + trashed["trashTimestamp"] = 1700000000 + mock_response = _ocs_response({"pages": [trashed]}) + mocker.patch.object(CollectivesClient, "_make_request", return_value=mock_response) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.get_trashed_pages(collective_id=1) + + assert len(result) == 1 + assert result[0]["title"] == "Trashed Page" + + +async def test_restore_page(mocker): + """Test restoring a page from trash.""" + mock_response = _ocs_response({"page": _sample_page(10, "Restored Page")}) + mock_request = mocker.patch.object( + CollectivesClient, "_make_request", return_value=mock_response + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + result = await client.restore_page(collective_id=1, page_id=10) + + assert result["title"] == "Restored Page" + call_args = mock_request.call_args + assert call_args[0][0] == "PATCH" + assert "/pages/trash/10" in call_args[0][1] + + +# --- Error Handling --- + + +async def test_get_collectives_403(mocker): + """Test 403 response raises HTTPStatusError.""" + mock_response = create_mock_response( + status_code=403, json_data={"message": "Forbidden"} + ) + mock_response.raise_for_status = lambda: (_ for _ in ()).throw( + httpx.HTTPStatusError( + "Forbidden", request=mock_response.request, response=mock_response + ) + ) + mocker.patch.object( + CollectivesClient, + "_make_request", + side_effect=httpx.HTTPStatusError( + "Forbidden", + request=httpx.Request("GET", "http://test"), + response=mock_response, + ), + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + with pytest.raises(httpx.HTTPStatusError): + await client.get_collectives() + + +async def test_get_page_404(mocker): + """Test 404 response raises HTTPStatusError.""" + mock_response = create_mock_response( + status_code=404, json_data={"message": "Not found"} + ) + mocker.patch.object( + CollectivesClient, + "_make_request", + side_effect=httpx.HTTPStatusError( + "Not Found", + request=httpx.Request("GET", "http://test"), + response=mock_response, + ), + ) + + client = CollectivesClient(mocker.AsyncMock(spec=httpx.AsyncClient), "testuser") + with pytest.raises(httpx.HTTPStatusError): + await client.get_page(collective_id=1, page_id=999) diff --git a/tests/server/test_collectives_mcp.py b/tests/server/test_collectives_mcp.py new file mode 100644 index 00000000..dee52889 --- /dev/null +++ b/tests/server/test_collectives_mcp.py @@ -0,0 +1,412 @@ +"""Integration tests for Nextcloud Collectives MCP tools.""" + +import json +import logging +import uuid + +import httpx +import pytest +from mcp import ClientSession + +logger = logging.getLogger(__name__) +pytestmark = pytest.mark.integration + +# Nextcloud credentials for direct API cleanup (matches docker-compose.yml) +_NC_BASE = "http://localhost:8080" +_NC_AUTH = ("admin", "admin") +_OCS_HEADERS = { + "OCS-APIRequest": "true", + "Accept": "application/json", +} + + +# --- Fixtures --- + + +@pytest.fixture(scope="session") +async def temporary_collective(nc_mcp_client: ClientSession): + """Create a temporary collective for testing. Cleaned up after session.""" + unique_suffix = uuid.uuid4().hex[:8] + name = f"MCP Test Collective {unique_suffix}" + + result = await nc_mcp_client.call_tool( + "collectives_create_collective", + {"name": name, "emoji": "๐Ÿงช"}, + ) + assert result.isError is False, f"Failed to create collective: {result.content}" + data = json.loads(result.content[0].text) + collective_id = data["id"] + logger.info(f"Created temporary collective: {name} (ID: {collective_id})") + + # Get the landing page ID (auto-created with each collective) + pages_result = await nc_mcp_client.call_tool( + "collectives_get_pages", + {"collective_id": collective_id}, + ) + pages_data = json.loads(pages_result.content[0].text) + landing_page_id = pages_data["pages"][0]["id"] + + yield { + "id": collective_id, + "name": name, + "landing_page_id": landing_page_id, + } + + # Cleanup: trash and permanently delete the collective via direct OCS API + try: + async with httpx.AsyncClient(base_url=_NC_BASE, auth=_NC_AUTH) as client: + api = "/ocs/v2.php/apps/collectives/api/v1.0" + await client.delete( + f"{api}/collectives/{collective_id}", + headers=_OCS_HEADERS, + ) + await client.delete( + f"{api}/collectives/trash/{collective_id}", + headers=_OCS_HEADERS, + ) + logger.info(f"Cleaned up collective: {collective_id}") + except Exception as e: + logger.warning(f"Cleanup of collective {collective_id} failed: {e}") + + +# --- Tool Discovery --- + + +async def test_collectives_tools_available(nc_mcp_client: ClientSession): + """Verify all Collectives MCP tools are registered.""" + tools = await nc_mcp_client.list_tools() + tool_names = [tool.name for tool in tools.tools] + + expected_tools = [ + "collectives_get_collectives", + "collectives_create_collective", + "collectives_update_collective", + "collectives_get_pages", + "collectives_get_page", + "collectives_create_page", + "collectives_move_page", + "collectives_trash_page", + "collectives_restore_page", + "collectives_set_page_emoji", + "collectives_search_pages", + "collectives_get_tags", + "collectives_create_tag", + "collectives_assign_tag", + "collectives_remove_tag", + "collectives_get_trashed_pages", + ] + + for expected in expected_tools: + assert expected in tool_names, ( + f"Expected tool '{expected}' not found in available tools" + ) + + logger.info(f"All {len(expected_tools)} Collectives tools registered") + + +# --- Collective CRUD --- + + +async def test_collectives_list( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test listing collectives includes the temporary one.""" + result = await nc_mcp_client.call_tool("collectives_get_collectives", {}) + assert result.isError is False + + data = json.loads(result.content[0].text) + assert data["success"] is True + assert data["total"] >= 1 + + collective_ids = [c["id"] for c in data["collectives"]] + assert temporary_collective["id"] in collective_ids + logger.info(f"Found {data['total']} collectives") + + +async def test_collectives_update_emoji( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test updating a collective's emoji.""" + result = await nc_mcp_client.call_tool( + "collectives_update_collective", + {"collective_id": temporary_collective["id"], "emoji": "๐Ÿ“–"}, + ) + assert result.isError is False + + data = json.loads(result.content[0].text) + assert data["success"] is True + assert data["collective_id"] == temporary_collective["id"] + logger.info("Collective emoji updated") + + +# --- Page CRUD --- + + +async def test_collectives_page_workflow( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test the full page lifecycle: create, read, set emoji, trash, restore.""" + cid = temporary_collective["id"] + landing_id = temporary_collective["landing_page_id"] + + # 1. Create a page + unique_title = f"Test Page {uuid.uuid4().hex[:8]}" + create_result = await nc_mcp_client.call_tool( + "collectives_create_page", + {"collective_id": cid, "parent_id": landing_id, "title": unique_title}, + ) + assert create_result.isError is False + create_data = json.loads(create_result.content[0].text) + page_id = create_data["id"] + assert create_data["collective_id"] == cid + assert create_data["parent_id"] == landing_id + logger.info(f"Created page: {unique_title} (ID: {page_id})") + + # 2. List pages โ€” should include the new page + list_result = await nc_mcp_client.call_tool( + "collectives_get_pages", + {"collective_id": cid}, + ) + assert list_result.isError is False + list_data = json.loads(list_result.content[0].text) + page_ids = [p["id"] for p in list_data["pages"]] + assert page_id in page_ids + logger.info(f"Page found in list ({list_data['total']} pages)") + + # 3. Get page with content + get_result = await nc_mcp_client.call_tool( + "collectives_get_page", + {"collective_id": cid, "page_id": page_id}, + ) + assert get_result.isError is False + get_data = json.loads(get_result.content[0].text) + assert get_data["page"]["id"] == page_id + assert get_data["page"]["title"] == unique_title + # New pages have empty content (empty string or None) + logger.info("Page metadata retrieved") + + # 4. Set page emoji + emoji_result = await nc_mcp_client.call_tool( + "collectives_set_page_emoji", + {"collective_id": cid, "page_id": page_id, "emoji": "๐Ÿš€"}, + ) + assert emoji_result.isError is False + logger.info("Page emoji set") + + # 5. Trash the page + trash_result = await nc_mcp_client.call_tool( + "collectives_trash_page", + {"collective_id": cid, "page_id": page_id}, + ) + assert trash_result.isError is False + logger.info("Page trashed") + + # 6. Verify page is in trash + trashed_result = await nc_mcp_client.call_tool( + "collectives_get_trashed_pages", + {"collective_id": cid}, + ) + assert trashed_result.isError is False + trashed_data = json.loads(trashed_result.content[0].text) + trashed_ids = [p["id"] for p in trashed_data["pages"]] + assert page_id in trashed_ids + logger.info("Page found in trash") + + # 7. Restore from trash + restore_result = await nc_mcp_client.call_tool( + "collectives_restore_page", + {"collective_id": cid, "page_id": page_id}, + ) + assert restore_result.isError is False + logger.info("Page restored from trash") + + # 8. Verify page is back in pages list + list_result2 = await nc_mcp_client.call_tool( + "collectives_get_pages", + {"collective_id": cid}, + ) + list_data2 = json.loads(list_result2.content[0].text) + page_ids2 = [p["id"] for p in list_data2["pages"]] + assert page_id in page_ids2 + logger.info("Page verified restored to pages list") + + +async def test_collectives_get_landing_page_content( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test that the landing page has auto-generated content readable via WebDAV.""" + cid = temporary_collective["id"] + landing_id = temporary_collective["landing_page_id"] + + result = await nc_mcp_client.call_tool( + "collectives_get_page", + {"collective_id": cid, "page_id": landing_id}, + ) + assert result.isError is False + + data = json.loads(result.content[0].text) + assert data["page"]["fileName"] == "Readme.md" + assert data["content"] is not None, ( + "Landing page should have auto-generated content" + ) + assert "Welcome" in data["content"], "Landing page should contain welcome text" + logger.info(f"Landing page content: {len(data['content'])} bytes") + + +async def test_collectives_move_page( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test moving a page (rename).""" + cid = temporary_collective["id"] + landing_id = temporary_collective["landing_page_id"] + + # Create a page to move + create_result = await nc_mcp_client.call_tool( + "collectives_create_page", + { + "collective_id": cid, + "parent_id": landing_id, + "title": f"Movable Page {uuid.uuid4().hex[:8]}", + }, + ) + assert create_result.isError is False + page_id = json.loads(create_result.content[0].text)["id"] + + # Move (rename) the page + new_title = f"Renamed Page {uuid.uuid4().hex[:8]}" + move_result = await nc_mcp_client.call_tool( + "collectives_move_page", + { + "collective_id": cid, + "page_id": page_id, + "title": new_title, + }, + ) + assert move_result.isError is False + + data = json.loads(move_result.content[0].text) + assert data["page_id"] == page_id + assert "moved" in data["message"] + logger.info(f"Page renamed to: {new_title}") + + # Cleanup + await nc_mcp_client.call_tool( + "collectives_trash_page", + {"collective_id": cid, "page_id": page_id}, + ) + + +# --- Tags --- + + +async def test_collectives_tag_workflow( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test tag lifecycle: create tag, assign to page, remove from page.""" + cid = temporary_collective["id"] + landing_id = temporary_collective["landing_page_id"] + + # 1. Create a tag + tag_name = f"test-tag-{uuid.uuid4().hex[:6]}" + create_tag_result = await nc_mcp_client.call_tool( + "collectives_create_tag", + {"collective_id": cid, "name": tag_name, "color": "FF5733"}, + ) + assert create_tag_result.isError is False + tag_data = json.loads(create_tag_result.content[0].text) + tag_id = tag_data["id"] + assert tag_data["name"] == tag_name + assert tag_data["color"] == "FF5733" + logger.info(f"Created tag: {tag_name} (ID: {tag_id})") + + # 2. List tags โ€” should include the new tag + list_tags_result = await nc_mcp_client.call_tool( + "collectives_get_tags", + {"collective_id": cid}, + ) + assert list_tags_result.isError is False + tags_data = json.loads(list_tags_result.content[0].text) + tag_ids = [t["id"] for t in tags_data["tags"]] + assert tag_id in tag_ids + logger.info(f"Tag found in list ({tags_data['total']} tags)") + + # 3. Create a page to tag + page_result = await nc_mcp_client.call_tool( + "collectives_create_page", + { + "collective_id": cid, + "parent_id": landing_id, + "title": f"Tagged Page {uuid.uuid4().hex[:8]}", + }, + ) + assert page_result.isError is False + page_id = json.loads(page_result.content[0].text)["id"] + + # 4. Assign tag to page + assign_result = await nc_mcp_client.call_tool( + "collectives_assign_tag", + {"collective_id": cid, "page_id": page_id, "tag_id": tag_id}, + ) + assert assign_result.isError is False + logger.info(f"Tag {tag_id} assigned to page {page_id}") + + # 5. Remove tag from page + remove_result = await nc_mcp_client.call_tool( + "collectives_remove_tag", + {"collective_id": cid, "page_id": page_id, "tag_id": tag_id}, + ) + assert remove_result.isError is False + logger.info(f"Tag {tag_id} removed from page {page_id}") + + # Cleanup + await nc_mcp_client.call_tool( + "collectives_trash_page", + {"collective_id": cid, "page_id": page_id}, + ) + + +# --- Search --- + + +async def test_collectives_search( + nc_mcp_client: ClientSession, temporary_collective: dict +): + """Test full-text search within a collective.""" + cid = temporary_collective["id"] + + # Search for text in the landing page (contains "Welcome") + result = await nc_mcp_client.call_tool( + "collectives_search_pages", + {"collective_id": cid, "query": "Welcome"}, + ) + assert result.isError is False + + data = json.loads(result.content[0].text) + assert data["success"] is True + assert data["query"] == "Welcome" + assert data["collective_id"] == cid + # Search may or may not find results depending on indexing timing + logger.info(f"Search returned {data['total']} results for 'Welcome'") + + +# --- Error Handling --- + + +async def test_collectives_get_page_not_found(nc_mcp_client: ClientSession): + """Test getting a non-existent page returns an error.""" + result = await nc_mcp_client.call_tool( + "collectives_get_page", + {"collective_id": 999999, "page_id": 999999}, + ) + assert result.isError is True + logger.info("Non-existent page correctly returned error") + + +async def test_collectives_get_pages_not_found(nc_mcp_client: ClientSession): + """Test listing pages for a non-existent collective returns an error.""" + result = await nc_mcp_client.call_tool( + "collectives_get_pages", + {"collective_id": 999999}, + ) + assert result.isError is True + logger.info("Non-existent collective correctly returned error")