Compare commits

...

13 Commits

Author SHA1 Message Date
github-actions[bot] 9ae2a0fc6f bump: version 0.6.1 → 0.7.0 2025-08-03 12:47:13 +00:00
Chris Coutinho 8386644dfd Merge pull request #104 from cbcoutinho/feature/vcard
Initialize Contacts App
2025-08-03 14:46:48 +02:00
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 645 additions and 81 deletions
+1 -1
View File
@@ -16,7 +16,7 @@ jobs:
- name: Docker meta
id: meta
uses: docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804 # v5
uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5
with:
# list of Docker images to use as base name for tags
images: |
+13
View File
@@ -1,3 +1,16 @@
## v0.7.0 (2025-08-03)
### Feat
- **contacts**: Initialize Contacts App
## 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)
### 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. |
| **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. |
| **Contacts** | ✅ Full Support | Create, read, update, and delete contacts and address books via CardDAV. |
## 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_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
| 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.server import (
configure_calendar_tools,
configure_contacts_tools,
configure_notes_tools,
configure_tables_tools,
configure_webdav_tools,
@@ -56,6 +57,7 @@ configure_notes_tools(mcp)
configure_tables_tools(mcp)
configure_webdav_tools(mcp)
configure_calendar_tools(mcp)
configure_contacts_tools(mcp)
def run():
+2
View File
@@ -5,6 +5,7 @@ from httpx import AsyncClient, Auth, BasicAuth, Request, Response
from ..controllers.notes_search import NotesSearchController
from .calendar import CalendarClient
from .contacts import ContactsClient
from .notes import NotesClient
from .tables import TablesClient
from .webdav import WebDAVClient
@@ -43,6 +44,7 @@ class NextcloudClient:
self.webdav = WebDAVClient(self._client, username)
self.tables = TablesClient(self._client, username)
self.calendar = CalendarClient(self._client, username)
self.contacts = ContactsClient(self._client, username)
# Initialize controllers
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 .tables import configure_tables_tools
from .webdav import configure_webdav_tools
from .contacts import configure_contacts_tools
__all__ = [
"configure_calendar_tools",
"configure_notes_tools",
"configure_tables_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]
name = "nextcloud-mcp-server"
version = "0.6.0"
version = "0.7.0"
description = ""
authors = [
{name = "Chris Coutinho",email = "chris@coutinho.io"}
@@ -11,7 +11,8 @@ dependencies = [
"mcp[cli] (>=1.10,<1.11)",
"httpx (>=0.28.1,<0.29.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]
+81 -1
View File
@@ -13,7 +13,7 @@ from nextcloud_mcp_server.client import NextcloudClient
logger = logging.getLogger(__name__)
@pytest.fixture(scope="session")
@pytest.fixture(scope="module")
async def nc_client() -> AsyncGenerator[NextcloudClient, Any]:
"""
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,
# 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
@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
def test_calendar_name():
"""Unique calendar name for testing."""
@@ -32,16 +22,14 @@ def test_calendar_name():
@pytest.fixture
async def temporary_calendar(
calendar_test_client: NextcloudClient, test_calendar_name: str
):
async def temporary_calendar(nc_client: NextcloudClient, test_calendar_name: str):
"""Create a temporary calendar for testing and clean up afterward."""
calendar_name = test_calendar_name
try:
# Create a test calendar
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,
display_name=f"Test Calendar {calendar_name}",
description="Temporary calendar for integration testing",
@@ -62,16 +50,14 @@ async def temporary_calendar(
# Cleanup: Delete the temporary calendar
try:
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}")
except Exception as e:
logger.error(f"Error deleting temporary calendar {calendar_name}: {e}")
@pytest.fixture
async def temporary_event(
calendar_test_client: NextcloudClient, temporary_calendar: str
):
async def temporary_event(nc_client: NextcloudClient, temporary_calendar: str):
"""Create a temporary event for testing and clean up afterward."""
event_uid = None
calendar_name = temporary_calendar
@@ -91,9 +77,7 @@ async def temporary_event(
try:
logger.info(f"Creating temporary event in calendar: {calendar_name}")
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result.get("uid")
if not event_uid:
@@ -107,9 +91,7 @@ async def temporary_event(
if event_uid:
try:
logger.info(f"Cleaning up temporary event: {event_uid}")
await calendar_test_client.calendar.delete_event(
calendar_name, event_uid
)
await nc_client.calendar.delete_event(calendar_name, event_uid)
logger.info(f"Successfully deleted temporary event: {event_uid}")
except HTTPStatusError as e:
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."""
calendars = await calendar_test_client.calendar.list_calendars()
calendars = await nc_client.calendar.list_calendars()
assert isinstance(calendars, list)
@@ -144,7 +126,7 @@ async def test_list_calendars(calendar_test_client: NextcloudClient):
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."""
calendar_name = temporary_calendar
@@ -163,9 +145,7 @@ async def test_create_and_delete_event(
}
try:
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
assert "uid" in result
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}")
# 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
)
assert retrieved_event["uid"] == event_uid
@@ -181,9 +161,7 @@ async def test_create_and_delete_event(
assert retrieved_event["location"] == "Test Room"
# Delete event
delete_result = await calendar_test_client.calendar.delete_event(
calendar_name, event_uid
)
delete_result = await nc_client.calendar.delete_event(calendar_name, event_uid)
assert delete_result["status_code"] in [200, 204, 404]
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(
calendar_test_client: NextcloudClient, temporary_calendar: str
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating an all-day event."""
calendar_name = temporary_calendar
@@ -209,21 +187,19 @@ async def test_create_all_day_event(
}
try:
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created all-day event with UID: {event_uid}")
# Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event(
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "All Day Test Event"
assert retrieved_event.get("all_day") is True
# 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:
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(
calendar_test_client: NextcloudClient, temporary_calendar: str
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test creating a recurring event."""
calendar_name = temporary_calendar
@@ -248,30 +224,26 @@ async def test_create_recurring_event(
}
try:
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created recurring event with UID: {event_uid}")
# Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event(
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "Weekly Recurring Test"
assert retrieved_event.get("recurring") is True
# 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:
logger.error(f"Recurring event test failed: {e}")
raise
async def test_list_events_in_range(
calendar_test_client: NextcloudClient, temporary_event: dict
):
async def test_list_events_in_range(nc_client: NextcloudClient, temporary_event: dict):
"""Test listing events within a date range."""
calendar_name = temporary_event["calendar_name"]
@@ -279,7 +251,7 @@ async def test_list_events_in_range(
start_datetime = datetime.now()
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,
start_datetime=start_datetime,
end_datetime=end_datetime,
@@ -300,9 +272,7 @@ async def test_list_events_in_range(
assert "start_datetime" in event
async def test_update_event(
calendar_test_client: NextcloudClient, temporary_event: dict
):
async def test_update_event(nc_client: NextcloudClient, temporary_event: dict):
"""Test updating an existing event."""
calendar_name = temporary_event["calendar_name"]
event_uid = temporary_event["uid"]
@@ -316,15 +286,13 @@ async def test_update_event(
}
try:
result = await calendar_test_client.calendar.update_event(
result = await nc_client.calendar.update_event(
calendar_name, event_uid, updated_data
)
assert result["uid"] == event_uid
# Verify updates
updated_event, _ = await calendar_test_client.calendar.get_event(
calendar_name, event_uid
)
updated_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid)
assert updated_event["title"] == "Updated Test Event Title"
assert updated_event["description"] == "Updated description for test event"
assert updated_event["location"] == "Updated Location"
@@ -338,7 +306,7 @@ async def test_update_event(
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."""
calendar_name = temporary_calendar
@@ -356,14 +324,12 @@ async def test_create_event_with_attendees(
}
try:
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created event with attendees, UID: {event_uid}")
# Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event(
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
assert retrieved_event["title"] == "Meeting with Attendees"
@@ -371,7 +337,7 @@ async def test_create_event_with_attendees(
assert retrieved_event["status"] == "TENTATIVE"
# 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:
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(
calendar_test_client: NextcloudClient, temporary_calendar: str
nc_client: NextcloudClient, temporary_calendar: str
):
"""Test retrieving a non-existent event."""
calendar_name = temporary_calendar
fake_uid = f"nonexistent-{uuid.uuid4()}"
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
logger.info(f"Correctly got 404 for nonexistent event: {fake_uid}")
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."""
calendar_name = temporary_calendar
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
logger.info(f"Correctly got 404 for deleting nonexistent event: {fake_uid}")
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."""
calendar_name = temporary_calendar
@@ -423,14 +389,12 @@ async def test_event_with_url_and_categories(
}
try:
result = await calendar_test_client.calendar.create_event(
calendar_name, event_data
)
result = await nc_client.calendar.create_event(calendar_name, event_data)
event_uid = result["uid"]
logger.info(f"Created event with metadata, UID: {event_uid}")
# Verify event
retrieved_event, _ = await calendar_test_client.calendar.get_event(
retrieved_event, _ = await nc_client.calendar.get_event(
calendar_name, event_uid
)
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
# 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:
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(
calendar_test_client: NextcloudClient,
nc_client: NextcloudClient,
):
"""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}"
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")
+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
@pytest.fixture(scope="session")
@pytest.fixture(scope="module")
async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]:
"""
Fixture to get information about the sample table that comes with Nextcloud Tables.
Generated
+12 -1
View File
@@ -505,13 +505,14 @@ wheels = [
[[package]]
name = "nextcloud-mcp-server"
version = "0.6.0"
version = "0.7.0"
source = { editable = "." }
dependencies = [
{ name = "httpx" },
{ name = "icalendar" },
{ name = "mcp", extra = ["cli"] },
{ name = "pillow" },
{ name = "pythonvcard4" },
]
[package.dev-dependencies]
@@ -530,6 +531,7 @@ requires-dist = [
{ name = "icalendar", specifier = ">=6.0.0,<7.0.0" },
{ name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" },
{ name = "pillow", specifier = ">=11.2.1,<12.0.0" },
{ name = "pythonvcard4", specifier = ">=0.2.0" },
]
[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" },
]
[[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]]
name = "pyyaml"
version = "6.0.2"