test(contacts): Add unit/integration tests for a few tools

Author: Chris Coutinho

nextcloud_mcp_server/app.py

@@ -9,10 +9,10 @@
 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()

nextcloud_mcp_server/client/contacts.py

@@ -87,6 +87,62 @@ async def list_addressbooks(self):
         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."""
 
@@ -135,6 +191,7 @@ async def list_contacts(self, *, addressbook: str):
             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 @@ async def list_contacts(self, *, addressbook: str):
 
             contacts.append(
                 {
+                    "vcard_id": vcard_id,
                     "getetag": getetag,
                     "contact": {
                         "fullname": contact.fn,

nextcloud_mcp_server/server/contacts.py

@@ -20,3 +20,35 @@ 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."""
+        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)

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": "[email protected]",
+        "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}"
+            )

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

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": "[email protected]",
+        "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