import json 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_connectivity(nc_mcp_client: ClientSession): """Test basic MCP server connectivity and list available tools/resources.""" # List available tools tools = await nc_mcp_client.list_tools() logger.info("Available MCP tools:") tool_names = [] for tool in tools.tools: logger.info(" - %s: %s", tool.name, tool.description) tool_names.append(tool.name) # Verify expected tools are present expected_tools = [ "nc_notes_create_note", "nc_notes_update_note", "nc_notes_append_content", "nc_notes_search_notes", "nc_notes_delete_note", "nc_tables_list_tables", "nc_tables_get_schema", "nc_tables_read_table", "nc_tables_insert_row", "nc_tables_update_row", "nc_tables_delete_row", "nc_webdav_list_directory", "nc_webdav_read_file", "nc_webdav_write_file", "nc_webdav_create_directory", "nc_webdav_delete_resource", "nc_webdav_move_resource", "nc_webdav_copy_resource", "nc_webdav_search_files", "nc_webdav_find_by_name", "nc_webdav_find_by_type", "nc_webdav_list_favorites", "nc_calendar_list_calendars", "nc_calendar_create_event", "nc_calendar_list_events", "nc_calendar_get_event", "nc_calendar_update_event", "nc_calendar_delete_event", "nc_calendar_create_meeting", "nc_calendar_get_upcoming_events", "nc_calendar_find_availability", "nc_calendar_bulk_operations", "nc_calendar_manage_calendar", "nc_calendar_list_todos", "nc_calendar_create_todo", "nc_calendar_update_todo", "nc_calendar_delete_todo", "nc_calendar_search_todos", "deck_create_board", "nc_cookbook_import_recipe", "nc_cookbook_list_recipes", "nc_cookbook_get_recipe", "nc_cookbook_create_recipe", "nc_cookbook_update_recipe", "nc_cookbook_delete_recipe", "nc_cookbook_search_recipes", "nc_cookbook_list_categories", "nc_cookbook_get_recipes_in_category", "nc_cookbook_list_keywords", "nc_cookbook_get_recipes_with_keywords", "nc_cookbook_set_config", "nc_cookbook_reindex", ] for expected_tool in expected_tools: assert expected_tool in tool_names, ( f"Expected tool '{expected_tool}' not found in available tools" ) # List available resource templates templates = await nc_mcp_client.list_resource_templates() logger.info("\nAvailable resource templates:") template_uris = [] for template in templates.resourceTemplates: logger.info(" - %s", template.uriTemplate) template_uris.append(template.uriTemplate) # Verify expected resource templates # Note: Notes attachments are now handled via tools, not resource templates expected_templates = [] for expected_template in expected_templates: assert expected_template in template_uris, ( f"Expected template '{expected_template}' not found" ) # List available resources resources = await nc_mcp_client.list_resources() logger.info("\nAvailable resources:") resource_uris = [] for resource in resources.resources: logger.info(" - %s: %s", resource.uri, resource.name) resource_uris.append(str(resource.uri)) # Convert to string for comparison # Verify expected resources expected_resources = [ "nc://capabilities", "notes://settings", "nc://Deck/boards", "cookbook://version", "cookbook://config", ] for expected_resource in expected_resources: assert expected_resource in resource_uris, ( f"Expected resource '{expected_resource}' not found" ) # List available prompts prompts = await nc_mcp_client.list_prompts() logger.info("\nAvailable prompts:") for prompt in prompts.prompts: logger.info(" - %s", prompt.name) async def test_mcp_notes_crud_workflow( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test complete Notes CRUD workflow via MCP tools with verification via NextcloudClient.""" unique_suffix = uuid.uuid4().hex[:8] test_title = f"MCP Test Note {unique_suffix}" test_content = f"This is test content for note {unique_suffix}" test_category = "MCPTesting" created_note = None try: # 1. Create note via MCP logger.info("Creating note via MCP: %s", test_title) create_result = await nc_mcp_client.call_tool( "nc_notes_create_note", {"title": test_title, "content": test_content, "category": test_category}, ) assert create_result.isError is False, ( f"MCP note creation failed: {create_result.content}" ) created_note = create_result.content[0].text note_data = json.loads(created_note) # Parse the returned JSON note_id = note_data["id"] create_etag = note_data["etag"] # Verify create response includes ETag logger.info("Note created via MCP with ID: %s, ETag: %s", note_id, create_etag) assert "etag" in note_data, "Create response should include ETag" assert create_etag, "Create ETag should not be empty" # 2. Verify creation via direct NextcloudClient direct_note = await nc_client.notes.get_note(note_id) assert direct_note["title"] == test_title, ( f"Title mismatch: {direct_note['title']} != {test_title}" ) assert direct_note["content"] == test_content, "Content mismatch" assert direct_note["category"] == test_category, "Category mismatch" # 3. Read note via MCP logger.info("Reading note via MCP: %s", note_id) read_result = await nc_mcp_client.call_tool( "nc_notes_get_note", {"note_id": note_id} ) read_note_data = read_result.content[0].text read_note_data = json.loads(read_note_data) assert read_note_data["title"] == test_title assert read_note_data["content"] == test_content assert read_note_data["category"] == test_category # 4. Update note via MCP updated_title = f"Updated {test_title}" updated_content = f"Updated content: {test_content}" etag = read_note_data["etag"] logger.info("Updating note via MCP: %s", note_id) update_result = await nc_mcp_client.call_tool( "nc_notes_update_note", { "note_id": note_id, "etag": etag, "title": updated_title, "content": updated_content, "category": test_category, }, ) assert update_result.isError is False, ( f"MCP note update failed: {update_result.content}" ) # Verify update response includes new ETag updated_note_data = json.loads(update_result.content[0].text) update_etag = updated_note_data["etag"] logger.info("Note updated via MCP, new ETag: %s", update_etag) assert "etag" in updated_note_data, "Update response should include ETag" assert update_etag, "Update ETag should not be empty" assert update_etag != etag, "ETag should change after update" # 5. Verify update via direct NextcloudClient updated_direct_note = await nc_client.notes.get_note(note_id) assert updated_direct_note["title"] == updated_title assert updated_direct_note["content"] == updated_content # 6. Append content via MCP append_content = "\n\nThis is appended content via MCP." logger.info("Appending content to note via MCP: %s", note_id) append_result = await nc_mcp_client.call_tool( "nc_notes_append_content", {"note_id": note_id, "content": append_content} ) assert append_result.isError is False, ( f"MCP note append failed: {append_result.content}" ) # Verify append response includes new ETag appended_note_data = json.loads(append_result.content[0].text) append_etag = appended_note_data["etag"] logger.info("Content appended via MCP, new ETag: %s", append_etag) assert "etag" in appended_note_data, "Append response should include ETag" assert append_etag, "Append ETag should not be empty" assert append_etag != update_etag, "ETag should change after append" # 7. Verify append via direct NextcloudClient appended_direct_note = await nc_client.notes.get_note(note_id) assert append_content in appended_direct_note["content"] # 8. Search for note via MCP logger.info("Searching for note via MCP with query: %s", unique_suffix) search_result = await nc_mcp_client.call_tool( "nc_notes_search_notes", {"query": unique_suffix} ) assert search_result.isError is False, ( f"MCP note search failed: {search_result.content}" ) search_notes_text = search_result.content[0].text logger.info("Search result text: %s", search_notes_text) search_response = json.loads(search_notes_text) # Expect structured response with Pydantic format assert isinstance(search_response, dict), ( f"Expected search response to be a dict with structured format, got: {type(search_response)}" ) assert "results" in search_response, ( f"Expected 'results' field in search response, got keys: {list(search_response.keys())}" ) assert "success" in search_response and search_response["success"], ( f"Expected successful search response, got: {search_response}" ) search_notes = search_response["results"] assert isinstance(search_notes, list), ( f"Expected results to be a list, got: {type(search_notes)}" ) # Find our note in search results found_note = None for note in search_notes: if isinstance(note, dict) and note.get("id") == note_id: found_note = note break assert found_note is not None, ( f"Created note not found in search results. Search returned: {search_response}" ) assert found_note["title"] == updated_title # 9. Delete note via MCP logger.info("Deleting note via MCP: %s", note_id) delete_result = await nc_mcp_client.call_tool( "nc_notes_delete_note", {"note_id": note_id} ) assert delete_result.isError is False, ( f"MCP note deletion failed: {delete_result.content}" ) # 10. Verify deletion via direct NextcloudClient try: await nc_client.notes.get_note(note_id) pytest.fail("Note should have been deleted but was still found") except Exception: # Expected - note should be deleted logger.info("Successfully verified note %s was deleted", note_id) created_note = None # Mark as cleaned up finally: # Cleanup in case of test failure if created_note is not None: try: note_data = json.loads(created_note) await nc_client.notes.delete_note(note_data["id"]) logger.info("Cleaned up note %s after test failure", note_data["id"]) except Exception as e: logger.warning("Failed to cleanup note: %s", e) async def test_mcp_notes_etag_conflict( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test that MCP note updates fail when using stale ETags.""" unique_suffix = uuid.uuid4().hex[:8] test_title = f"ETag Test Note {unique_suffix}" test_content = f"This is test content for ETag testing {unique_suffix}" test_category = "ETTesting" created_note = None try: # 1. Create note via MCP logger.info("Creating note for ETag conflict test: %s", test_title) create_result = await nc_mcp_client.call_tool( "nc_notes_create_note", {"title": test_title, "content": test_content, "category": test_category}, ) assert create_result.isError is False note_data = json.loads(create_result.content[0].text) note_id = note_data["id"] original_etag = note_data["etag"] created_note = note_data # 2. Update note to change ETag first_update_result = await nc_mcp_client.call_tool( "nc_notes_update_note", { "note_id": note_id, "etag": original_etag, "title": f"First Update {test_title}", "content": test_content, "category": test_category, }, ) assert first_update_result.isError is False updated_data = json.loads(first_update_result.content[0].text) new_etag = updated_data["etag"] assert new_etag != original_etag, "ETag should have changed after update" # 3. Try to update with the stale (original) ETag - this should fail logger.info("Attempting update with stale ETag: %s", original_etag) conflict_result = await nc_mcp_client.call_tool( "nc_notes_update_note", { "note_id": note_id, "etag": original_etag, # Use stale ETag "title": "This should fail", "content": "This update should be rejected", "category": test_category, }, ) # 4. Verify the update was rejected with a 412 error # With McpError, tools now return proper error responses assert conflict_result.isError is True, "Update with stale ETag should fail" response_content = conflict_result.content[0].text assert "modified by someone else" in response_content, ( f"Expected conflict error message, got: {response_content}" ) logger.info("Successfully verified ETag conflict handling via MCP") finally: # Cleanup if created_note is not None: try: await nc_client.notes.delete_note(created_note["id"]) logger.info("Cleaned up test note %s", created_note["id"]) except Exception as e: logger.warning("Failed to cleanup test note: %s", e) async def test_mcp_webdav_workflow( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test WebDAV file operations via MCP tools with verification via NextcloudClient.""" unique_suffix = uuid.uuid4().hex[:8] test_dir = f"mcp_test_dir_{unique_suffix}" test_file = f"mcp_test_file_{unique_suffix}.txt" test_file_path = f"{test_dir}/{test_file}" test_content = f"This is test content for MCP WebDAV testing {unique_suffix}" try: # 1. Create directory via MCP logger.info("Creating directory via MCP: %s", test_dir) create_dir_result = await nc_mcp_client.call_tool( "nc_webdav_create_directory", {"path": test_dir} ) assert create_dir_result.isError is False, ( f"MCP directory creation failed: {create_dir_result.content}" ) # 2. Verify directory creation via direct WebDAV dir_listing = await nc_client.webdav.list_directory("") dir_names = [item["name"] for item in dir_listing if item["is_directory"]] assert test_dir in dir_names, f"Directory {test_dir} not found in root listing" # 3. Write file via MCP logger.info("Writing file via MCP: %s", test_file_path) write_result = await nc_mcp_client.call_tool( "nc_webdav_write_file", { "path": test_file_path, "content": test_content, "content_type": "text/plain", }, ) assert write_result.isError is False, ( f"MCP file write failed: {write_result.content}" ) # 4. Verify file creation via direct WebDAV file_listing = await nc_client.webdav.list_directory(test_dir) file_names = [item["name"] for item in file_listing if not item["is_directory"]] assert test_file in file_names, ( f"File {test_file} not found in directory listing" ) # 5. Read file via MCP logger.info("Reading file via MCP: %s", test_file_path) read_result = await nc_mcp_client.call_tool( "nc_webdav_read_file", {"path": test_file_path} ) assert read_result.isError is False, ( f"MCP file read failed: {read_result.content}" ) read_data = json.loads(read_result.content[0].text) assert read_data["content"] == test_content, "File content mismatch" assert read_data["path"] == test_file_path assert "text/plain" in read_data["content_type"] # 6. Verify file content via direct WebDAV direct_content, direct_content_type = await nc_client.webdav.read_file( test_file_path ) assert direct_content.decode("utf-8") == test_content # 7. List directory via MCP logger.info("Listing directory via MCP: %s", test_dir) list_result = await nc_mcp_client.call_tool( "nc_webdav_list_directory", {"path": test_dir} ) assert list_result.isError is False, ( f"MCP directory listing failed: {list_result.content}" ) listing_text = list_result.content[0].text logger.info("Directory listing response: %s", listing_text) listing_data = json.loads(listing_text) # Extract files from DirectoryListing response assert "files" in listing_data, "Expected 'files' field in directory listing" files = listing_data["files"] assert isinstance(files, list), ( f"Expected files to be a list, got: {type(files)}" ) # Verify metadata assert listing_data["path"] == test_dir assert listing_data["total_count"] >= 1 assert listing_data["files_count"] >= 1 # Find our file in the listing found_file = None for item in files: if isinstance(item, dict) and item.get("name") == test_file: found_file = item break assert found_file is not None, ( f"File {test_file} not found in MCP directory listing" ) assert found_file["is_directory"] is False assert found_file["size"] == len(test_content.encode("utf-8")) finally: # Cleanup try: logger.info("Cleaning up test file: %s", test_file_path) await nc_mcp_client.call_tool( "nc_webdav_delete_resource", {"path": test_file_path} ) logger.info("Cleaning up test directory: %s", test_dir) await nc_mcp_client.call_tool( "nc_webdav_delete_resource", {"path": test_dir} ) except Exception as e: logger.warning("Failed to cleanup WebDAV resources: %s", e) async def test_mcp_resources_access( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test accessing MCP resources and compare with direct API calls.""" # 1. Test capabilities resource logger.info("Testing capabilities resource via MCP") caps_result = await nc_mcp_client.read_resource("nc://capabilities") assert len(caps_result.contents) == 1 mcp_capabilities = json.loads(caps_result.contents[0].text) # Compare with direct API call direct_capabilities = await nc_client.capabilities() # Basic validation - both should have similar structure # Both return full OCS response structure assert "ocs" in mcp_capabilities assert "data" in mcp_capabilities["ocs"] assert "version" in mcp_capabilities["ocs"]["data"] assert "ocs" in direct_capabilities assert "data" in direct_capabilities["ocs"] assert "version" in direct_capabilities["ocs"]["data"] # 2. Test notes settings resource logger.info("Testing notes settings resource via MCP") settings_result = await nc_mcp_client.read_resource("notes://settings") assert len(settings_result.contents) == 1 mcp_settings = json.loads(settings_result.contents[0].text) # Compare with direct API call direct_settings = await nc_client.notes.get_settings() # Both should have settings data assert isinstance(mcp_settings, dict) assert isinstance(direct_settings, dict) logger.info("Successfully verified MCP resources match direct API calls") async def test_mcp_calendar_workflow( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test complete Calendar workflow via MCP tools with verification via NextcloudClient.""" unique_suffix = uuid.uuid4().hex[:8] test_event_title = f"MCP Test Event {unique_suffix}" test_location = f"MCP Test Location {unique_suffix}" created_event = None calendar_name = None try: # 1. List calendars via MCP logger.info("Listing calendars via MCP") calendars_result = await nc_mcp_client.call_tool( "nc_calendar_list_calendars", {} ) assert calendars_result.isError is False, ( f"MCP calendar listing failed: {calendars_result.content}" ) calendars_response = json.loads(calendars_result.content[0].text) # Debug output to understand the structure logger.info("calendars_response type: %s", type(calendars_response)) logger.info("calendars_response content: %s", calendars_response) # Expect structured response with Pydantic format assert isinstance(calendars_response, dict), ( f"Expected calendar response to be a dict with structured format, got: {type(calendars_response)}" ) assert "calendars" in calendars_response, ( f"Expected 'calendars' field in response, got keys: {list(calendars_response.keys())}" ) assert "success" in calendars_response and calendars_response["success"], ( f"Expected successful calendar response, got: {calendars_response}" ) calendars_list = calendars_response["calendars"] assert isinstance(calendars_list, list), ( f"Expected calendars to be a list, got: {type(calendars_list)}" ) if not calendars_list: pytest.skip("No calendars available for testing") # Use the first available calendar calendar_name = calendars_list[0]["name"] logger.info("Using calendar: %s", calendar_name) # 2. Create event via MCP from datetime import datetime, timedelta tomorrow = datetime.now() + timedelta(days=1) start_datetime = tomorrow.strftime("%Y-%m-%dT14:00:00") end_datetime = tomorrow.strftime("%Y-%m-%dT15:00:00") event_data = { "calendar_name": calendar_name, "title": test_event_title, "start_datetime": start_datetime, "end_datetime": end_datetime, "description": f"Test event created via MCP {unique_suffix}", "location": test_location, "categories": "testing,mcp", "status": "CONFIRMED", "priority": 5, } logger.info("Creating event via MCP: %s", test_event_title) create_result = await nc_mcp_client.call_tool( "nc_calendar_create_event", event_data ) assert create_result.isError is False, ( f"MCP event creation failed: {create_result.content}" ) created_event_data = json.loads(create_result.content[0].text) event_uid = created_event_data["uid"] created_event = {"uid": event_uid, "calendar_name": calendar_name} logger.info("Event created via MCP with UID: %s", event_uid) # 3. Verify creation via direct NextcloudClient direct_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid) assert direct_event["title"] == test_event_title assert direct_event["location"] == test_location assert "testing" in direct_event.get("categories", "") # 4. Get event via MCP logger.info("Getting event via MCP: %s", event_uid) get_result = await nc_mcp_client.call_tool( "nc_calendar_get_event", {"calendar_name": calendar_name, "event_uid": event_uid}, ) assert get_result.isError is False, ( f"MCP event get failed: {get_result.content}" ) get_event_data = json.loads(get_result.content[0].text) assert get_event_data["title"] == test_event_title assert get_event_data["location"] == test_location # 5. **TEST nc_calendar_list_events - This is the main tool we're testing** logger.info("Testing nc_calendar_list_events via MCP") # Get today and next week for date range today = datetime.now() next_week = today + timedelta(days=7) start_date = today.strftime("%Y-%m-%d") end_date = next_week.strftime("%Y-%m-%d") list_events_data = { "calendar_name": calendar_name, "start_date": start_date, "end_date": end_date, "limit": 50, "location_contains": "MCP Test", "title_contains": unique_suffix, } list_result = await nc_mcp_client.call_tool( "nc_calendar_list_events", list_events_data ) assert list_result.isError is False, ( f"MCP list events failed: {list_result.content}" ) events_response = json.loads(list_result.content[0].text) # Debug output to understand what nc_calendar_list_events returns logger.info("list_events result type: %s", type(events_response)) logger.info("list_events result content: %s", events_response) # Response is now a ListEventsResponse with an "events" field assert isinstance(events_response, dict), "Expected response dict" events_data = events_response.get("events", []) assert isinstance(events_data, list), "Expected events list" # Our created event should be in the list found_event = None for event in events_data: if event.get("uid") == event_uid: found_event = event break assert found_event is not None, ( f"Created event {event_uid} not found in events list" ) assert found_event["summary"] == test_event_title # 6. Test list events across all calendars logger.info("Testing nc_calendar_list_events across all calendars") all_calendars_data = { "calendar_name": "", # Will be ignored "search_all_calendars": True, "start_date": start_date, "end_date": end_date, "title_contains": unique_suffix, } all_list_result = await nc_mcp_client.call_tool( "nc_calendar_list_events", all_calendars_data ) assert all_list_result.isError is False, ( f"MCP list all events failed: {all_list_result.content}" ) all_events_response = json.loads(all_list_result.content[0].text) # Response is now a ListEventsResponse with an "events" field assert isinstance(all_events_response, dict), "Expected response dict" all_events_data = all_events_response.get("events", []) assert isinstance(all_events_data, list), "Expected events list" # Our event should still be found when searching all calendars found_in_all = any(event.get("uid") == event_uid for event in all_events_data) assert found_in_all, "Event not found when searching all calendars" # 7. Update event via MCP updated_title = f"Updated {test_event_title}" updated_description = f"Updated description {unique_suffix}" update_data = { "calendar_name": calendar_name, "event_uid": event_uid, "title": updated_title, "description": updated_description, "priority": 1, } logger.info("Updating event via MCP: %s", event_uid) update_result = await nc_mcp_client.call_tool( "nc_calendar_update_event", update_data ) assert update_result.isError is False, ( f"MCP event update failed: {update_result.content}" ) # 8. Verify update via direct NextcloudClient updated_direct_event, _ = await nc_client.calendar.get_event( calendar_name, event_uid ) assert updated_direct_event["title"] == updated_title assert updated_direct_event["description"] == updated_description assert updated_direct_event["priority"] == 1 # 9. Test upcoming events via MCP logger.info("Testing nc_calendar_get_upcoming_events via MCP") upcoming_result = await nc_mcp_client.call_tool( "nc_calendar_get_upcoming_events", {"calendar_name": calendar_name, "days_ahead": 7, "limit": 10}, ) assert upcoming_result.isError is False, ( f"MCP upcoming events failed: {upcoming_result.content}" ) upcoming_response = json.loads(upcoming_result.content[0].text) # Response is now an UpcomingEventsResponse with an "events" field assert isinstance(upcoming_response, dict), "Expected response dict" upcoming_events = upcoming_response.get("events", []) assert isinstance(upcoming_events, list), "Expected upcoming events list" # 10. Delete event via MCP logger.info("Deleting event via MCP: %s", event_uid) delete_result = await nc_mcp_client.call_tool( "nc_calendar_delete_event", {"calendar_name": calendar_name, "event_uid": event_uid}, ) assert delete_result.isError is False, ( f"MCP event deletion failed: {delete_result.content}" ) # 11. Verify deletion via direct NextcloudClient try: await nc_client.calendar.get_event(calendar_name, event_uid) pytest.fail("Event should have been deleted but was still found") except Exception: # Expected - event should be deleted logger.info("Successfully verified event %s was deleted", event_uid) created_event = None # Mark as cleaned up except Exception as e: if "Calendar app may not be enabled" in str( e ) or "No calendars available" in str(e): pytest.skip("Calendar functionality not available for testing") raise finally: # Cleanup in case of test failure if created_event is not None: try: await nc_client.calendar.delete_event( created_event["calendar_name"], created_event["uid"] ) logger.info( "Cleaned up event %s after test failure", created_event["uid"] ) except Exception as e: logger.warning("Failed to cleanup event: %s", e)