diff --git a/README.md b/README.md index af8ad55..4bd4b98 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ Currently, the server primarily interacts with the Nextcloud Notes API, providin * `nc_notes_create_note`: Create a new note. * `nc_notes_update_note`: Update an existing note by ID. +* `nc_notes_append_content`: Append content to an existing note with a clear separator. * `nc_notes_delete_note`: Delete a note by ID. * `nc_notes_search_notes`: Search notes by title or content. * `nc_get_note`: Get a specific note by ID. diff --git a/nextcloud_mcp_server/client.py b/nextcloud_mcp_server/client.py index fc52f4b..f066723 100644 --- a/nextcloud_mcp_server/client.py +++ b/nextcloud_mcp_server/client.py @@ -159,9 +159,37 @@ class NextcloudClient: except Exception as e: logger.error(f"Error cleaning up old attachment directory for note {note_id}: {e}") # Continue with update even if cleanup failed - + return updated_note + def notes_append_content(self, *, note_id: int, content: str): + """Append content to an existing note with a standard separator""" + logger.info(f"Appending content to note {note_id}") + + # Get current note + current_note = self.notes_get_note(note_id=note_id) + + # Use fixed separator for consistency + separator = "\n---\n" + + # Combine content + existing_content = current_note.get("content", "") + if existing_content: + new_content = existing_content + separator + content + else: + new_content = content # No separator needed for empty notes + + logger.info(f"Combining existing content ({len(existing_content)} chars) with new content ({len(content)} chars)") + + # Update with combined content + return self.notes_update_note( + note_id=note_id, + etag=current_note["etag"], + content=new_content, + title=None, # Keep existing title + category=None # Keep existing category + ) + def notes_search_notes(self, *, query: str): """ Search notes using token-based matching with relevance ranking. @@ -263,7 +291,7 @@ class NextcloudClient: # Construct path to old attachment directory old_category_path_part = f"{old_category}/" if old_category else "" old_attachment_dir_path = f"Notes/{old_category_path_part}.attachments.{note_id}/" - + logger.info(f"Cleaning up old attachment directory: {old_attachment_dir_path}") try: delete_result = self.delete_webdav_resource(path=old_attachment_dir_path) @@ -299,7 +327,7 @@ class NextcloudClient: logger.info(f"Resource '{webdav_path}' doesn't exist, no deletion needed.") return {"status_code": 404} # For other errors, continue with deletion attempt - + # Proceed with deletion response = self._client.delete(webdav_path, headers=headers) response.raise_for_status() # Raises for 4xx/5xx status codes @@ -308,7 +336,7 @@ class NextcloudClient: return {"status_code": response.status_code} except HTTPStatusError as e: - logger.error( + logger.warning( "HTTP error deleting WebDAV resource '%s': %s", webdav_path, e, @@ -321,7 +349,7 @@ class NextcloudClient: logger.info("Resource '%s' not found, no deletion needed.", webdav_path) return {"status_code": 404} # Indicate resource was not found except Exception as e: - logger.error( + logger.warning( "Unexpected error deleting WebDAV resource '%s': %s", webdav_path, e, @@ -334,20 +362,20 @@ class NextcloudClient: try: note_details = self.notes_get_note(note_id=note_id) category = note_details.get("category", "") - + # Check for other potential categories (if any note was moved between categories) # We can't reliably detect this without a dedicated tracking mechanism, but we can # implement a basic check for common category names and empty category potential_categories = [] if category: potential_categories.append(category) # Current category first - + # Add empty category (uncategorized notes) if category != "": potential_categories.append("") - + # We could add logic here to check for other common categories if needed - + logger.info(f"Note {note_id} has category: '{category}', will check attachment directories in: {potential_categories}") except HTTPStatusError as e: # If note doesn't exist (404), we can't delete attachments anyway. @@ -380,7 +408,7 @@ class NextcloudClient: except Exception as e: # Log the error but don't re-raise, as API note deletion itself was successful # Also, we want to try other potential categories even if one fails - logger.error(f"Failed during WebDAV deletion for category '{cat}' attachment directory: {e}") + logger.warning(f"Failed during WebDAV deletion for category '{cat}' attachment directory: {e}") return json_response @@ -409,7 +437,7 @@ class NextcloudClient: logger.info(f"Uploading attachment for note {note_id} (category: '{category or ''}') to WebDAV path: {attachment_path}") # Log current auth settings to diagnose the issue - logger.info("WebDAV auth settings - Username: %s, Auth Type: %s", + logger.info("WebDAV auth settings - Username: %s, Auth Type: %s", self.username, type(self._client.auth).__name__) if not mime_type: @@ -423,7 +451,7 @@ class NextcloudClient: # by checking the Notes directory notes_dir_path = f"{webdav_base}/Notes" logger.info("Testing WebDAV access to Notes directory: %s", notes_dir_path) - + # Log details of the auth being used by the client for this specific request if self._client.auth: auth_header = self._client.auth.auth_flow(self._client.build_request("GET", notes_dir_path)).__next__().headers.get("Authorization") @@ -433,9 +461,9 @@ class NextcloudClient: propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} logger.info("Headers for PROPFIND (Notes dir): %s", propfind_headers) - notes_dir_response = self._client.request("PROPFIND", notes_dir_path, + notes_dir_response = self._client.request("PROPFIND", notes_dir_path, headers=propfind_headers) - + if notes_dir_response.status_code == 401: logger.error("WebDAV authentication failed for Notes directory. Please verify WebDAV permissions.") raise HTTPStatusError( @@ -447,9 +475,9 @@ class NextcloudClient: logger.error("Error accessing WebDAV Notes directory: %s", notes_dir_response.status_code) notes_dir_response.raise_for_status() else: - logger.info("Successfully accessed WebDAV Notes directory (Status: %s)", + logger.info("Successfully accessed WebDAV Notes directory (Status: %s)", notes_dir_response.status_code) - + # Ensure the parent directory exists using MKCOL # parent_dir_path is now determined by the helper method logger.info("Ensuring attachments directory exists: %s", parent_dir_path) @@ -465,7 +493,7 @@ class NextcloudClient: ) mkcol_response.raise_for_status() else: - logger.info("Created/verified directory: %s (Status: %s)", + logger.info("Created/verified directory: %s (Status: %s)", parent_dir_path, mkcol_response.status_code) # Proceed with the PUT request diff --git a/nextcloud_mcp_server/server.py b/nextcloud_mcp_server/server.py index fce6e46..1f0457c 100644 --- a/nextcloud_mcp_server/server.py +++ b/nextcloud_mcp_server/server.py @@ -98,6 +98,14 @@ def nc_notes_update_note( ) +@mcp.tool() +def nc_notes_append_content(note_id: int, content: str, ctx: Context): + """Append content to an existing note with a clear separator""" + logger.info("Appending content to note %s", note_id) + client: NextcloudClient = ctx.request_context.lifespan_context.client + return client.notes_append_content(note_id=note_id, content=content) + + @mcp.tool() def nc_notes_search_notes(query: str, ctx: Context): """Search notes by title or content, returning only id, title, and category.""" diff --git a/tests/integration/test_notes_api.py b/tests/integration/test_notes_api.py index 92c444f..3126434 100644 --- a/tests/integration/test_notes_api.py +++ b/tests/integration/test_notes_api.py @@ -19,10 +19,10 @@ def test_notes_api_create_and_read(nc_client: NextcloudClient, temporary_note: d """ created_note_data = temporary_note # Get data from fixture note_id = created_note_data["id"] - + logger.info(f"Reading note created by fixture, ID: {note_id}") read_note = nc_client.notes_get_note(note_id=note_id) - + assert read_note["id"] == note_id assert read_note["title"] == created_note_data["title"] assert read_note["content"] == created_note_data["content"] @@ -40,7 +40,7 @@ def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict): update_title = f"Updated Title {uuid.uuid4().hex[:8]}" update_content = f"Updated Content {uuid.uuid4().hex[:8]}" - + logger.info(f"Attempting to update note ID: {note_id} with etag: {original_etag}") updated_note = nc_client.notes_update_note( note_id=note_id, @@ -50,7 +50,7 @@ def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict): # category=original_category # Explicitly pass category if required by update ) logger.info(f"Note updated: {updated_note}") - + assert updated_note["id"] == note_id assert updated_note["title"] == update_title assert updated_note["content"] == update_content @@ -111,4 +111,130 @@ def test_notes_api_delete_nonexistent(nc_client: NextcloudClient): assert excinfo.value.response.status_code == 404 logger.info(f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404.") +def test_notes_api_append_content_to_existing_note(nc_client: NextcloudClient, temporary_note: dict): + """ + Tests appending content to an existing note using the new append functionality. + """ + created_note_data = temporary_note + note_id = created_note_data["id"] + original_content = created_note_data["content"] + + append_text = f"Appended content {uuid.uuid4().hex[:8]}" + + logger.info(f"Appending content to note ID: {note_id}") + updated_note = nc_client.notes_append_content( + note_id=note_id, + content=append_text + ) + logger.info(f"Note after append: {updated_note}") + + # Verify the note was updated + assert updated_note["id"] == note_id + assert "etag" in updated_note + assert updated_note["etag"] != created_note_data["etag"] # Etag must change + + # Verify content has the separator and appended text + expected_content = original_content + "\n---\n" + append_text + assert updated_note["content"] == expected_content + + # Verify by reading the note again + time.sleep(1) # Allow potential propagation delay + read_note = nc_client.notes_get_note(note_id=note_id) + assert read_note["content"] == expected_content + logger.info(f"Successfully appended content to note ID: {note_id}") + +def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient): + """ + Tests appending content to an empty note (no separator should be added). + """ + # Create an empty note + test_title = f"Empty Note {uuid.uuid4().hex[:8]}" + test_category = "Test" + + logger.info(f"Creating empty note for append test") + empty_note = nc_client.notes_create_note( + title=test_title, + content="", # Empty content + category=test_category + ) + note_id = empty_note["id"] + + try: + append_text = f"First content {uuid.uuid4().hex[:8]}" + + logger.info(f"Appending content to empty note ID: {note_id}") + updated_note = nc_client.notes_append_content( + note_id=note_id, + content=append_text + ) + + # For empty notes, content should just be the appended text (no separator) + assert updated_note["content"] == append_text + + # Verify by reading the note again + time.sleep(1) + read_note = nc_client.notes_get_note(note_id=note_id) + assert read_note["content"] == append_text + logger.info(f"Successfully appended content to empty note ID: {note_id}") + + finally: + # Clean up the test note + try: + nc_client.notes_delete_note(note_id=note_id) + logger.info(f"Cleaned up test note ID: {note_id}") + except Exception as e: + logger.warning(f"Failed to clean up test note ID: {note_id}: {e}") + +def test_notes_api_append_content_multiple_times(nc_client: NextcloudClient, temporary_note: dict): + """ + Tests appending content multiple times to verify separator behavior. + """ + created_note_data = temporary_note + note_id = created_note_data["id"] + original_content = created_note_data["content"] + + first_append = f"First append {uuid.uuid4().hex[:8]}" + second_append = f"Second append {uuid.uuid4().hex[:8]}" + + logger.info(f"Performing multiple appends to note ID: {note_id}") + + # First append + updated_note = nc_client.notes_append_content( + note_id=note_id, + content=first_append + ) + + expected_content_after_first = original_content + "\n---\n" + first_append + assert updated_note["content"] == expected_content_after_first + + # Second append + updated_note = nc_client.notes_append_content( + note_id=note_id, + content=second_append + ) + + expected_content_after_second = expected_content_after_first + "\n---\n" + second_append + assert updated_note["content"] == expected_content_after_second + + # Verify by reading the note again + time.sleep(1) + read_note = nc_client.notes_get_note(note_id=note_id) + assert read_note["content"] == expected_content_after_second + logger.info(f"Successfully performed multiple appends to note ID: {note_id}") + +def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient): + """ + Tests that appending to a non-existent note fails with 404. + """ + non_existent_id = 999999999 + + logger.info(f"Attempting to append to non-existent note ID: {non_existent_id}") + with pytest.raises(HTTPStatusError) as excinfo: + nc_client.notes_append_content( + note_id=non_existent_id, + content="This should fail" + ) + assert excinfo.value.response.status_code == 404 + logger.info(f"Appending to non-existent note ID: {non_existent_id} correctly failed with 404.") + # --- Attachment tests moved to test_attachments.py ---