From 72cb62a1018747bf29af25323c7f1e7764b35321 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:36:16 +0200 Subject: [PATCH] test(contacts): Add unit/integration tests for a few tools --- nextcloud_mcp_server/app.py | 2 +- nextcloud_mcp_server/client/contacts.py | 58 ++++++++++++ nextcloud_mcp_server/server/contacts.py | 32 +++++++ tests/conftest.py | 80 +++++++++++++++++ tests/integration/test_contacts_mcp.py | 86 ++++++++++++++++++ tests/integration/test_contacts_operations.py | 88 +++++++++++++++++++ 6 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_contacts_mcp.py create mode 100644 tests/integration/test_contacts_operations.py diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index e66fa65..c34971d 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -9,10 +9,10 @@ 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, - configure_contacts_tools, ) setup_logging() diff --git a/nextcloud_mcp_server/client/contacts.py b/nextcloud_mcp_server/client/contacts.py index ce8801a..11ecc3f 100644 --- a/nextcloud_mcp_server/client/contacts.py +++ b/nextcloud_mcp_server/client/contacts.py @@ -87,6 +87,62 @@ class ContactsClient(BaseNextcloudClient): 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""" + + + + + + + + {display_name} + + + """ + + 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.""" @@ -135,6 +191,7 @@ class ContactsClient(BaseNextcloudClient): 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) @@ -162,6 +219,7 @@ class ContactsClient(BaseNextcloudClient): contacts.append( { + "vcard_id": vcard_id, "getetag": getetag, "contact": { "fullname": contact.fn, diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py index d749010..13f89a3 100644 --- a/nextcloud_mcp_server/server/contacts.py +++ b/nextcloud_mcp_server/server/contacts.py @@ -20,3 +20,35 @@ def configure_contacts_tools(mcp: FastMCP): """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.""" + 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.""" + 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) diff --git a/tests/conftest.py b/tests/conftest.py index cb45de8..bf43c95 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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 +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}" + ) diff --git a/tests/integration/test_contacts_mcp.py b/tests/integration/test_contacts_mcp.py new file mode 100644 index 0000000..f391fa8 --- /dev/null +++ b/tests/integration/test_contacts_mcp.py @@ -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 diff --git a/tests/integration/test_contacts_operations.py b/tests/integration/test_contacts_operations.py new file mode 100644 index 0000000..85628dc --- /dev/null +++ b/tests/integration/test_contacts_operations.py @@ -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