feat(deck): Initialize Deck app client/server

This commit is contained in:
Chris Coutinho
2025-09-11 00:10:25 +02:00
parent b1eb4d2497
commit 167053578d
12 changed files with 1620 additions and 6 deletions
+327
View File
@@ -0,0 +1,327 @@
import logging
import uuid
import pytest
from httpx import HTTPStatusError
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.models.deck import DeckStack, DeckCard, DeckLabel
logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration
# Board CRUD Tests
async def test_deck_board_crud_workflow(
nc_client: NextcloudClient, temporary_board: dict
):
"""
Test complete board CRUD workflow using the temporary_board fixture.
"""
board_data = temporary_board
board_id = board_data["id"]
original_title = board_data["title"]
original_color = board_data["color"]
logger.info(f"Testing CRUD operations on board ID: {board_id}")
# Read the board
read_board = await nc_client.deck.get_board(board_id)
assert read_board.id == board_id
assert read_board.title == original_title
assert read_board.color == original_color
logger.info(f"Successfully read board ID: {board_id}")
# Update the board
updated_title = f"Updated {original_title}"
updated_color = "00FF00" # Green color
await nc_client.deck.update_board(
board_id, title=updated_title, color=updated_color
)
# Verify the update
updated_board = await nc_client.deck.get_board(board_id)
assert updated_board.title == updated_title
assert updated_board.color == updated_color
logger.info(f"Successfully updated board ID: {board_id}")
async def test_deck_list_boards(nc_client: NextcloudClient):
"""
Test listing all boards with different options.
"""
# Test basic listing
boards = await nc_client.deck.get_boards()
assert isinstance(boards, list)
logger.info(f"Found {len(boards)} boards")
# Test with details
detailed_boards = await nc_client.deck.get_boards(details=True)
assert isinstance(detailed_boards, list)
logger.info(f"Found {len(detailed_boards)} boards with details")
async def test_deck_board_operations_nonexistent(nc_client: NextcloudClient):
"""
Test operations on non-existent board return appropriate errors.
"""
non_existent_id = 999999999
# Test get non-existent board
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.deck.get_board(non_existent_id)
assert excinfo.value.response.status_code in [
404,
403,
] # 403 might be returned for access denied
logger.info(
f"Get non-existent board correctly failed with {excinfo.value.response.status_code}"
)
# Test update non-existent board
with pytest.raises(HTTPStatusError) as excinfo:
await nc_client.deck.update_board(non_existent_id, title="Should Fail")
assert excinfo.value.response.status_code in [
404,
403,
400,
] # 400 for bad request on invalid board ID
logger.info(
f"Update non-existent board correctly failed with {excinfo.value.response.status_code}"
)
# Stack CRUD Tests
async def test_deck_stack_crud_workflow(
nc_client: NextcloudClient, temporary_board: dict
):
"""
Test complete stack CRUD workflow.
"""
board_id = temporary_board["id"]
stack_title = f"Test Stack {uuid.uuid4().hex[:8]}"
stack_order = 1
stack = None
try:
# Create stack
stack = await nc_client.deck.create_stack(board_id, stack_title, stack_order)
assert isinstance(stack, DeckStack)
assert stack.title == stack_title
assert stack.order == stack_order
stack_id = stack.id
logger.info(f"Created stack ID: {stack_id}")
# Read stack
read_stack = await nc_client.deck.get_stack(board_id, stack_id)
assert read_stack.id == stack_id
assert read_stack.title == stack_title
logger.info(f"Successfully read stack ID: {stack_id}")
# Update stack
updated_title = f"Updated {stack_title}"
updated_order = 2
await nc_client.deck.update_stack(
board_id, stack_id, title=updated_title, order=updated_order
)
# Verify update
updated_stack = await nc_client.deck.get_stack(board_id, stack_id)
assert updated_stack.title == updated_title
assert updated_stack.order == updated_order
logger.info(f"Successfully updated stack ID: {stack_id}")
# List stacks
stacks = await nc_client.deck.get_stacks(board_id)
assert isinstance(stacks, list)
assert any(s.id == stack_id for s in stacks)
logger.info(f"Found stack ID: {stack_id} in board stacks list")
finally:
# Clean up - delete stack
if stack and hasattr(stack, "id"):
try:
await nc_client.deck.delete_stack(board_id, stack.id)
logger.info(f"Cleaned up stack ID: {stack.id}")
except Exception as e:
logger.warning(f"Failed to clean up stack ID: {stack.id}: {e}")
# Card CRUD Tests
async def test_deck_card_crud_workflow(
nc_client: NextcloudClient, temporary_board_with_stack: tuple
):
"""
Test complete card CRUD workflow.
"""
board_data, stack_data = temporary_board_with_stack
board_id = board_data["id"]
stack_id = stack_data["id"]
card_title = f"Test Card {uuid.uuid4().hex[:8]}"
card_description = f"Test description for card {uuid.uuid4().hex[:8]}"
card = None
try:
# Create card
card = await nc_client.deck.create_card(
board_id, stack_id, card_title, description=card_description
)
assert isinstance(card, DeckCard)
assert card.title == card_title
assert card.description == card_description
card_id = card.id
logger.info(f"Created card ID: {card_id}")
# Read card
read_card = await nc_client.deck.get_card(board_id, stack_id, card_id)
assert read_card.id == card_id
assert read_card.title == card_title
logger.info(f"Successfully read card ID: {card_id}")
# Update card
updated_title = f"Updated {card_title}"
updated_description = f"Updated description for {card_title}"
await nc_client.deck.update_card(
board_id,
stack_id,
card_id,
title=updated_title,
description=updated_description,
)
# Verify update
updated_card = await nc_client.deck.get_card(board_id, stack_id, card_id)
assert updated_card.title == updated_title
assert updated_card.description == updated_description
logger.info(f"Successfully updated card ID: {card_id}")
# Archive and unarchive card
await nc_client.deck.archive_card(board_id, stack_id, card_id)
logger.info(f"Archived card ID: {card_id}")
await nc_client.deck.unarchive_card(board_id, stack_id, card_id)
logger.info(f"Unarchived card ID: {card_id}")
finally:
# Clean up - delete card
if card and hasattr(card, "id"):
try:
await nc_client.deck.delete_card(board_id, stack_id, card.id)
logger.info(f"Cleaned up card ID: {card.id}")
except Exception as e:
logger.warning(f"Failed to clean up card ID: {card.id}: {e}")
# Label CRUD Tests
async def test_deck_label_crud_workflow(
nc_client: NextcloudClient, temporary_board: dict
):
"""
Test complete label CRUD workflow.
"""
board_id = temporary_board["id"]
label_title = f"Test Label {uuid.uuid4().hex[:8]}"
label_color = "FF0000" # Red
label = None
try:
# Create label
label = await nc_client.deck.create_label(board_id, label_title, label_color)
assert isinstance(label, DeckLabel)
assert label.title == label_title
assert label.color == label_color
label_id = label.id
logger.info(f"Created label ID: {label_id}")
# Read label
read_label = await nc_client.deck.get_label(board_id, label_id)
assert read_label.id == label_id
assert read_label.title == label_title
logger.info(f"Successfully read label ID: {label_id}")
# Update label
updated_title = f"Updated {label_title}"
updated_color = "00FF00" # Green
await nc_client.deck.update_label(
board_id, label_id, title=updated_title, color=updated_color
)
# Verify update
updated_label = await nc_client.deck.get_label(board_id, label_id)
assert updated_label.title == updated_title
assert updated_label.color == updated_color
logger.info(f"Successfully updated label ID: {label_id}")
finally:
# Clean up - delete label
if label and hasattr(label, "id"):
try:
await nc_client.deck.delete_label(board_id, label.id)
logger.info(f"Cleaned up label ID: {label.id}")
except Exception as e:
logger.warning(f"Failed to clean up label ID: {label.id}: {e}")
# Configuration and Comments Tests
async def test_deck_config_operations(nc_client: NextcloudClient):
"""
Test deck configuration operations.
"""
# Get config
config = await nc_client.deck.get_config()
assert config is not None
logger.info(f"Retrieved deck config: {config}")
async def test_deck_comments_workflow(
nc_client: NextcloudClient, temporary_board_with_card: tuple
):
"""
Test comment operations on a card.
"""
board_data, stack_data, card_data = temporary_board_with_card
card_id = card_data["id"]
comment_message = f"Test comment {uuid.uuid4().hex[:8]}"
comment = None
try:
# Create comment
comment = await nc_client.deck.create_comment(card_id, comment_message)
assert comment.message == comment_message
comment_id = comment.id
logger.info(f"Created comment ID: {comment_id}")
# List comments
comments = await nc_client.deck.get_comments(card_id)
assert isinstance(comments, list)
assert any(c.id == comment_id for c in comments)
logger.info(f"Found comment ID: {comment_id} in card comments")
# Update comment
updated_message = f"Updated {comment_message}"
updated_comment = await nc_client.deck.update_comment(
card_id, comment_id, updated_message
)
assert updated_comment.message == updated_message
logger.info(f"Successfully updated comment ID: {comment_id}")
finally:
# Clean up - delete comment
if comment and hasattr(comment, "id"):
try:
await nc_client.deck.delete_comment(card_id, comment.id)
logger.info(f"Cleaned up comment ID: {comment.id}")
except Exception as e:
logger.warning(f"Failed to clean up comment ID: {comment.id}: {e}")
+268
View File
@@ -0,0 +1,268 @@
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_list_boards", "deck_create_board", "deck_get_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(f"Found expected deck tool: {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(f"Found expected deck resource template: {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(f"Found expected deck resource: {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(f"Creating board via MCP: {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(f"Board created via MCP with ID: {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(f"Reading board via MCP resource: {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. Get board via MCP tool
logger.info(f"Getting board via MCP tool: {board_id}")
get_result = await nc_mcp_client.call_tool(
"deck_get_board",
{"board_id": board_id},
)
assert get_result.isError is False, f"MCP board get failed: {get_result.content}"
get_board_response = json.loads(get_result.content[0].text)
get_board_data = get_board_response["board"]
assert get_board_data["title"] == board_title
assert get_board_data["color"] == board_color
logger.info("Board retrieved via MCP tool successfully")
# 5. List boards via MCP tool
logger.info("Listing boards via MCP tool")
list_result = await nc_mcp_client.call_tool("deck_list_boards", {})
assert list_result.isError is False, f"MCP board list failed: {list_result.content}"
boards_response = json.loads(list_result.content[0].text)
boards_data = boards_response["boards"]
assert isinstance(boards_data, list)
# Verify our board is in the list
board_ids = [board["id"] for board in boards_data]
assert board_id in board_ids, "Created board not found in list"
logger.info(f"Board {board_id} found in boards list")
# 6. List boards with details via MCP tool
logger.info("Listing boards with details via MCP tool")
list_details_result = await nc_mcp_client.call_tool(
"deck_list_boards", {"details": True}
)
assert list_details_result.isError is False, (
f"MCP board list with details failed: {list_details_result.content}"
)
detailed_boards_response = json.loads(list_details_result.content[0].text)
detailed_boards_data = detailed_boards_response["boards"]
assert isinstance(detailed_boards_data, list)
logger.info("Boards listed with details successfully")
# 7. 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(f"Cleaned up board ID: {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 get non-existent board via MCP tool
logger.info(f"Testing get non-existent board via MCP: {non_existent_id}")
get_result = await nc_mcp_client.call_tool(
"deck_get_board",
{"board_id": non_existent_id},
)
assert get_result.isError is True, "Expected error for non-existent board"
logger.info("Get non-existent board correctly failed via MCP tool")
# Test read non-existent board via MCP resource
logger.info(f"Testing read non-existent board via MCP resource: {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(f"Read non-existent board correctly failed via MCP resource: {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(f"Valid board created successfully with ID: {board_id}")
# Clean up - delete board
await nc_client.deck.delete_board(board_id)
logger.info(f"Cleaned up board ID: {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(f"Reading board via MCP resource: {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 and verify our board is there
logger.info("Listing boards with details via MCP tool")
list_result = await nc_mcp_client.call_tool("deck_list_boards", {"details": True})
boards_response = json.loads(list_result.content[0].text)
boards_data = boards_response["boards"]
board_found = any(board["id"] == board_id for board in boards_data)
assert board_found, "Board not found in detailed list"
logger.info("Board found in detailed boards list")
# 3. Get board via MCP tool and verify it matches our data
logger.info(f"Getting board via MCP tool: {board_id}")
get_result = await nc_mcp_client.call_tool(
"deck_get_board",
{"board_id": board_id},
)
assert get_result.isError is False, "MCP board get failed"
get_board_response = json.loads(get_result.content[0].text)
get_board_data = get_board_response["board"]
assert get_board_data["title"] == board_title
logger.info("Board data verified via MCP tool")
+4 -1
View File
@@ -51,6 +51,9 @@ async def test_mcp_connectivity(nc_mcp_client: ClientSession):
"nc_calendar_find_availability",
"nc_calendar_bulk_operations",
"nc_calendar_manage_calendar",
"deck_list_boards",
"deck_create_board",
"deck_get_board",
]
for expected_tool in expected_tools:
@@ -83,7 +86,7 @@ async def test_mcp_connectivity(nc_mcp_client: ClientSession):
resource_uris.append(str(resource.uri)) # Convert to string for comparison
# Verify expected resources
expected_resources = ["nc://capabilities", "notes://settings"]
expected_resources = ["nc://capabilities", "notes://settings", "nc://Deck/boards"]
for expected_resource in expected_resources:
assert expected_resource in resource_uris, (