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_deck_mcp_connectivity(nc_mcp_client: ClientSession): """Test deck MCP tools are available and functional.""" # List available tools tools = await nc_mcp_client.list_tools() tool_names = [tool.name for tool in tools.tools] # Verify expected deck tools are present expected_deck_tools = ["deck_create_board"] for expected_tool in expected_deck_tools: assert expected_tool in tool_names, ( f"Expected deck tool '{expected_tool}' not found in available tools" ) logger.info("Found expected deck tool: %s", expected_tool) # List available resource templates templates = await nc_mcp_client.list_resource_templates() template_uris = [template.uriTemplate for template in templates.resourceTemplates] # Verify expected deck resource templates expected_deck_templates = [ "nc://Deck/boards/{board_id}", ] for expected_template in expected_deck_templates: assert expected_template in template_uris, ( f"Expected deck template '{expected_template}' not found" ) logger.info("Found expected deck resource template: %s", expected_template) # List available resources resources = await nc_mcp_client.list_resources() resource_uris = [str(resource.uri) for resource in resources.resources] # Verify expected deck resources expected_deck_resources = [ "nc://Deck/boards", ] for expected_resource in expected_deck_resources: assert expected_resource in resource_uris, ( f"Expected deck resource '{expected_resource}' not found" ) logger.info("Found expected deck resource: %s", expected_resource) async def test_deck_board_crud_workflow_mcp( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test complete Deck board CRUD workflow via MCP tools with verification via NextcloudClient.""" unique_suffix = uuid.uuid4().hex[:8] board_title = f"MCP Test Board {unique_suffix}" board_color = "0000FF" # Blue # 1. Create board via MCP logger.info("Creating board via MCP: %s", board_title) create_result = await nc_mcp_client.call_tool( "deck_create_board", {"title": board_title, "color": board_color}, ) assert create_result.isError is False, ( f"MCP board creation failed: {create_result.content}" ) created_board_json = create_result.content[0].text created_board_response = json.loads(created_board_json) board_id = created_board_response["id"] logger.info("Board created via MCP with ID: %s", board_id) assert created_board_response["title"] == board_title assert created_board_response["color"] == board_color # 2. Verify creation via direct NextcloudClient direct_board = await nc_client.deck.get_board(board_id) assert direct_board.title == board_title, ( f"Title mismatch: {direct_board.title} != {board_title}" ) assert direct_board.color == board_color, "Color mismatch" logger.info("Board creation verified via direct client") # 3. Read board via MCP resource logger.info("Reading board via MCP resource: %s", board_id) read_result = await nc_mcp_client.read_resource(f"nc://Deck/boards/{board_id}") assert len(read_result.contents) == 1, "Expected exactly one content item" read_board_data = json.loads(read_result.contents[0].text) assert read_board_data["title"] == board_title assert read_board_data["color"] == board_color logger.info("Board read via MCP resource successfully") # 4. Verify board via direct read of resource logger.info("Verifying board via resource read: %s", board_id) # This was already done in step 3, so we'll just log confirmation logger.info("Board structure verified successfully") # 5. Read boards list via MCP resource logger.info("Reading boards list via MCP resource") boards_resource_result = await nc_mcp_client.read_resource("nc://Deck/boards") assert len(boards_resource_result.contents) == 1, ( "Expected exactly one content item" ) boards_resource_data = json.loads(boards_resource_result.contents[0].text) assert isinstance(boards_resource_data, list) # Resources return raw lists # Verify our board is in the resource list resource_board_ids = [board["id"] for board in boards_resource_data] assert board_id in resource_board_ids, "Created board not found in resource list" logger.info("Board found in boards resource list") # Clean up - delete board await nc_client.deck.delete_board(board_id) logger.info("Cleaned up board ID: %s", board_id) async def test_deck_board_operations_error_handling_mcp(nc_mcp_client: ClientSession): """Test MCP deck tools handle errors appropriately.""" non_existent_id = 999999999 # Test create board with invalid parameters via MCP tool logger.info("Testing board creation with invalid parameters via MCP") create_result = await nc_mcp_client.call_tool( "deck_create_board", {"title": "", "color": "FF0000"}, ) assert create_result.isError is True, "Expected error for invalid board creation" logger.info("Invalid board creation correctly failed via MCP tool") # Test read non-existent board via MCP resource logger.info("Testing read non-existent board via MCP resource: %s", non_existent_id) try: read_result = await nc_mcp_client.read_resource( f"nc://Deck/boards/{non_existent_id}" ) # If no error is thrown, check if the result indicates an error assert len(read_result.contents) == 0, ( "Expected empty content for non-existent board" ) except Exception as e: logger.info("Read non-existent board correctly failed via MCP resource: %s", e) async def test_deck_board_creation_validation_mcp(nc_mcp_client: ClientSession): """Test deck board creation validation via MCP tools.""" # Test creating board with empty title should fail logger.info("Testing board creation with empty title via MCP") create_result = await nc_mcp_client.call_tool( "deck_create_board", {"title": "", "color": "FF0000"}, ) assert create_result.isError is True, "Expected error for empty board title" logger.info("Empty title board creation correctly failed via MCP") async def test_deck_board_creation_success_mcp( nc_mcp_client: ClientSession, nc_client: NextcloudClient ): """Test deck board creation with valid parameters via MCP tools.""" # Test creating board with valid parameters logger.info("Testing board creation with valid parameters via MCP") create_result = await nc_mcp_client.call_tool( "deck_create_board", {"title": f"Valid Board {uuid.uuid4().hex[:8]}", "color": "00FF00"}, ) assert create_result.isError is False, "Valid board creation should succeed" created_board = json.loads(create_result.content[0].text) board_id = created_board["id"] logger.info("Valid board created successfully with ID: %s", board_id) # Clean up - delete board await nc_client.deck.delete_board(board_id) logger.info("Cleaned up board ID: %s", board_id) async def test_deck_workflow_integration_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """Test a complete deck workflow using MCP tools with temporary resources.""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] board_title = board_data["title"] # 1. Read board via MCP to verify the structure logger.info("Reading board via MCP resource: %s", board_id) read_result = await nc_mcp_client.read_resource(f"nc://Deck/boards/{board_id}") board_mcp_data = json.loads(read_result.contents[0].text) assert board_mcp_data["title"] == board_title logger.info("Board structure verified via MCP resource") # 2. List boards via MCP resource and verify our board is there logger.info("Listing boards via MCP resource") list_result = await nc_mcp_client.read_resource("nc://Deck/boards") boards_data = json.loads(list_result.contents[0].text) board_found = any(board["id"] == board_id for board in boards_data) assert board_found, "Board not found in boards list" logger.info("Board found in boards list") # 3. Verify board data matches via resource (already done in step 1) logger.info("Board data verification completed for board: %s", board_id) logger.info("Board structure and data verified successfully") # Card Comment Tests async def test_deck_card_comment_crud_workflow_mcp( nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_board_with_card: tuple, ): """Full CRUD lifecycle for card comments via MCP tools.""" _, _, card_data = temporary_board_with_card card_id = card_data["id"] # 1. Create a top-level comment via MCP create_result = await nc_mcp_client.call_tool( "deck_create_card_comment", {"card_id": card_id, "message": "Initial comment"}, ) assert create_result.isError is False, ( f"Comment creation failed: {create_result.content}" ) create_response = json.loads(create_result.content[0].text) assert create_response["success"] is True comment = create_response["comment"] comment_id = comment["id"] assert comment["objectId"] == card_id assert comment["message"] == "Initial comment" assert comment["replyTo"] is None logger.info("Created comment ID %s on card %s", comment_id, card_id) # 2. List comments via MCP — verify the new comment is present list_result = await nc_mcp_client.call_tool( "deck_get_card_comments", {"card_id": card_id} ) assert list_result.isError is False, f"List comments failed: {list_result.content}" listed = json.loads(list_result.content[0].text) assert listed["count"] >= 1 listed_ids = [c["id"] for c in listed["results"]] assert comment_id in listed_ids, "Created comment not in list" # 3. Cross-check via direct client direct_comments = await nc_client.deck.get_comments(card_id) direct_ids = [c.id for c in direct_comments] assert comment_id in direct_ids, "Created comment not visible via direct client" # 4. Update the comment via MCP update_result = await nc_mcp_client.call_tool( "deck_update_card_comment", { "card_id": card_id, "comment_id": comment_id, "message": "Edited comment", }, ) assert update_result.isError is False, ( f"Comment update failed: {update_result.content}" ) update_response = json.loads(update_result.content[0].text) updated = update_response["comment"] assert updated["id"] == comment_id assert updated["message"] == "Edited comment" # 5. Delete the comment via MCP delete_result = await nc_mcp_client.call_tool( "deck_delete_card_comment", {"card_id": card_id, "comment_id": comment_id}, ) assert delete_result.isError is False, ( f"Comment delete failed: {delete_result.content}" ) delete_response = json.loads(delete_result.content[0].text) assert delete_response["success"] is True assert delete_response["card_id"] == card_id assert delete_response["comment_id"] == comment_id # 6. Verify the comment is gone final_list_result = await nc_mcp_client.call_tool( "deck_get_card_comments", {"card_id": card_id} ) final_listed = json.loads(final_list_result.content[0].text) final_ids = [c["id"] for c in final_listed["results"]] assert comment_id not in final_ids, "Comment still present after delete" async def test_deck_card_comment_reply_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """Replying with parent_id populates replyTo on the new comment.""" _, _, card_data = temporary_board_with_card card_id = card_data["id"] # Create the parent comment parent_result = await nc_mcp_client.call_tool( "deck_create_card_comment", {"card_id": card_id, "message": "Parent message"}, ) assert parent_result.isError is False parent = json.loads(parent_result.content[0].text)["comment"] parent_id = parent["id"] # Create a reply reply_result = await nc_mcp_client.call_tool( "deck_create_card_comment", { "card_id": card_id, "message": "Reply message", "parent_id": parent_id, }, ) assert reply_result.isError is False, f"Reply failed: {reply_result.content}" reply = json.loads(reply_result.content[0].text)["comment"] assert reply["message"] == "Reply message" assert reply["replyTo"] is not None, "replyTo should be populated for replies" assert reply["replyTo"]["id"] == parent_id assert reply["replyTo"]["message"] == "Parent message" async def test_deck_card_comment_message_too_long_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """Creating a comment longer than 1000 chars is rejected client-side.""" _, _, card_data = temporary_board_with_card card_id = card_data["id"] too_long = "x" * 1001 result = await nc_mcp_client.call_tool( "deck_create_card_comment", {"card_id": card_id, "message": too_long}, ) assert result.isError is True, "Expected validation error for >1000 char message" # Compact retrieval (summary projection + board overview) async def test_deck_get_stacks_summary_default_omits_full_card_fields_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """deck_get_stacks defaults to compact summaries: the card row carries title/labels but not the heavy full-card fields (owner/type/etag).""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] result = await nc_mcp_client.call_tool("deck_get_stacks", {"board_id": board_id}) assert result.isError is False, f"deck_get_stacks failed: {result.content}" payload = json.loads(result.content[0].text) cards = [c for stack in payload["stacks"] for c in (stack.get("cards") or [])] card = next(c for c in cards if c["id"] == card_data["id"]) # Summary fields present... assert card["title"] == card_data["title"] assert "hasDescription" in card assert "labels" in card and isinstance(card["labels"], list) # ...heavy full-card fields absent. assert "owner" not in card assert "type" not in card async def test_deck_get_stacks_detail_full_keeps_card_fields_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """detail="full" restores the complete card objects (owner/type present).""" board_data, _, card_data = temporary_board_with_card board_id = board_data["id"] result = await nc_mcp_client.call_tool( "deck_get_stacks", {"board_id": board_id, "detail": "full"} ) assert result.isError is False, f"deck_get_stacks(full) failed: {result.content}" payload = json.loads(result.content[0].text) cards = [c for stack in payload["stacks"] for c in (stack.get("cards") or [])] card = next(c for c in cards if c["id"] == card_data["id"]) assert "owner" in card assert "type" in card async def test_deck_get_board_overview_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """deck_get_board_overview returns board title + stacks with compact rows.""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] result = await nc_mcp_client.call_tool( "deck_get_board_overview", {"board_id": board_id} ) assert result.isError is False, f"board overview failed: {result.content}" payload = json.loads(result.content[0].text) assert payload["board_id"] == board_id assert payload["title"] == board_data["title"] assert payload["total_cards"] >= 1 stack = next(s for s in payload["stacks"] if s["id"] == stack_data["id"]) assert stack["card_count"] == len(stack["cards"]) card_ids = [c["id"] for c in stack["cards"]] assert card_data["id"] in card_ids # Archived-card visibility in the active list tools (issue #842) async def _create_and_archive_card( nc_mcp_client: ClientSession, board_id: int, stack_id: int ) -> int: """Create a card in the given stack, archive it, and return its id. The card is cleaned up when the temporary stack is deleted by the fixture teardown (deleting a stack removes its cards).""" create_result = await nc_mcp_client.call_tool( "deck_create_card", { "board_id": board_id, "stack_id": stack_id, "title": f"Archived Card {uuid.uuid4().hex[:8]}", }, ) assert create_result.isError is False, f"create failed: {create_result.content}" archived_id = json.loads(create_result.content[0].text)["id"] archive_result = await nc_mcp_client.call_tool( "deck_archive_card", {"board_id": board_id, "stack_id": stack_id, "card_id": archived_id}, ) assert archive_result.isError is False, f"archive failed: {archive_result.content}" return archived_id async def test_deck_get_cards_status_includes_archived_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """deck_get_cards: archived cards appear under status="all"/"archived" and are excluded under the default "open" — the regression behind issue #842.""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] stack_id = stack_data["id"] open_id = card_data["id"] archived_id = await _create_and_archive_card(nc_mcp_client, board_id, stack_id) async def card_ids(status: str) -> list[int]: result = await nc_mcp_client.call_tool( "deck_get_cards", {"board_id": board_id, "stack_id": stack_id, "status": status}, ) assert result.isError is False, f"deck_get_cards({status}) failed" return [c["id"] for c in json.loads(result.content[0].text)["cards"]] open_ids = await card_ids("open") assert open_id in open_ids assert archived_id not in open_ids, "archived card must not show under 'open'" all_ids = await card_ids("all") assert open_id in all_ids and archived_id in all_ids, ( "status='all' must include both open and archived cards" ) archived_only = await card_ids("archived") assert archived_only == [archived_id], ( f"status='archived' should return only the archived card, got {archived_only}" ) async def test_deck_get_stack_status_includes_archived_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """deck_get_stack (single stack) honours archived cards for status "all"/"archived" too, mirroring deck_get_cards (issue #842).""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] stack_id = stack_data["id"] archived_id = await _create_and_archive_card(nc_mcp_client, board_id, stack_id) async def stack_card_ids(status: str) -> list[int]: result = await nc_mcp_client.call_tool( "deck_get_stack", {"board_id": board_id, "stack_id": stack_id, "status": status}, ) assert result.isError is False, f"deck_get_stack({status}) failed" payload = json.loads(result.content[0].text) return [c["id"] for c in (payload.get("cards") or [])] assert archived_id not in await stack_card_ids("open") assert archived_id in await stack_card_ids("all") assert await stack_card_ids("archived") == [archived_id] async def test_deck_get_stacks_and_overview_include_archived_mcp( nc_mcp_client: ClientSession, temporary_board_with_card: tuple ): """deck_get_stacks and deck_get_board_overview surface archived cards when status="all" (issue #842), keyed onto the correct stack.""" board_data, stack_data, card_data = temporary_board_with_card board_id = board_data["id"] stack_id = stack_data["id"] archived_id = await _create_and_archive_card(nc_mcp_client, board_id, stack_id) async def stacks_card_ids(status: str) -> list[int]: result = await nc_mcp_client.call_tool( "deck_get_stacks", {"board_id": board_id, "status": status} ) assert result.isError is False, f"deck_get_stacks({status}) failed" payload = json.loads(result.content[0].text) stack = next(s for s in payload["stacks"] if s["id"] == stack_id) return [c["id"] for c in (stack.get("cards") or [])] async def overview_card_ids(status: str) -> list[int]: result = await nc_mcp_client.call_tool( "deck_get_board_overview", {"board_id": board_id, "status": status} ) assert result.isError is False, f"board overview({status}) failed" payload = json.loads(result.content[0].text) stack = next(s for s in payload["stacks"] if s["id"] == stack_id) assert stack["card_count"] == len(stack["cards"]) return [c["id"] for c in stack["cards"]] # status="all": both the open and archived card present. stacks_all = await stacks_card_ids("all") assert card_data["id"] in stacks_all assert archived_id in stacks_all, "archived card missing from deck_get_stacks(all)" overview_all = await overview_card_ids("all") assert archived_id in overview_all, "archived card missing from board overview(all)" # status="archived": only the archived card, open card excluded. assert await stacks_card_ids("archived") == [archived_id] assert await overview_card_ids("archived") == [archived_id]