Compare commits

...

11 Commits

Author SHA1 Message Date
Chris Coutinho 1dfdad5fad Update README, docstrings, and test scope for temporary_addressbook 2025-08-03 14:42:16 +02:00
Chris Coutinho 72cb62a101 test(contacts): Add unit/integration tests for a few tools 2025-08-03 14:36:16 +02:00
Chris Coutinho 21fc55320b Fix scoping 2025-08-03 14:25:01 +02:00
Chris Coutinho ad3e288203 test: Replace test_*_clients with single nc_client for tests 2025-08-03 14:22:45 +02:00
Chris Coutinho 0a97357a9c remove main.py 2025-08-03 14:17:29 +02:00
Chris Coutinho 70f01bf40a Add files 2025-08-03 14:16:55 +02:00
Chris Coutinho 37b1057d2a feat(contacts): Initialize Contacts App 2025-08-03 14:15:37 +02:00
Chris Coutinho ad95140416 Merge pull request #102 from cbcoutinho/renovate/docker-metadata-action-digest
chore(deps): update docker/metadata-action digest to c1e5197
2025-08-01 12:43:12 +02:00
github-actions[bot] 73fb56f73d bump: version 0.6.0 → 0.6.1 2025-08-01 10:41:12 +00:00
Chris Coutinho 9cc5300aa8 Merge pull request #96 from cbcoutinho/refactor/server
Refactor server tools and resources
2025-08-01 12:40:52 +02:00
renovate-bot-cbcoutinho[bot] acc505aa01 chore(deps): update docker/metadata-action digest to c1e5197 2025-08-01 10:06:53 +00:00
16 changed files with 639 additions and 81 deletions
+1 -1
View File
@@ -16,7 +16,7 @@ jobs:
- name: Docker meta - name: Docker meta
id: meta id: meta
uses: docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804 # v5 uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5
with: with:
# list of Docker images to use as base name for tags # list of Docker images to use as base name for tags
images: | images: |
+7
View File
@@ -1,3 +1,10 @@
## v0.6.1 (2025-08-01)
### Fix
- **calendar**: Fix iCalendar date vs datetime format
- **calendar**: Remove try/except in calendar API
## v0.6.0 (2025-07-29) ## v0.6.0 (2025-07-29)
### Feat ### Feat
+12
View File
@@ -16,6 +16,7 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
| **Calendar** | ✅ Full Support | Complete calendar integration - create, update, delete events. Support for recurring events, reminders, attendees, and all-day events via CalDAV. | | **Calendar** | ✅ Full Support | Complete calendar integration - create, update, delete events. Support for recurring events, reminders, attendees, and all-day events via CalDAV. |
| **Tables** | ⚠️ Row Operations | Read table schemas and perform CRUD operations on table rows. Table management not yet supported. | | **Tables** | ⚠️ Row Operations | Read table schemas and perform CRUD operations on table rows. Table management not yet supported. |
| **Files (WebDAV)** | ✅ Full Support | Complete file system access - browse directories, read/write files, create/delete resources. | | **Files (WebDAV)** | ✅ Full Support | Complete file system access - browse directories, read/write files, create/delete resources. |
| **Contacts** | ✅ Full Support | Create, read, update, and delete contacts and address books via CardDAV. |
## Available Tools ## Available Tools
@@ -46,6 +47,17 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
| `nc_calendar_bulk_operations` | **New:** Bulk update, delete, or move events matching filter criteria | | `nc_calendar_bulk_operations` | **New:** Bulk update, delete, or move events matching filter criteria |
| `nc_calendar_manage_calendar` | **New:** Create, delete, and manage calendar properties | | `nc_calendar_manage_calendar` | **New:** Create, delete, and manage calendar properties |
### Contacts Tools
| Tool | Description |
|------|-------------|
| `nc_contacts_list_addressbooks` | List all available addressbooks for the user |
| `nc_contacts_list_contacts` | List all contacts in a specific addressbook |
| `nc_contacts_create_addressbook` | Create a new addressbook |
| `nc_contacts_delete_addressbook` | Delete an addressbook |
| `nc_contacts_create_contact` | Create a new contact in an addressbook |
| `nc_contacts_delete_contact` | Delete a contact from an addressbook |
### Tables Tools ### Tables Tools
| Tool | Description | | Tool | Description |
+3
View File
@@ -0,0 +1,3 @@
#!/bin/bash
php /var/www/html/occ app:enable contacts
+2
View File
@@ -9,6 +9,7 @@ from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import setup_logging from nextcloud_mcp_server.config import setup_logging
from nextcloud_mcp_server.server import ( from nextcloud_mcp_server.server import (
configure_calendar_tools, configure_calendar_tools,
configure_contacts_tools,
configure_notes_tools, configure_notes_tools,
configure_tables_tools, configure_tables_tools,
configure_webdav_tools, configure_webdav_tools,
@@ -56,6 +57,7 @@ configure_notes_tools(mcp)
configure_tables_tools(mcp) configure_tables_tools(mcp)
configure_webdav_tools(mcp) configure_webdav_tools(mcp)
configure_calendar_tools(mcp) configure_calendar_tools(mcp)
configure_contacts_tools(mcp)
def run(): def run():
+2
View File
@@ -5,6 +5,7 @@ from httpx import AsyncClient, Auth, BasicAuth, Request, Response
from ..controllers.notes_search import NotesSearchController from ..controllers.notes_search import NotesSearchController
from .calendar import CalendarClient from .calendar import CalendarClient
from .contacts import ContactsClient
from .notes import NotesClient from .notes import NotesClient
from .tables import TablesClient from .tables import TablesClient
from .webdav import WebDAVClient from .webdav import WebDAVClient
@@ -43,6 +44,7 @@ class NextcloudClient:
self.webdav = WebDAVClient(self._client, username) self.webdav = WebDAVClient(self._client, username)
self.tables = TablesClient(self._client, username) self.tables = TablesClient(self._client, username)
self.calendar = CalendarClient(self._client, username) self.calendar = CalendarClient(self._client, username)
self.contacts = ContactsClient(self._client, username)
# Initialize controllers # Initialize controllers
self._notes_search = NotesSearchController() self._notes_search = NotesSearchController()
+235
View File
@@ -0,0 +1,235 @@
"""CardDAV client for NextCloud contacts operations."""
import logging
from .base import BaseNextcloudClient
import xml.etree.ElementTree as ET
from pythonvCard4.vcard import Contact
logger = logging.getLogger(__name__)
class ContactsClient(BaseNextcloudClient):
"""Client for NextCloud CardDAV contact operations."""
def _get_carddav_base_path(self) -> str:
"""Helper to get the base CardDAV path for contacts."""
return f"/remote.php/dav/addressbooks/users/{self.username}"
async def list_addressbooks(self):
"""List all available addressbooks for the user."""
carddav_path = self._get_carddav_base_path()
propfind_body = """<?xml version="1.0" encoding="utf-8"?>
<d:propfind xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
<d:prop>
<d:displayname/>
<d:getctag />
</d:prop>
</d:propfind>"""
headers = {
# "Depth": "0",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"PROPFIND", carddav_path, content=propfind_body, headers=headers
)
ns = {"d": "DAV:"}
# logger.info(response.content)
root = ET.fromstring(response.content)
addressbooks = []
for response_elem in root.findall(".//d:response", ns):
href = response_elem.find(".//d:href", ns)
if href is None:
continue
href_text = href.text or ""
if not href_text.endswith("/"):
continue # Skip non-addressbook resources
# Extract addressbook name from href
addressbook_name = href_text.rstrip("/").split("/")[-1]
if not addressbook_name or addressbook_name == self.username:
continue
# Get properties
propstat = response_elem.find(".//d:propstat", ns)
if propstat is None:
continue
prop = propstat.find(".//d:prop", ns)
if prop is None:
continue
displayname_elem = prop.find(".//d:displayname", ns)
displayname = (
displayname_elem.text
if displayname_elem is not None
else addressbook_name
)
getctag_elem = prop.find(".//d:getctag", ns)
getctag = getctag_elem.text if getctag_elem is not None else None
addressbooks.append(
{
"name": addressbook_name,
"display_name": displayname,
"getctag": getctag,
}
)
logger.debug(f"Found {len(addressbooks)} addressbooks")
return addressbooks
async def create_addressbook(self, *, name: str, display_name: str):
"""Create a new addressbook."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{name}/"
prop_body = f"""<?xml version="1.0" encoding="utf-8"?>
<d:mkcol xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:carddav">
<d:set>
<d:prop>
<d:resourcetype>
<d:collection/>
<c:addressbook/>
</d:resourcetype>
<d:displayname>{display_name}</d:displayname>
</d:prop>
</d:set>
</d:mkcol>"""
headers = {
"Content-Type": "application/xml",
}
await self._make_request("MKCOL", url, content=prop_body, headers=headers)
async def delete_addressbook(self, *, name: str):
"""Delete an addressbook."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{name}/"
await self._make_request("DELETE", url)
async def create_contact(self, *, addressbook: str, uid: str, contact_data: dict):
"""Create a new contact."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
contact = Contact(fn=contact_data.get("fn"), uid=uid)
if "email" in contact_data:
contact.email = [{"value": contact_data["email"], "type": ["HOME"]}]
if "tel" in contact_data:
contact.tel = [{"value": contact_data["tel"], "type": ["HOME"]}]
vcard = contact.to_vcard()
headers = {
"Content-Type": "text/vcard; charset=utf-8",
"If-None-Match": "*",
}
await self._make_request("PUT", url, content=vcard, headers=headers)
async def delete_contact(self, *, addressbook: str, uid: str):
"""Delete a contact."""
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
await self._make_request("DELETE", url)
async def list_contacts(self, *, addressbook: str):
"""List all available contacts for addressbook."""
carddav_path = self._get_carddav_base_path()
report_body = """<?xml version="1.0" encoding="utf-8"?>
<card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
<d:prop>
<d:getetag />
<card:address-data />
</d:prop>
</card:addressbook-query>"""
headers = {
"Depth": "1",
"Content-Type": "application/xml",
"Accept": "application/xml",
}
response = await self._make_request(
"REPORT",
f"{carddav_path}/{addressbook}",
content=report_body,
headers=headers,
)
ns = {"d": "DAV:", "card": "urn:ietf:params:xml:ns:carddav"}
# logger.info(response.text)
root = ET.fromstring(response.content)
contacts = []
for response_elem in root.findall(".//d:response", ns):
href = response_elem.find(".//d:href", ns)
if href is None:
logger.info("Skip missing href")
continue
href_text = href.text or ""
# logger.info("Href text: %s", href_text)
# if not href_text.endswith("/"):
# logger.info("# Skip non-addressbook resources")
# continue
# Extract vcard id from href
vcard_id = href_text.rstrip("/").split("/")[-1]
if not vcard_id:
logger.info("Skip missing vcard_id")
continue
vcard_id = vcard_id.replace(".vcf", "")
# Get properties
propstat = response_elem.find(".//d:propstat", ns)
if propstat is None:
logger.info("Skip missing propstat")
continue
prop = propstat.find(".//d:prop", ns)
if prop is None:
logger.info("Skip missing prop")
continue
getetag_elem = prop.find(".//d:getetag", ns)
getetag = getetag_elem.text if getetag_elem is not None else None
addressdata_elem = prop.find(".//card:address-data", ns)
addressdata = (
addressdata_elem.text if addressdata_elem is not None else None
)
if addressdata is None:
logger.info("Skip missing addressdata")
continue
contact = Contact.from_vcard(addressdata)
contacts.append(
{
"vcard_id": vcard_id,
"getetag": getetag,
"contact": {
"fullname": contact.fn,
"nickname": contact.nickname,
"birthday": contact.bday,
"email": contact.email,
},
"addressdata": addressdata,
}
)
logger.debug(f"Found {len(contacts)} contacts")
return contacts
+2
View File
@@ -2,10 +2,12 @@ from .calendar import configure_calendar_tools
from .notes import configure_notes_tools from .notes import configure_notes_tools
from .tables import configure_tables_tools from .tables import configure_tables_tools
from .webdav import configure_webdav_tools from .webdav import configure_webdav_tools
from .contacts import configure_contacts_tools
__all__ = [ __all__ = [
"configure_calendar_tools", "configure_calendar_tools",
"configure_notes_tools", "configure_notes_tools",
"configure_tables_tools", "configure_tables_tools",
"configure_webdav_tools", "configure_webdav_tools",
"configure_contacts_tools",
] ]
+65
View File
@@ -0,0 +1,65 @@
import logging
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
def configure_contacts_tools(mcp: FastMCP):
# Contacts tools
@mcp.tool()
async def nc_contacts_list_addressbooks(ctx: Context):
"""List all addressbooks for the user."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.list_addressbooks()
@mcp.tool()
async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str):
"""List all addressbooks for the user."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.list_contacts(addressbook=addressbook)
@mcp.tool()
async def nc_contacts_create_addressbook(
ctx: Context, *, name: str, display_name: str
):
"""Create a new addressbook.
Args:
name: The name of the addressbook.
display_name: The display name of the addressbook.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.create_addressbook(
name=name, display_name=display_name
)
@mcp.tool()
async def nc_contacts_delete_addressbook(ctx: Context, *, name: str):
"""Delete an addressbook."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.delete_addressbook(name=name)
@mcp.tool()
async def nc_contacts_create_contact(
ctx: Context, *, addressbook: str, uid: str, contact_data: dict
):
"""Create a new contact.
Args:
addressbook: The name of the addressbook to create the contact in.
uid: The unique ID for the contact.
contact_data: A dictionary with the contact's details, e.g. {"fn": "John Doe", "email": "john.doe@example.com"}.
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.create_contact(
addressbook=addressbook, uid=uid, contact_data=contact_data
)
@mcp.tool()
async def nc_contacts_delete_contact(ctx: Context, *, addressbook: str, uid: str):
"""Delete a contact."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
return await client.contacts.delete_contact(addressbook=addressbook, uid=uid)
+3 -2
View File
@@ -1,6 +1,6 @@
[project] [project]
name = "nextcloud-mcp-server" name = "nextcloud-mcp-server"
version = "0.6.0" version = "0.6.1"
description = "" description = ""
authors = [ authors = [
{name = "Chris Coutinho",email = "chris@coutinho.io"} {name = "Chris Coutinho",email = "chris@coutinho.io"}
@@ -11,7 +11,8 @@ dependencies = [
"mcp[cli] (>=1.10,<1.11)", "mcp[cli] (>=1.10,<1.11)",
"httpx (>=0.28.1,<0.29.0)", "httpx (>=0.28.1,<0.29.0)",
"pillow (>=11.2.1,<12.0.0)", "pillow (>=11.2.1,<12.0.0)",
"icalendar (>=6.0.0,<7.0.0)" "icalendar (>=6.0.0,<7.0.0)",
"pythonvcard4>=0.2.0",
] ]
[tool.pytest.ini_options] [tool.pytest.ini_options]
+81 -1
View File
@@ -13,7 +13,7 @@ from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@pytest.fixture(scope="session") @pytest.fixture(scope="module")
async def nc_client() -> AsyncGenerator[NextcloudClient, Any]: async def nc_client() -> AsyncGenerator[NextcloudClient, Any]:
""" """
Fixture to create a NextcloudClient instance for integration tests. Fixture to create a NextcloudClient instance for integration tests.
@@ -170,3 +170,83 @@ async def temporary_note_with_attachment(
# Note: The temporary_note fixture's finally block will handle note deletion, # Note: The temporary_note fixture's finally block will handle note deletion,
# which should also trigger the WebDAV directory deletion attempt. # which should also trigger the WebDAV directory deletion attempt.
@pytest.fixture(scope="module")
async def temporary_addressbook(nc_client: NextcloudClient):
"""
Fixture to create a temporary addressbook for a test and ensure its deletion afterward.
Yields the created addressbook dictionary.
"""
addressbook_name = f"test-addressbook-{uuid.uuid4().hex[:8]}"
logger.info(f"Creating temporary addressbook: {addressbook_name}")
try:
await nc_client.contacts.create_addressbook(
name=addressbook_name, display_name=f"Test Addressbook {addressbook_name}"
)
logger.info(f"Temporary addressbook created: {addressbook_name}")
yield addressbook_name
finally:
logger.info(f"Cleaning up temporary addressbook: {addressbook_name}")
try:
await nc_client.contacts.delete_addressbook(name=addressbook_name)
logger.info(
f"Successfully deleted temporary addressbook: {addressbook_name}"
)
except HTTPStatusError as e:
if e.response.status_code != 404:
logger.error(
f"HTTP error deleting temporary addressbook {addressbook_name}: {e}"
)
else:
logger.warning(
f"Temporary addressbook {addressbook_name} already deleted (404)."
)
except Exception as e:
logger.error(
f"Unexpected error deleting temporary addressbook {addressbook_name}: {e}"
)
@pytest.fixture
async def temporary_contact(nc_client: NextcloudClient, temporary_addressbook: str):
"""
Fixture to create a temporary contact in a temporary addressbook and ensure its deletion.
Yields the created contact's UID.
"""
contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
addressbook_name = temporary_addressbook
contact_data = {
"fn": "John Doe",
"email": "john.doe@example.com",
"tel": "1234567890",
}
logger.info(f"Creating temporary contact in addressbook: {addressbook_name}")
try:
await nc_client.contacts.create_contact(
addressbook=addressbook_name,
uid=contact_uid,
contact_data=contact_data,
)
logger.info(f"Temporary contact created with UID: {contact_uid}")
yield contact_uid
finally:
logger.info(f"Cleaning up temporary contact: {contact_uid}")
try:
await nc_client.contacts.delete_contact(
addressbook=addressbook_name, uid=contact_uid
)
logger.info(f"Successfully deleted temporary contact: {contact_uid}")
except HTTPStatusError as e:
if e.response.status_code != 404:
logger.error(
f"HTTP error deleting temporary contact {contact_uid}: {e}"
)
else:
logger.warning(
f"Temporary contact {contact_uid} already deleted (404)."
)
except Exception as e:
logger.error(
f"Unexpected error deleting temporary contact {contact_uid}: {e}"
)
+39 -75
View File
@@ -15,16 +15,6 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
@pytest.fixture
async def calendar_test_client():
"""Create a new, isolated NextcloudClient for calendar tests."""
client = NextcloudClient.from_env()
try:
yield client
finally:
await client.close()
@pytest.fixture @pytest.fixture
def test_calendar_name(): def test_calendar_name():
"""Unique calendar name for testing.""" """Unique calendar name for testing."""
@@ -32,16 +22,14 @@ def test_calendar_name():
@pytest.fixture @pytest.fixture
async def temporary_calendar( async def temporary_calendar(nc_client: NextcloudClient, test_calendar_name: str):
calendar_test_client: NextcloudClient, test_calendar_name: str
):
"""Create a temporary calendar for testing and clean up afterward.""" """Create a temporary calendar for testing and clean up afterward."""
calendar_name = test_calendar_name calendar_name = test_calendar_name
try: try:
# Create a test calendar # Create a test calendar
logger.info(f"Creating temporary calendar: {calendar_name}") logger.info(f"Creating temporary calendar: {calendar_name}")
result = await calendar_test_client.calendar.create_calendar( result = await nc_client.calendar.create_calendar(
calendar_name=calendar_name, calendar_name=calendar_name,
display_name=f"Test Calendar {calendar_name}", display_name=f"Test Calendar {calendar_name}",
description="Temporary calendar for integration testing", description="Temporary calendar for integration testing",
@@ -62,16 +50,14 @@ async def temporary_calendar(
# Cleanup: Delete the temporary calendar # Cleanup: Delete the temporary calendar
try: try:
logger.info(f"Cleaning up temporary calendar: {calendar_name}") logger.info(f"Cleaning up temporary calendar: {calendar_name}")
await calendar_test_client.calendar.delete_calendar(calendar_name) await nc_client.calendar.delete_calendar(calendar_name)
logger.info(f"Successfully deleted temporary calendar: {calendar_name}") logger.info(f"Successfully deleted temporary calendar: {calendar_name}")
except Exception as e: except Exception as e:
logger.error(f"Error deleting temporary calendar {calendar_name}: {e}") logger.error(f"Error deleting temporary calendar {calendar_name}: {e}")
@pytest.fixture @pytest.fixture
async def temporary_event( async def temporary_event(nc_client: NextcloudClient, temporary_calendar: str):
calendar_test_client: NextcloudClient, temporary_calendar: str
):
"""Create a temporary event for testing and clean up afterward.""" """Create a temporary event for testing and clean up afterward."""
event_uid = None event_uid = None
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -91,9 +77,7 @@ async def temporary_event(
try: try:
logger.info(f"Creating temporary event in calendar: {calendar_name}") logger.info(f"Creating temporary event in calendar: {calendar_name}")
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
event_uid = result.get("uid") event_uid = result.get("uid")
if not event_uid: if not event_uid:
@@ -107,9 +91,7 @@ async def temporary_event(
if event_uid: if event_uid:
try: try:
logger.info(f"Cleaning up temporary event: {event_uid}") logger.info(f"Cleaning up temporary event: {event_uid}")
await calendar_test_client.calendar.delete_event( await nc_client.calendar.delete_event(calendar_name, event_uid)
calendar_name, event_uid
)
logger.info(f"Successfully deleted temporary event: {event_uid}") logger.info(f"Successfully deleted temporary event: {event_uid}")
except HTTPStatusError as e: except HTTPStatusError as e:
if e.response.status_code != 404: if e.response.status_code != 404:
@@ -120,9 +102,9 @@ async def temporary_event(
) )
async def test_list_calendars(calendar_test_client: NextcloudClient): async def test_list_calendars(nc_client: NextcloudClient):
"""Test listing available calendars.""" """Test listing available calendars."""
calendars = await calendar_test_client.calendar.list_calendars() calendars = await nc_client.calendar.list_calendars()
assert isinstance(calendars, list) assert isinstance(calendars, list)
@@ -144,7 +126,7 @@ async def test_list_calendars(calendar_test_client: NextcloudClient):
async def test_create_and_delete_event( async def test_create_and_delete_event(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test creating and deleting a basic event.""" """Test creating and deleting a basic event."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -163,9 +145,7 @@ async def test_create_and_delete_event(
} }
try: try:
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
assert "uid" in result assert "uid" in result
assert result["status_code"] in [200, 201, 204] assert result["status_code"] in [200, 201, 204]
@@ -173,7 +153,7 @@ async def test_create_and_delete_event(
logger.info(f"Created event with UID: {event_uid}") logger.info(f"Created event with UID: {event_uid}")
# Verify event was created by retrieving it # Verify event was created by retrieving it
retrieved_event, etag = await calendar_test_client.calendar.get_event( retrieved_event, etag = await nc_client.calendar.get_event(
calendar_name, event_uid calendar_name, event_uid
) )
assert retrieved_event["uid"] == event_uid assert retrieved_event["uid"] == event_uid
@@ -181,9 +161,7 @@ async def test_create_and_delete_event(
assert retrieved_event["location"] == "Test Room" assert retrieved_event["location"] == "Test Room"
# Delete event # Delete event
delete_result = await calendar_test_client.calendar.delete_event( delete_result = await nc_client.calendar.delete_event(calendar_name, event_uid)
calendar_name, event_uid
)
assert delete_result["status_code"] in [200, 204, 404] assert delete_result["status_code"] in [200, 204, 404]
logger.info(f"Successfully deleted event: {event_uid}") logger.info(f"Successfully deleted event: {event_uid}")
@@ -194,7 +172,7 @@ async def test_create_and_delete_event(
async def test_create_all_day_event( async def test_create_all_day_event(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test creating an all-day event.""" """Test creating an all-day event."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -209,21 +187,19 @@ async def test_create_all_day_event(
} }
try: try:
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
event_uid = result["uid"] event_uid = result["uid"]
logger.info(f"Created all-day event with UID: {event_uid}") logger.info(f"Created all-day event with UID: {event_uid}")
# Verify event # Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event( retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid calendar_name, event_uid
) )
assert retrieved_event["title"] == "All Day Test Event" assert retrieved_event["title"] == "All Day Test Event"
assert retrieved_event.get("all_day") is True assert retrieved_event.get("all_day") is True
# Cleanup # Cleanup
await calendar_test_client.calendar.delete_event(calendar_name, event_uid) await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e: except Exception as e:
logger.error(f"All-day event test failed: {e}") logger.error(f"All-day event test failed: {e}")
@@ -231,7 +207,7 @@ async def test_create_all_day_event(
async def test_create_recurring_event( async def test_create_recurring_event(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test creating a recurring event.""" """Test creating a recurring event."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -248,30 +224,26 @@ async def test_create_recurring_event(
} }
try: try:
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
event_uid = result["uid"] event_uid = result["uid"]
logger.info(f"Created recurring event with UID: {event_uid}") logger.info(f"Created recurring event with UID: {event_uid}")
# Verify event # Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event( retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid calendar_name, event_uid
) )
assert retrieved_event["title"] == "Weekly Recurring Test" assert retrieved_event["title"] == "Weekly Recurring Test"
assert retrieved_event.get("recurring") is True assert retrieved_event.get("recurring") is True
# Cleanup # Cleanup
await calendar_test_client.calendar.delete_event(calendar_name, event_uid) await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e: except Exception as e:
logger.error(f"Recurring event test failed: {e}") logger.error(f"Recurring event test failed: {e}")
raise raise
async def test_list_events_in_range( async def test_list_events_in_range(nc_client: NextcloudClient, temporary_event: dict):
calendar_test_client: NextcloudClient, temporary_event: dict
):
"""Test listing events within a date range.""" """Test listing events within a date range."""
calendar_name = temporary_event["calendar_name"] calendar_name = temporary_event["calendar_name"]
@@ -279,7 +251,7 @@ async def test_list_events_in_range(
start_datetime = datetime.now() start_datetime = datetime.now()
end_datetime = datetime.now() + timedelta(days=7) end_datetime = datetime.now() + timedelta(days=7)
events = await calendar_test_client.calendar.get_calendar_events( events = await nc_client.calendar.get_calendar_events(
calendar_name=calendar_name, calendar_name=calendar_name,
start_datetime=start_datetime, start_datetime=start_datetime,
end_datetime=end_datetime, end_datetime=end_datetime,
@@ -300,9 +272,7 @@ async def test_list_events_in_range(
assert "start_datetime" in event assert "start_datetime" in event
async def test_update_event( async def test_update_event(nc_client: NextcloudClient, temporary_event: dict):
calendar_test_client: NextcloudClient, temporary_event: dict
):
"""Test updating an existing event.""" """Test updating an existing event."""
calendar_name = temporary_event["calendar_name"] calendar_name = temporary_event["calendar_name"]
event_uid = temporary_event["uid"] event_uid = temporary_event["uid"]
@@ -316,15 +286,13 @@ async def test_update_event(
} }
try: try:
result = await calendar_test_client.calendar.update_event( result = await nc_client.calendar.update_event(
calendar_name, event_uid, updated_data calendar_name, event_uid, updated_data
) )
assert result["uid"] == event_uid assert result["uid"] == event_uid
# Verify updates # Verify updates
updated_event, _ = await calendar_test_client.calendar.get_event( updated_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid)
calendar_name, event_uid
)
assert updated_event["title"] == "Updated Test Event Title" assert updated_event["title"] == "Updated Test Event Title"
assert updated_event["description"] == "Updated description for test event" assert updated_event["description"] == "Updated description for test event"
assert updated_event["location"] == "Updated Location" assert updated_event["location"] == "Updated Location"
@@ -338,7 +306,7 @@ async def test_update_event(
async def test_create_event_with_attendees( async def test_create_event_with_attendees(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test creating an event with attendees.""" """Test creating an event with attendees."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -356,14 +324,12 @@ async def test_create_event_with_attendees(
} }
try: try:
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
event_uid = result["uid"] event_uid = result["uid"]
logger.info(f"Created event with attendees, UID: {event_uid}") logger.info(f"Created event with attendees, UID: {event_uid}")
# Verify event # Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event( retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid calendar_name, event_uid
) )
assert retrieved_event["title"] == "Meeting with Attendees" assert retrieved_event["title"] == "Meeting with Attendees"
@@ -371,7 +337,7 @@ async def test_create_event_with_attendees(
assert retrieved_event["status"] == "TENTATIVE" assert retrieved_event["status"] == "TENTATIVE"
# Cleanup # Cleanup
await calendar_test_client.calendar.delete_event(calendar_name, event_uid) await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e: except Exception as e:
logger.error(f"Event with attendees test failed: {e}") logger.error(f"Event with attendees test failed: {e}")
@@ -379,33 +345,33 @@ async def test_create_event_with_attendees(
async def test_get_nonexistent_event( async def test_get_nonexistent_event(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test retrieving a non-existent event.""" """Test retrieving a non-existent event."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
fake_uid = f"nonexistent-{uuid.uuid4()}" fake_uid = f"nonexistent-{uuid.uuid4()}"
with pytest.raises(HTTPStatusError) as exc_info: with pytest.raises(HTTPStatusError) as exc_info:
await calendar_test_client.calendar.get_event(calendar_name, fake_uid) await nc_client.calendar.get_event(calendar_name, fake_uid)
assert exc_info.value.response.status_code == 404 assert exc_info.value.response.status_code == 404
logger.info(f"Correctly got 404 for nonexistent event: {fake_uid}") logger.info(f"Correctly got 404 for nonexistent event: {fake_uid}")
async def test_delete_nonexistent_event( async def test_delete_nonexistent_event(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test deleting a non-existent event.""" """Test deleting a non-existent event."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
fake_uid = f"nonexistent-{uuid.uuid4()}" fake_uid = f"nonexistent-{uuid.uuid4()}"
result = await calendar_test_client.calendar.delete_event(calendar_name, fake_uid) result = await nc_client.calendar.delete_event(calendar_name, fake_uid)
assert result["status_code"] == 404 assert result["status_code"] == 404
logger.info(f"Correctly got 404 for deleting nonexistent event: {fake_uid}") logger.info(f"Correctly got 404 for deleting nonexistent event: {fake_uid}")
async def test_event_with_url_and_categories( async def test_event_with_url_and_categories(
calendar_test_client: NextcloudClient, temporary_calendar: str nc_client: NextcloudClient, temporary_calendar: str
): ):
"""Test creating an event with URL and multiple categories.""" """Test creating an event with URL and multiple categories."""
calendar_name = temporary_calendar calendar_name = temporary_calendar
@@ -423,14 +389,12 @@ async def test_event_with_url_and_categories(
} }
try: try:
result = await calendar_test_client.calendar.create_event( result = await nc_client.calendar.create_event(calendar_name, event_data)
calendar_name, event_data
)
event_uid = result["uid"] event_uid = result["uid"]
logger.info(f"Created event with metadata, UID: {event_uid}") logger.info(f"Created event with metadata, UID: {event_uid}")
# Verify event # Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event( retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid calendar_name, event_uid
) )
assert retrieved_event["title"] == "Event with URL and Categories" assert retrieved_event["title"] == "Event with URL and Categories"
@@ -441,7 +405,7 @@ async def test_event_with_url_and_categories(
assert retrieved_event.get("priority") == 2 assert retrieved_event.get("priority") == 2
# Cleanup # Cleanup
await calendar_test_client.calendar.delete_event(calendar_name, event_uid) await nc_client.calendar.delete_event(calendar_name, event_uid)
except Exception as e: except Exception as e:
logger.error(f"Event with metadata test failed: {e}") logger.error(f"Event with metadata test failed: {e}")
@@ -449,7 +413,7 @@ async def test_event_with_url_and_categories(
async def test_calendar_operations_error_handling( async def test_calendar_operations_error_handling(
calendar_test_client: NextcloudClient, nc_client: NextcloudClient,
): ):
"""Test error handling for calendar operations.""" """Test error handling for calendar operations."""
@@ -457,6 +421,6 @@ async def test_calendar_operations_error_handling(
fake_calendar = f"nonexistent_calendar_{uuid.uuid4().hex}" fake_calendar = f"nonexistent_calendar_{uuid.uuid4().hex}"
with pytest.raises(HTTPStatusError): with pytest.raises(HTTPStatusError):
await calendar_test_client.calendar.get_calendar_events(fake_calendar) await nc_client.calendar.get_calendar_events(fake_calendar)
logger.info("Error handling tests completed successfully") logger.info("Error handling tests completed successfully")
+86
View File
@@ -0,0 +1,86 @@
"""Integration tests for Contacts MCP tools."""
import logging
import uuid
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_contacts_workflow(
nc_mcp_client: ClientSession, nc_client: NextcloudClient
):
"""Test complete Contacts workflow via MCP tools with verification via NextcloudClient."""
addressbook_name = f"mcp-test-addressbook-{uuid.uuid4().hex[:8]}"
unique_suffix = uuid.uuid4().hex[:8]
contact_uid = f"mcp-contact-{unique_suffix}"
contact_data = {
"fn": f"MCP Contact {unique_suffix}",
"email": f"mcp.contact.{unique_suffix}@example.com",
"tel": "1234567890",
}
try:
# 1. Create address book via MCP
logger.info(f"Creating address book via MCP: {addressbook_name}")
create_ab_result = await nc_mcp_client.call_tool(
"nc_contacts_create_addressbook",
{"name": addressbook_name, "display_name": f"MCP Test {addressbook_name}"},
)
assert create_ab_result.isError is False
# 2. Verify address book creation
addressbooks = await nc_client.contacts.list_addressbooks()
assert any(ab["name"] == addressbook_name for ab in addressbooks)
# 3. Create contact via MCP
logger.info(f"Creating contact in {addressbook_name} via MCP")
create_c_result = await nc_mcp_client.call_tool(
"nc_contacts_create_contact",
{
"addressbook": addressbook_name,
"uid": contact_uid,
"contact_data": contact_data,
},
)
assert create_c_result.isError is False
# 4. Verify contact creation
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
assert any(c["vcard_id"] == contact_uid for c in contacts)
# 5. Delete contact via MCP
logger.info(f"Deleting contact {contact_uid} via MCP")
delete_c_result = await nc_mcp_client.call_tool(
"nc_contacts_delete_contact",
{"addressbook": addressbook_name, "uid": contact_uid},
)
assert delete_c_result.isError is False
# 6. Verify contact deletion
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
assert not any(c["vcard_id"] == contact_uid for c in contacts)
# 7. Delete address book via MCP
logger.info(f"Deleting address book {addressbook_name} via MCP")
delete_ab_result = await nc_mcp_client.call_tool(
"nc_contacts_delete_addressbook", {"name": addressbook_name}
)
assert delete_ab_result.isError is False
# 8. Verify address book deletion
addressbooks = await nc_client.contacts.list_addressbooks()
assert not any(ab["name"] == addressbook_name for ab in addressbooks)
finally:
# Cleanup in case of failure
try:
await nc_client.contacts.delete_addressbook(name=addressbook_name)
except Exception:
pass
@@ -0,0 +1,88 @@
"""Integration tests for Contacts CardDAV operations."""
import logging
import uuid
import pytest
from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
# Mark all tests in this module as integration tests
pytestmark = pytest.mark.integration
async def test_list_addressbooks(nc_client: NextcloudClient):
"""Test listing available addressbooks."""
addressbooks = await nc_client.contacts.list_addressbooks()
assert isinstance(addressbooks, list)
if not addressbooks:
pytest.skip("No addressbooks available - Contacts app may not be enabled")
logger.info(f"Found {len(addressbooks)} addressbooks")
# Check structure of addressbooks
for addressbook in addressbooks:
assert "name" in addressbook
assert "display_name" in addressbook
assert "getctag" in addressbook
logger.info(
f"Addressbook: {addressbook['name']} - {addressbook['display_name']}"
)
async def test_create_and_delete_addressbook(
nc_client: NextcloudClient, temporary_addressbook: str
):
"""Test creating and deleting a basic addressbook."""
addressbooks = await nc_client.contacts.list_addressbooks()
addressbook_names = [ab["name"] for ab in addressbooks]
assert temporary_addressbook in addressbook_names
async def test_list_contacts(
nc_client: NextcloudClient, temporary_addressbook: str, temporary_contact: str
):
"""Test listing contacts in an addressbook."""
contacts = await nc_client.contacts.list_contacts(addressbook=temporary_addressbook)
contact_uids = [c["vcard_id"] for c in contacts]
assert temporary_contact in contact_uids
async def test_full_contact_workflow(
nc_client: NextcloudClient, temporary_addressbook: str
):
"""Test the full workflow of creating, retrieving, and deleting a contact."""
addressbook_name = temporary_addressbook
contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
contact_data = {
"fn": "Jane Doe",
"email": "jane.doe@example.com",
"tel": "9876543210",
}
# Create contact
await nc_client.contacts.create_contact(
addressbook=addressbook_name,
uid=contact_uid,
contact_data=contact_data,
)
# Verify contact was created by listing
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
contact_uids = [c["vcard_id"] for c in contacts]
assert contact_uid in contact_uids
# Delete contact
await nc_client.contacts.delete_contact(
addressbook=addressbook_name, uid=contact_uid
)
# Verify contact was deleted
contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
contact_uids = [c["vcard_id"] for c in contacts]
assert contact_uid not in contact_uids
+1 -1
View File
@@ -14,7 +14,7 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
@pytest.fixture(scope="session") @pytest.fixture(scope="module")
async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]: async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]:
""" """
Fixture to get information about the sample table that comes with Nextcloud Tables. Fixture to get information about the sample table that comes with Nextcloud Tables.
Generated
+12 -1
View File
@@ -505,13 +505,14 @@ wheels = [
[[package]] [[package]]
name = "nextcloud-mcp-server" name = "nextcloud-mcp-server"
version = "0.6.0" version = "0.6.1"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "httpx" }, { name = "httpx" },
{ name = "icalendar" }, { name = "icalendar" },
{ name = "mcp", extra = ["cli"] }, { name = "mcp", extra = ["cli"] },
{ name = "pillow" }, { name = "pillow" },
{ name = "pythonvcard4" },
] ]
[package.dev-dependencies] [package.dev-dependencies]
@@ -530,6 +531,7 @@ requires-dist = [
{ name = "icalendar", specifier = ">=6.0.0,<7.0.0" }, { name = "icalendar", specifier = ">=6.0.0,<7.0.0" },
{ name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" }, { name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" },
{ name = "pillow", specifier = ">=11.2.1,<12.0.0" }, { name = "pillow", specifier = ">=11.2.1,<12.0.0" },
{ name = "pythonvcard4", specifier = ">=0.2.0" },
] ]
[package.metadata.requires-dev] [package.metadata.requires-dev]
@@ -843,6 +845,15 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/45/58/38b5afbc1a800eeea951b9285d3912613f2603bdf897a4ab0f4bd7f405fc/python_multipart-0.0.20-py3-none-any.whl", hash = "sha256:8a62d3a8335e06589fe01f2a3e178cdcc632f3fbe0d492ad9ee0ec35aab1f104", size = 24546, upload-time = "2024-12-16T19:45:44.423Z" }, { url = "https://files.pythonhosted.org/packages/45/58/38b5afbc1a800eeea951b9285d3912613f2603bdf897a4ab0f4bd7f405fc/python_multipart-0.0.20-py3-none-any.whl", hash = "sha256:8a62d3a8335e06589fe01f2a3e178cdcc632f3fbe0d492ad9ee0ec35aab1f104", size = 24546, upload-time = "2024-12-16T19:45:44.423Z" },
] ]
[[package]]
name = "pythonvcard4"
version = "0.2.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/34/04/02d5952a9d8cbcb9e62b4fc4f6f842e8d43aead6e307f83e6fd6f7352fbd/pythonvcard4-0.2.0.tar.gz", hash = "sha256:236bba2769e459645cfa776407ff07856aced45b437116bf40ddb39bbcefdb6d", size = 5530, upload-time = "2025-04-26T23:18:48.963Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/0d/2f/ee10d88bbe12e4e9e06f81589d999687038e5cd5fec6c05aed57c50aede6/pythonvcard4-0.2.0-py3-none-any.whl", hash = "sha256:dce31355dd50aee537f8883de86f301510e407bc1755a68ec8d5055b64f5c660", size = 5890, upload-time = "2025-04-26T23:18:48.2Z" },
]
[[package]] [[package]]
name = "pyyaml" name = "pyyaml"
version = "6.0.2" version = "6.0.2"