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

2025-08-03 14:36:16 +02:00
parent 21fc55320b
commit 72cb62a101
6 changed files with 345 additions and 1 deletions
@@ -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()
@@ -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"""<?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 @@ 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,
@@ -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)
@@ -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}"
+            )
@@ -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