From dea882c2f5b02f443e8031da7ffaea6cd32bb835 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Tue, 6 May 2025 16:44:33 +0200 Subject: [PATCH] Fix tests --- nextcloud_mcp_server/client.py | 89 ++++++++++--- tests/conftest.py | 16 +++ tests/test_client.py | 181 ++++++++++++-------------- tests/test_note_attachment_cleanup.py | 73 +++++------ update_webdav_auth.py | 99 +++++++------- 5 files changed, 249 insertions(+), 209 deletions(-) create mode 100644 tests/conftest.py diff --git a/nextcloud_mcp_server/client.py b/nextcloud_mcp_server/client.py index 8e0bb82..2a2fe53 100644 --- a/nextcloud_mcp_server/client.py +++ b/nextcloud_mcp_server/client.py @@ -154,27 +154,68 @@ class NextcloudClient: "category": note.get("category"), "modified": note.get("modified"), } - ) - return search_results + ) + raise e + + def delete_webdav_resource(self, *, path: str): + """Delete a resource (file or directory) via WebDAV DELETE.""" + # Ensure path ends with a slash if it's a directory + if not path.endswith('/'): + # This is a heuristic; a more robust solution would check resource type first + # but for the specific case of deleting the attachment directory, this is acceptable. + path_with_slash = f"{path}/" + else: + path_with_slash = path + + webdav_path = f"{self._get_webdav_base_path()}/{path_with_slash.lstrip('/')}" + logger.info("Deleting WebDAV resource: %s", webdav_path) + + headers = {"OCS-APIRequest": "true"} + try: + response = self._client.delete(webdav_path, headers=headers) + response.raise_for_status() # Raises for 4xx/5xx status codes + logger.info("Successfully deleted WebDAV resource '%s' (Status: %s)", webdav_path, response.status_code) + # DELETE typically returns 204 No Content on success + return {"status_code": response.status_code} + + except HTTPStatusError as e: + logger.error( + "HTTP error deleting WebDAV resource '%s': %s", + webdav_path, + e, + ) + # It's expected to get a 404 if the resource doesn't exist, which is fine. + # We only re-raise if it's not a 404. + if e.response.status_code != 404: + raise e + else: + 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( + "Unexpected error deleting WebDAV resource '%s': %s", + webdav_path, + e, + ) + raise e def notes_delete_note(self, *, note_id: int): # First delete the note through the Notes API response = self._client.delete(f"/apps/notes/api/v1/notes/{note_id}") response.raise_for_status() json_response = response.json() - - # Note that we don't try to delete the attachments directory via WebDAV - # This is because Nextcloud Notes app does not delete the attachments - # when a note is deleted - this is the expected behavior of the app - # and not a bug. Attachments remain orphaned in the system. - - # If we did try to delete them, it would fail with a 401 Unauthorized - # or CSRF check error because WebDAV requires a different authentication - # method for DELETE operations than for GET/PUT operations. - - # The test_note_attachment_cleanup.py test explicitly verifies this behavior - # to ensure our understanding is correct. - + + # Now, attempt to delete the associated attachments directory via WebDAV + # Add a trailing slash as suggested + attachment_dir_path = f"Notes/.attachments.{note_id}/" + logger.info(f"Attempting to delete attachment directory for note {note_id} via WebDAV: {attachment_dir_path}") + try: + self.delete_webdav_resource(path=attachment_dir_path) + logger.info(f"Successfully attempted to delete attachment directory for note {note_id}.") + except Exception as e: + # Log the error but don't re-raise, as note deletion itself was successful + logger.error(f"Failed to delete attachment directory for note {note_id}: {e}") + return json_response # Removed incorrect get_note_attachment method that used Notes API @@ -200,14 +241,24 @@ class NextcloudClient: if not mime_type: mime_type = "application/octet-stream" # Default if guessing fails - headers = {"Content-Type": mime_type} + headers = {"Content-Type": mime_type, "OCS-APIRequest": "true"} try: # First check if we can access WebDAV at all with current credentials # 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") + logger.info("Authorization header for PROPFIND (Notes dir): %s", auth_header if auth_header else "Not present or generated by auth flow") + else: + logger.info("No httpx.Auth object configured on the client for PROPFIND (Notes dir).") + + 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, - headers={"Depth": "0"}) + headers=propfind_headers) if notes_dir_response.status_code == 401: logger.error("WebDAV authentication failed for Notes directory. Please verify WebDAV permissions.") @@ -226,7 +277,9 @@ class NextcloudClient: # Ensure the parent directory exists using MKCOL parent_dir_path = f"{webdav_base}/Notes/.attachments.{note_id}" logger.info("Creating attachments directory: %s", parent_dir_path) - mkcol_response = self._client.request("MKCOL", parent_dir_path) + mkcol_headers = {"OCS-APIRequest": "true"} + logger.info("Headers for MKCOL (Attachments dir): %s", mkcol_headers) + mkcol_response = self._client.request("MKCOL", parent_dir_path, headers=mkcol_headers) # MKCOL should return 201 Created or 405 Method Not Allowed (if exists) # We can ignore 405, but raise for other errors if mkcol_response.status_code not in [201, 405]: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..95aee8d --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,16 @@ +import pytest +import os +import logging +from nextcloud_mcp_server.client import NextcloudClient + +logger = logging.getLogger(__name__) + +@pytest.fixture(scope="session") +def nc_client() -> NextcloudClient: + """ + Fixture to create a NextcloudClient instance for integration tests. + """ + assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set" + assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set" + assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set" + return NextcloudClient.from_env() diff --git a/tests/test_client.py b/tests/test_client.py index 70b5ed4..5114f63 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -190,40 +190,32 @@ def test_note_attachment_integration(nc_client: NextcloudClient): attachment_mime = "text/plain" logger.info(f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}") - try: - # Try to add the attachment, but don't fail the test if WebDAV isn't available - upload_response = nc_client.add_note_attachment( - note_id=note_id, - filename=attachment_filename, - content=attachment_content, - mime_type=attachment_mime - ) - - # If we get here, WebDAV is working - continue with attachment tests - assert upload_response and "status_code" in upload_response - assert upload_response["status_code"] in [201, 204] - logger.info(f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']}).") - time.sleep(1) # Allow time for upload processing + # Assuming WebDAV should work now, directly call add_note_attachment + upload_response = nc_client.add_note_attachment( + note_id=note_id, + filename=attachment_filename, + content=attachment_content, + mime_type=attachment_mime + ) + + assert upload_response and "status_code" in upload_response + assert upload_response["status_code"] in [201, 204] + logger.info(f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']}).") + time.sleep(1) # Allow time for upload processing - # --- Get and Verify Attachment --- - logger.info(f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}") - retrieved_content, retrieved_mime = nc_client.get_note_attachment( - note_id=note_id, - filename=attachment_filename - ) - logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes") + # --- Get and Verify Attachment --- + logger.info(f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}") + retrieved_content, retrieved_mime = nc_client.get_note_attachment( + note_id=note_id, + filename=attachment_filename + ) + logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes") - # --- Verify Attachment --- - assert retrieved_content == attachment_content - # Check if the expected mime type is part of the retrieved one (to handle charset) - assert attachment_mime in retrieved_mime - logger.info("Retrieved attachment content and mime type verified successfully.") - - except HTTPStatusError as e: - if e.response.status_code == 401: - pytest.skip("Skipping attachment tests due to WebDAV permission issues (401 Unauthorized)") - else: - raise # Re-raise other HTTP errors + # --- Verify Attachment --- + assert retrieved_content == attachment_content + # Check if the expected mime type is part of the retrieved one (to handle charset) + assert attachment_mime in retrieved_mime + logger.info("Retrieved attachment content and mime type verified successfully.") finally: # --- Delete Note (Cleanup) --- @@ -274,39 +266,31 @@ def test_note_attachment_with_category_integration(nc_client: NextcloudClient): attachment_mime = "text/plain" logger.info(f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}") - try: - # Try to add the attachment, but don't fail the test if WebDAV isn't available - upload_response = nc_client.add_note_attachment( - note_id=note_id, - filename=attachment_filename, - content=attachment_content, - mime_type=attachment_mime - ) - - # If we get here, WebDAV is working - continue with attachment tests - assert upload_response and "status_code" in upload_response - assert upload_response["status_code"] in [201, 204] - logger.info(f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']}).") - time.sleep(1) + # Assuming WebDAV should work now, directly call add_note_attachment + upload_response = nc_client.add_note_attachment( + note_id=note_id, + filename=attachment_filename, + content=attachment_content, + mime_type=attachment_mime + ) + + assert upload_response and "status_code" in upload_response + assert upload_response["status_code"] in [201, 204] + logger.info(f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']}).") + time.sleep(1) - # --- Get and Verify Attachment --- - logger.info(f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}") - retrieved_content, retrieved_mime = nc_client.get_note_attachment( - note_id=note_id, - filename=attachment_filename - ) - logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes") + # --- Get and Verify Attachment --- + logger.info(f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}") + retrieved_content, retrieved_mime = nc_client.get_note_attachment( + note_id=note_id, + filename=attachment_filename + ) + logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes") - # --- Verify Attachment --- - assert retrieved_content == attachment_content - assert attachment_mime in retrieved_mime # Check if expected mime is part of retrieved - logger.info("Retrieved attachment content and mime type verified successfully for note with category.") - - except HTTPStatusError as e: - if e.response.status_code == 401: - pytest.skip("Skipping attachment tests due to WebDAV permission issues (401 Unauthorized)") - else: - raise # Re-raise other HTTP errors + # --- Verify Attachment --- + assert retrieved_content == attachment_content + assert attachment_mime in retrieved_mime # Check if expected mime is part of retrieved + logger.info("Retrieved attachment content and mime type verified successfully for note with category.") finally: # --- Delete Note (Cleanup) --- @@ -359,33 +343,33 @@ def test_attachment_cleanup_behavior(nc_client: NextcloudClient): attachment_content = f"Content for cleanup test".encode('utf-8') logger.info(f"Adding attachment '{attachment_filename}' to note ID: {note_id}") - try: - upload_response = nc_client.add_note_attachment( - note_id=note_id, - filename=attachment_filename, - content=attachment_content, - mime_type="text/plain" - ) - assert upload_response["status_code"] in [201, 204] - logger.info(f"Attachment added successfully (Status: {upload_response['status_code']}).") - time.sleep(1) - - # --- Verify Attachment Exists --- - retrieved_content, _ = nc_client.get_note_attachment( - note_id=note_id, - filename=attachment_filename - ) - assert retrieved_content == attachment_content - logger.info("Verified attachment exists and can be retrieved") - - # Attachment operations successful - continue with test - has_webdav_access = True - except HTTPStatusError as e: - if e.response.status_code == 401: - logger.info(f"WebDAV access denied (401 Unauthorized). Skipping attachment tests.") - pytest.skip("WebDAV access denied (401 Unauthorized)") - else: - raise # Re-raise other HTTP errors + # Removed try block as we expect WebDAV to work or fail the test + upload_response = nc_client.add_note_attachment( + note_id=note_id, + filename=attachment_filename, + content=attachment_content, + mime_type="text/plain" + ) + assert upload_response["status_code"] in [201, 204] + logger.info(f"Attachment added successfully (Status: {upload_response['status_code']}).") + time.sleep(1) + + # --- Verify Attachment Exists --- + retrieved_content, _ = nc_client.get_note_attachment( + note_id=note_id, + filename=attachment_filename + ) + assert retrieved_content == attachment_content + logger.info("Verified attachment exists and can be retrieved") + + # Attachment operations successful - continue with test + # has_webdav_access = True # No longer needed as we expect it to work or fail + # except HTTPStatusError as e: # Removed the try/except block that skipped on 401 + # if e.response.status_code == 401: + # logger.info(f"WebDAV access denied (401 Unauthorized). Skipping attachment tests.") + # pytest.skip("WebDAV access denied (401 Unauthorized)") + # else: + # raise # Re-raise other HTTP errors # --- Delete Note --- logger.info(f"Deleting note ID: {note_id}") @@ -399,17 +383,12 @@ def test_attachment_cleanup_behavior(nc_client: NextcloudClient): assert excinfo.value.response.status_code == 404 logger.info(f"Verified note deletion (404 received)") - # --- Document the expected behavior: attachments remain after note deletion --- - try: - # Try to get the attachment - expected to still exist - retrieved_content, _ = nc_client.get_note_attachment( + # --- Verify Attachment Is Deleted (New Behavior) --- + logger.info(f"Verifying attachment '{attachment_filename}' is deleted for note ID: {note_id}") + with pytest.raises(HTTPStatusError) as excinfo_attach_del: + nc_client.get_note_attachment( note_id=note_id, filename=attachment_filename ) - logger.info("EXPECTED BEHAVIOR: Attachment still exists after note deletion") - logger.info("This matches the behavior of the official Nextcloud Notes app") - except HTTPStatusError as e: - if e.response.status_code == 404: - logger.info("NOTE: Attachment was deleted with the note (unexpected but not a problem)") - else: - logger.info(f"Unexpected error when checking attachment: {e.response.status_code}") + assert excinfo_attach_del.value.response.status_code == 404 + logger.info(f"Attachment '{attachment_filename}' correctly not found (404) after note deletion.") diff --git a/tests/test_note_attachment_cleanup.py b/tests/test_note_attachment_cleanup.py index 61de4d6..cf8b558 100644 --- a/tests/test_note_attachment_cleanup.py +++ b/tests/test_note_attachment_cleanup.py @@ -10,31 +10,21 @@ logger = logging.getLogger(__name__) # Tests assume NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars are set -@pytest.fixture(scope="module") -def nc_client() -> NextcloudClient: - """ - Fixture to create a NextcloudClient instance for integration tests. - """ - assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set" - assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set" - assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set" - return NextcloudClient.from_env() - @pytest.mark.integration -def test_attachment_remains_after_note_deletion(nc_client: NextcloudClient): +def test_attachment_deleted_after_note_deletion(nc_client: NextcloudClient): """ - Test to verify and document that when a note is deleted, its attachments remain - in the system. This is the expected behavior of the Nextcloud Notes app. + Test to verify that when a note is deleted, its attachments are also deleted + by the MCP client's modified notes_delete_note method. """ # --- Create Note --- unique_id = str(uuid.uuid4()) note_title = f"Attachment Cleanup Test {unique_id}" - note_content = f"# Test for attachment cleanup behavior\n\nThis note will be deleted, but attachments should remain." + note_content = f"# Test for attachment cleanup behavior\n\nThis note and its attachments should be deleted." note_category = "CleanupTests" - + created_note = None note_id = None - + try: # Create the note logger.info(f"Creating note: {note_title}") @@ -47,11 +37,11 @@ def test_attachment_remains_after_note_deletion(nc_client: NextcloudClient): note_id = created_note["id"] logger.info(f"Note created with ID: {note_id}") time.sleep(1) - + # Create a simple text attachment - attachment_filename = f"orphan_test_{unique_id}.txt" + attachment_filename = f"cleanup_test_{unique_id}.txt" attachment_content = f"This is a test attachment for note {note_id}".encode('utf-8') - + # Attach the file to the note logger.info(f"Attaching text file to note {note_id}...") upload_response = nc_client.add_note_attachment( @@ -60,44 +50,43 @@ def test_attachment_remains_after_note_deletion(nc_client: NextcloudClient): content=attachment_content, mime_type="text/plain" ) - + assert upload_response["status_code"] in [201, 204] logger.info(f"Attachment added successfully (Status: {upload_response['status_code']}).") time.sleep(1) - - # Verify the attachment exists + + # Verify the attachment exists before deletion + logger.info(f"Verifying attachment exists before deletion...") content, mime_type = nc_client.get_note_attachment( note_id=note_id, filename=attachment_filename ) - - assert content == attachment_content, "Attachment content mismatch" - logger.info("Attachment verified") - - # Now delete the note + assert content == attachment_content, "Attachment content mismatch before deletion" + logger.info("Attachment verified before deletion") + + # Now delete the note (which should also delete the attachment directory) logger.info(f"Deleting note ID: {note_id}") nc_client.notes_delete_note(note_id=note_id) logger.info(f"Note deleted successfully.") time.sleep(1) - + # Verify the note is deleted with pytest.raises(HTTPStatusError) as excinfo: nc_client.notes_get_note(note_id=note_id) assert excinfo.value.response.status_code == 404 logger.info(f"Verified note deletion (404 Not Found)") - - # Now check if the attachment still exists (expected behavior: it should) - logger.info(f"Checking if attachment still exists after note deletion...") - orphaned_content, orphaned_mime = nc_client.get_note_attachment( - note_id=note_id, - filename=attachment_filename - ) - - # If we get here without an exception, the attachment still exists - logger.info("CONFIRMED: Attachment still exists after note deletion") - logger.info("This is the expected behavior of the Nextcloud Notes app") - assert orphaned_content == attachment_content, "Orphaned attachment content mismatch" - + + # Now check if the attachment is deleted (expected behavior: it should be) + logger.info(f"Checking if attachment is deleted after note deletion...") + with pytest.raises(HTTPStatusError) as excinfo: + nc_client.get_note_attachment( + note_id=note_id, + filename=attachment_filename + ) + # We expect a 404 because the attachment (and its directory) should be gone + assert excinfo.value.response.status_code == 404 + logger.info("CONFIRMED: Attachment is deleted after note deletion (404 Not Found)") + finally: - # No cleanup needed since we've already deleted the note + # No cleanup needed as the test itself cleans up the note and attachment pass diff --git a/update_webdav_auth.py b/update_webdav_auth.py index 0be4673..9a3210f 100644 --- a/update_webdav_auth.py +++ b/update_webdav_auth.py @@ -2,73 +2,76 @@ import sys import os import base64 -from nextcloud_mcp_server.client import NextcloudClient +from nextcloud_mcp_server.client import NextcloudClient, HTTPStatusError import logging +import time logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) -def test_webdav_auth(): +def test_webdav_auth_with_attachment(): """ - Test function to verify WebDAV authentication and compare with current implementation. + Test function to verify WebDAV authentication by attempting to use add_note_attachment. """ - # Create client using standard method client = NextcloudClient.from_env() print("Client authentication type:", type(client._client.auth).__name__) - # Get WebDAV base path username = os.environ["NEXTCLOUD_USERNAME"] - password = os.environ["NEXTCLOUD_PASSWORD"] webdav_base = client._get_webdav_base_path() - - # Test path for Notes directory notes_path = f"{webdav_base}/Notes" - print(f"Testing WebDAV access to: {notes_path}") - - # 1. Test with existing client auth + print(f"Target WebDAV Notes path for PROPFIND check: {notes_path}") + + temp_note_id = None try: - print("\nTest 1: Using existing client authentication") - response = client._client.request("PROPFIND", notes_path, headers={"Depth": "0"}) - print(f"Status code: {response.status_code}") - if response.status_code >= 400: - print(f"Error: {response.text}") - else: - print("Success! Current auth method works") - except Exception as e: - print(f"Error: {str(e)}") - - # 2. Test with explicit Authorization header - try: - print("\nTest 2: Using explicit Authorization header") - # Create base64 encoded credentials - auth_string = f"{username}:{password}" - auth_bytes = auth_string.encode('ascii') - base64_bytes = base64.b64encode(auth_bytes) - base64_auth = base64_bytes.decode('ascii') + # 1. Create a temporary note to get a note_id + print("\nCreating a temporary note...") + temp_note_title = f"Temp Note for WebDAV Test - {int(time.time())}" + created_note = client.notes_create_note(title=temp_note_title, content="Test content") + temp_note_id = created_note.get("id") + if not temp_note_id: + print("Error: Failed to create temporary note.") + return 1 + print(f"Temporary note created with ID: {temp_note_id}") + + # 2. Attempt to add an attachment (this will trigger the internal PROPFIND) + print(f"\nTest: Attempting add_note_attachment for note_id {temp_note_id} (uses client's BasicAuth)") + dummy_content = b"This is a test attachment." + dummy_filename = "test_attachment.txt" - # Make request with explicit Authorization header - headers = { - "Depth": "0", - "Authorization": f"Basic {base64_auth}" - } - - # Use client without auth to test explicit header - response = client._client.request( - "PROPFIND", - notes_path, - headers=headers, - auth=None # Override client auth + # The add_note_attachment method itself contains the PROPFIND check + # and will log details if it fails. + response_data = client.add_note_attachment( + note_id=temp_note_id, + filename=dummy_filename, + content=dummy_content, + mime_type="text/plain" ) - - print(f"Status code: {response.status_code}") - if response.status_code >= 400: - print(f"Error: {response.text}") + print(f"add_note_attachment response: {response_data}") + if response_data and response_data.get("status_code") in [201, 204]: + print("Success! add_note_attachment (and its internal PROPFIND) worked.") else: - print("Success! Explicit authorization header works") + print("Failure or unexpected response from add_note_attachment.") + # The client.py logs should show details of the PROPFIND if it failed. + + except HTTPStatusError as e: + print(f"HTTPStatusError during add_note_attachment: {e.response.status_code} - {e.response.text}") + if e.response.status_code == 401: + print("Reproduced 401 Unauthorized during add_note_attachment's PROPFIND check!") + else: + print("An HTTP error other than 401 occurred.") except Exception as e: - print(f"Error: {str(e)}") + print(f"An unexpected error occurred: {str(e)}") + finally: + # 3. Clean up: Delete the temporary note + if temp_note_id: + print(f"\nCleaning up: Deleting temporary note ID {temp_note_id}...") + try: + client.notes_delete_note(note_id=temp_note_id) + print(f"Successfully deleted temporary note ID {temp_note_id}.") + except Exception as e_del: + print(f"Error deleting temporary note ID {temp_note_id}: {str(e_del)}") return 0 if __name__ == "__main__": - sys.exit(test_webdav_auth()) + sys.exit(test_webdav_auth_with_attachment())