Sweep all 1676 G004 violations across 112 files, converting
`logger.<level>(f"…{x}…")` to `logger.<level>("…%s…", x)`.
Why: ruff rule G004 was added to pyproject.toml to enforce lazy
%-style logging — defers formatting until the log level is enabled
and lets structured log tooling match the unformatted template.
Conversion preserves rendered output byte-for-byte:
- `{x}` → `%s` + `x`
- `{x!r}` / `{x!s}` / `{x!a}` → `%r` / `%s` / `%a`
- Format specs (`{x:.2f}`, `{x:>10}`) → `%s` + `format(x, 'spec')`
(printf-style specs aren't 1:1 with Python format specs, so we
delegate to `format()` to keep identical output)
- Literal `%` → `%%`
- Concatenated f-strings (`f"a {x} " "b"`) flattened
- Trailing kwargs (`exc_info=True`) preserved
Verified:
- `uv run ruff check --select G004` → 0 violations
- `uv run ty check -- nextcloud_mcp_server` → passes
- `uv run pytest tests/unit/` → 1010 passed
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
570 lines
21 KiB
Python
570 lines
21 KiB
Python
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
|
|
|
|
|
|
# Stack MCP Tools Tests
|
|
async def test_deck_stack_mcp_tools(
|
|
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_board: dict
|
|
):
|
|
"""Test complete deck stack operations via MCP tools."""
|
|
board_id = temporary_board["id"]
|
|
stack_title = f"MCP Test Stack {uuid.uuid4().hex[:8]}"
|
|
stack_order = 1
|
|
|
|
# 1. Create stack via MCP tool
|
|
logger.info("Creating stack via MCP: %s", stack_title)
|
|
create_result = await nc_mcp_client.call_tool(
|
|
"deck_create_stack",
|
|
{"board_id": board_id, "title": stack_title, "order": stack_order},
|
|
)
|
|
|
|
assert create_result.isError is False, (
|
|
f"MCP stack creation failed: {create_result.content}"
|
|
)
|
|
created_stack_response = json.loads(create_result.content[0].text)
|
|
stack_id = created_stack_response["id"]
|
|
assert created_stack_response["title"] == stack_title
|
|
assert created_stack_response["order"] == stack_order
|
|
logger.info("Stack created via MCP with ID: %s", stack_id)
|
|
|
|
try:
|
|
# 2. Get stack via MCP resource
|
|
logger.info("Getting stack via MCP resource: %s", stack_id)
|
|
get_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks/{stack_id}"
|
|
)
|
|
|
|
assert len(get_result.contents) == 1, "Expected exactly one content item"
|
|
get_stack_response = json.loads(get_result.contents[0].text)
|
|
assert get_stack_response["title"] == stack_title
|
|
logger.info("Stack retrieved via MCP resource successfully")
|
|
|
|
# 3. Update stack via MCP tool
|
|
updated_title = f"Updated {stack_title}"
|
|
updated_order = 2
|
|
logger.info("Updating stack via MCP tool: %s", stack_id)
|
|
update_result = await nc_mcp_client.call_tool(
|
|
"deck_update_stack",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"title": updated_title,
|
|
"order": updated_order,
|
|
},
|
|
)
|
|
|
|
assert update_result.isError is False, (
|
|
f"MCP stack update failed: {update_result.content}"
|
|
)
|
|
logger.info("Stack updated via MCP tool successfully")
|
|
|
|
# 4. Verify update via direct client
|
|
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("Stack update verified via direct client")
|
|
|
|
# 5. List stacks via MCP resource
|
|
logger.info("Listing stacks via MCP resource")
|
|
list_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks"
|
|
)
|
|
|
|
assert len(list_result.contents) == 1, "Expected exactly one content item"
|
|
stacks_data = json.loads(list_result.contents[0].text)
|
|
assert isinstance(stacks_data, list)
|
|
|
|
# Verify our stack is in the list
|
|
stack_ids = [stack["id"] for stack in stacks_data]
|
|
assert stack_id in stack_ids, "Updated stack not found in list"
|
|
logger.info("Stack %s found in stacks list", stack_id)
|
|
|
|
# 6. Read stack via MCP resource
|
|
logger.info("Reading stack via MCP resource: %s", stack_id)
|
|
read_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks/{stack_id}"
|
|
)
|
|
read_stack_data = json.loads(read_result.contents[0].text)
|
|
assert read_stack_data["title"] == updated_title
|
|
logger.info("Stack read via MCP resource successfully")
|
|
|
|
finally:
|
|
# Clean up
|
|
await nc_client.deck.delete_stack(board_id, stack_id)
|
|
logger.info("Cleaned up stack ID: %s", stack_id)
|
|
|
|
|
|
# Card MCP Tools Tests
|
|
async def test_deck_card_mcp_tools(
|
|
nc_mcp_client: ClientSession,
|
|
nc_client: NextcloudClient,
|
|
temporary_board_with_stack: tuple,
|
|
):
|
|
"""Test complete deck card operations via MCP tools."""
|
|
board_data, stack_data = temporary_board_with_stack
|
|
board_id = board_data["id"]
|
|
stack_id = stack_data["id"]
|
|
card_title = f"MCP Test Card {uuid.uuid4().hex[:8]}"
|
|
card_description = f"Test description for {card_title}"
|
|
|
|
# 1. Create card via MCP tool
|
|
logger.info("Creating card via MCP: %s", card_title)
|
|
create_result = await nc_mcp_client.call_tool(
|
|
"deck_create_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"title": card_title,
|
|
"description": card_description,
|
|
"type": "plain",
|
|
"order": 1,
|
|
},
|
|
)
|
|
|
|
assert create_result.isError is False, (
|
|
f"MCP card creation failed: {create_result.content}"
|
|
)
|
|
created_card_response = json.loads(create_result.content[0].text)
|
|
card_id = created_card_response["id"]
|
|
assert created_card_response["title"] == card_title
|
|
assert created_card_response["description"] == card_description
|
|
logger.info("Card created via MCP with ID: %s", card_id)
|
|
|
|
try:
|
|
# 2. Get card via MCP resource
|
|
logger.info("Getting card via MCP resource: %s", card_id)
|
|
get_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks/{stack_id}/cards/{card_id}"
|
|
)
|
|
|
|
assert len(get_result.contents) == 1, "Expected exactly one content item"
|
|
get_card_response = json.loads(get_result.contents[0].text)
|
|
assert get_card_response["title"] == card_title
|
|
logger.info("Card retrieved via MCP resource successfully")
|
|
|
|
# 3. Update card via MCP tool
|
|
updated_title = f"Updated {card_title}"
|
|
updated_description = f"Updated description for {card_title}"
|
|
logger.info("Updating card via MCP tool: %s", card_id)
|
|
update_result = await nc_mcp_client.call_tool(
|
|
"deck_update_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"title": updated_title,
|
|
"description": updated_description,
|
|
},
|
|
)
|
|
|
|
assert update_result.isError is False, (
|
|
f"MCP card update failed: {update_result.content}"
|
|
)
|
|
logger.info("Card updated via MCP tool successfully")
|
|
|
|
# 4. Verify update via direct client
|
|
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("Card update verified via direct client")
|
|
|
|
# 5. Archive/unarchive card via MCP tools
|
|
logger.info("Archiving card via MCP tool: %s", card_id)
|
|
archive_result = await nc_mcp_client.call_tool(
|
|
"deck_archive_card",
|
|
{"board_id": board_id, "stack_id": stack_id, "card_id": card_id},
|
|
)
|
|
|
|
assert archive_result.isError is False, (
|
|
f"MCP card archive failed: {archive_result.content}"
|
|
)
|
|
logger.info("Card archived via MCP tool successfully")
|
|
|
|
logger.info("Unarchiving card via MCP tool: %s", card_id)
|
|
unarchive_result = await nc_mcp_client.call_tool(
|
|
"deck_unarchive_card",
|
|
{"board_id": board_id, "stack_id": stack_id, "card_id": card_id},
|
|
)
|
|
|
|
assert unarchive_result.isError is False, (
|
|
f"MCP card unarchive failed: {unarchive_result.content}"
|
|
)
|
|
logger.info("Card unarchived via MCP tool successfully")
|
|
|
|
# 6. Move card to different position via MCP tool
|
|
logger.info("Reordering card via MCP tool: %s", card_id)
|
|
reorder_result = await nc_mcp_client.call_tool(
|
|
"deck_reorder_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"order": 10,
|
|
"target_stack_id": stack_id,
|
|
},
|
|
)
|
|
|
|
assert reorder_result.isError is False, (
|
|
f"MCP card reorder failed: {reorder_result.content}"
|
|
)
|
|
logger.info("Card reordered via MCP tool successfully")
|
|
|
|
# 7. Read card via MCP resource
|
|
logger.info("Reading card via MCP resource: %s", card_id)
|
|
read_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks/{stack_id}/cards/{card_id}"
|
|
)
|
|
read_card_data = json.loads(read_result.contents[0].text)
|
|
assert read_card_data["title"] == updated_title
|
|
logger.info("Card read via MCP resource successfully")
|
|
|
|
finally:
|
|
# Clean up
|
|
await nc_client.deck.delete_card(board_id, stack_id, card_id)
|
|
logger.info("Cleaned up card ID: %s", card_id)
|
|
|
|
|
|
# Label MCP Tools Tests
|
|
async def test_deck_label_mcp_tools(
|
|
nc_mcp_client: ClientSession, nc_client: NextcloudClient, temporary_board: dict
|
|
):
|
|
"""Test complete deck label operations via MCP tools."""
|
|
board_id = temporary_board["id"]
|
|
label_title = f"MCP Test Label {uuid.uuid4().hex[:8]}"
|
|
label_color = "FF0000" # Red
|
|
|
|
# 1. Create label via MCP tool
|
|
logger.info("Creating label via MCP: %s", label_title)
|
|
create_result = await nc_mcp_client.call_tool(
|
|
"deck_create_label",
|
|
{"board_id": board_id, "title": label_title, "color": label_color},
|
|
)
|
|
|
|
assert create_result.isError is False, (
|
|
f"MCP label creation failed: {create_result.content}"
|
|
)
|
|
created_label_response = json.loads(create_result.content[0].text)
|
|
label_id = created_label_response["id"]
|
|
assert created_label_response["title"] == label_title
|
|
assert created_label_response["color"] == label_color
|
|
logger.info("Label created via MCP with ID: %s", label_id)
|
|
|
|
try:
|
|
# 2. Get label via MCP resource
|
|
logger.info("Getting label via MCP resource: %s", label_id)
|
|
get_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/labels/{label_id}"
|
|
)
|
|
|
|
assert len(get_result.contents) == 1, "Expected exactly one content item"
|
|
get_label_response = json.loads(get_result.contents[0].text)
|
|
assert get_label_response["title"] == label_title
|
|
logger.info("Label retrieved via MCP resource successfully")
|
|
|
|
# 3. Update label via MCP tool
|
|
updated_title = f"Updated {label_title}"
|
|
updated_color = "00FF00" # Green
|
|
logger.info("Updating label via MCP tool: %s", label_id)
|
|
update_result = await nc_mcp_client.call_tool(
|
|
"deck_update_label",
|
|
{
|
|
"board_id": board_id,
|
|
"label_id": label_id,
|
|
"title": updated_title,
|
|
"color": updated_color,
|
|
},
|
|
)
|
|
|
|
assert update_result.isError is False, (
|
|
f"MCP label update failed: {update_result.content}"
|
|
)
|
|
logger.info("Label updated via MCP tool successfully")
|
|
|
|
# 4. Verify update via direct client
|
|
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("Label update verified via direct client")
|
|
|
|
# 5. Read label via MCP resource
|
|
logger.info("Reading label via MCP resource: %s", label_id)
|
|
read_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/labels/{label_id}"
|
|
)
|
|
read_label_data = json.loads(read_result.contents[0].text)
|
|
assert read_label_data["title"] == updated_title
|
|
logger.info("Label read via MCP resource successfully")
|
|
|
|
finally:
|
|
# Clean up
|
|
await nc_client.deck.delete_label(board_id, label_id)
|
|
logger.info("Cleaned up label ID: %s", label_id)
|
|
|
|
|
|
# Label-Card Assignment Tests
|
|
async def test_deck_card_label_assignment_mcp_tools(
|
|
nc_mcp_client: ClientSession,
|
|
nc_client: NextcloudClient,
|
|
temporary_board_with_card: tuple,
|
|
):
|
|
"""Test card-label assignment operations via MCP tools."""
|
|
board_data, stack_data, card_data = temporary_board_with_card
|
|
board_id = board_data["id"]
|
|
stack_id = stack_data["id"]
|
|
card_id = card_data["id"]
|
|
|
|
# Create a label for assignment
|
|
label = await nc_client.deck.create_label(
|
|
board_id, "Assignment Test Label", "0000FF"
|
|
)
|
|
label_id = label.id
|
|
|
|
try:
|
|
# 1. Assign label to card via MCP tool
|
|
logger.info("Assigning label %s to card %s via MCP", label_id, card_id)
|
|
assign_result = await nc_mcp_client.call_tool(
|
|
"deck_assign_label_to_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"label_id": label_id,
|
|
},
|
|
)
|
|
|
|
assert assign_result.isError is False, (
|
|
f"MCP label assignment failed: {assign_result.content}"
|
|
)
|
|
logger.info("Label assigned to card via MCP tool successfully")
|
|
|
|
# 2. Verify assignment via direct client
|
|
card = await nc_client.deck.get_card(board_id, stack_id, card_id)
|
|
if card.labels:
|
|
label_ids = [label.id for label in card.labels]
|
|
assert label_id in label_ids, "Label not found in card labels"
|
|
logger.info("Label assignment verified via direct client")
|
|
|
|
# 3. Remove label from card via MCP tool
|
|
logger.info("Removing label %s from card %s via MCP", label_id, card_id)
|
|
remove_result = await nc_mcp_client.call_tool(
|
|
"deck_remove_label_from_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"label_id": label_id,
|
|
},
|
|
)
|
|
|
|
assert remove_result.isError is False, (
|
|
f"MCP label removal failed: {remove_result.content}"
|
|
)
|
|
logger.info("Label removed from card via MCP tool successfully")
|
|
|
|
# 4. Verify removal via direct client
|
|
card = await nc_client.deck.get_card(board_id, stack_id, card_id)
|
|
if card.labels:
|
|
label_ids = [label.id for label in card.labels]
|
|
assert label_id not in label_ids, (
|
|
"Label still found in card labels after removal"
|
|
)
|
|
logger.info("Label removal verified via direct client")
|
|
|
|
finally:
|
|
# Clean up
|
|
await nc_client.deck.delete_label(board_id, label_id)
|
|
logger.info("Cleaned up label ID: %s", label_id)
|
|
|
|
|
|
# User Assignment Tests
|
|
async def test_deck_card_user_assignment_mcp_tools(
|
|
nc_mcp_client: ClientSession,
|
|
nc_client: NextcloudClient,
|
|
temporary_board_with_card: tuple,
|
|
):
|
|
"""Test card-user assignment operations via MCP tools."""
|
|
board_data, stack_data, card_data = temporary_board_with_card
|
|
board_id = board_data["id"]
|
|
stack_id = stack_data["id"]
|
|
card_id = card_data["id"]
|
|
|
|
# Use the current user ID (admin in most test environments)
|
|
user_id = "admin"
|
|
|
|
# 1. Assign user to card via MCP tool
|
|
logger.info("Assigning user %s to card %s via MCP", user_id, card_id)
|
|
assign_result = await nc_mcp_client.call_tool(
|
|
"deck_assign_user_to_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"user_id": user_id,
|
|
},
|
|
)
|
|
|
|
assert assign_result.isError is False, (
|
|
f"MCP user assignment failed: {assign_result.content}"
|
|
)
|
|
logger.info("User assigned to card via MCP tool successfully")
|
|
|
|
# 2. Verify assignment via direct client
|
|
card = await nc_client.deck.get_card(board_id, stack_id, card_id)
|
|
if card.assignedUsers:
|
|
user_ids = []
|
|
for user in card.assignedUsers:
|
|
if hasattr(user, "participant"):
|
|
# It's a DeckAssignedUser with participant
|
|
user_ids.append(user.participant.uid)
|
|
elif hasattr(user, "uid"):
|
|
# It's a direct DeckUser
|
|
user_ids.append(user.uid)
|
|
assert user_id in user_ids, "User not found in card assigned users"
|
|
logger.info("User assignment verified via direct client")
|
|
|
|
# 3. Unassign user from card via MCP tool
|
|
logger.info("Unassigning user %s from card %s via MCP", user_id, card_id)
|
|
unassign_result = await nc_mcp_client.call_tool(
|
|
"deck_unassign_user_from_card",
|
|
{
|
|
"board_id": board_id,
|
|
"stack_id": stack_id,
|
|
"card_id": card_id,
|
|
"user_id": user_id,
|
|
},
|
|
)
|
|
|
|
assert unassign_result.isError is False, (
|
|
f"MCP user unassignment failed: {unassign_result.content}"
|
|
)
|
|
logger.info("User unassigned from card via MCP tool successfully")
|
|
|
|
# 4. Verify unassignment via direct client
|
|
card = await nc_client.deck.get_card(board_id, stack_id, card_id)
|
|
if card.assignedUsers:
|
|
user_ids = []
|
|
for user in card.assignedUsers:
|
|
if hasattr(user, "participant"):
|
|
# It's a DeckAssignedUser with participant
|
|
user_ids.append(user.participant.uid)
|
|
elif hasattr(user, "uid"):
|
|
# It's a direct DeckUser
|
|
user_ids.append(user.uid)
|
|
assert user_id not in user_ids, (
|
|
"User still found in card assigned users after removal"
|
|
)
|
|
logger.info("User unassignment verified via direct client")
|
|
|
|
|
|
# Error handling tests
|
|
async def test_deck_mcp_tools_error_handling(nc_mcp_client: ClientSession):
|
|
"""Test error handling for deck MCP tools with invalid parameters."""
|
|
non_existent_id = 999999999
|
|
|
|
# Test stack operations with non-existent board
|
|
stack_result = await nc_mcp_client.call_tool(
|
|
"deck_create_stack",
|
|
{"board_id": non_existent_id, "title": "Should Fail", "order": 1},
|
|
)
|
|
assert stack_result.isError is True, (
|
|
"Expected error for stack creation on non-existent board"
|
|
)
|
|
|
|
# Test card operations with non-existent IDs
|
|
card_result = await nc_mcp_client.call_tool(
|
|
"deck_create_card",
|
|
{
|
|
"board_id": non_existent_id,
|
|
"stack_id": non_existent_id,
|
|
"title": "Should Fail",
|
|
"type": "plain",
|
|
},
|
|
)
|
|
assert card_result.isError is True, (
|
|
"Expected error for card creation with non-existent IDs"
|
|
)
|
|
|
|
# Test label operations with non-existent board
|
|
label_result = await nc_mcp_client.call_tool(
|
|
"deck_create_label",
|
|
{"board_id": non_existent_id, "title": "Should Fail", "color": "FF0000"},
|
|
)
|
|
assert label_result.isError is True, (
|
|
"Expected error for label creation on non-existent board"
|
|
)
|
|
|
|
logger.info("Error handling tests passed for deck MCP tools")
|
|
|
|
|
|
# Resource template tests
|
|
async def test_deck_mcp_resource_templates(nc_mcp_client: ClientSession):
|
|
"""Test deck MCP resource templates are properly registered."""
|
|
templates = await nc_mcp_client.list_resource_templates()
|
|
template_uris = [template.uriTemplate for template in templates.resourceTemplates]
|
|
|
|
expected_templates = [
|
|
"nc://Deck/boards/{board_id}/stacks/{stack_id}",
|
|
"nc://Deck/boards/{board_id}/stacks/{stack_id}/cards/{card_id}",
|
|
"nc://Deck/boards/{board_id}/labels/{label_id}",
|
|
]
|
|
|
|
for expected_template in expected_templates:
|
|
assert expected_template in template_uris, (
|
|
f"Expected template '{expected_template}' not found"
|
|
)
|
|
logger.info("Found expected deck resource template: %s", expected_template)
|
|
|
|
|
|
# Listing resource tests
|
|
async def test_deck_mcp_listing_resources(
|
|
nc_mcp_client: ClientSession, temporary_board_with_card: tuple
|
|
):
|
|
"""Test deck MCP listing resources for stacks and cards."""
|
|
board_data, stack_data, card_data = temporary_board_with_card
|
|
board_id = board_data["id"]
|
|
stack_id = stack_data["id"]
|
|
|
|
# 1. Test listing stacks resource
|
|
logger.info("Reading stacks list via MCP resource for board %s", board_id)
|
|
stacks_resource_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks"
|
|
)
|
|
stacks_resource_data = json.loads(stacks_resource_result.contents[0].text)
|
|
assert isinstance(stacks_resource_data, list)
|
|
|
|
# Verify our stack is in the resource list
|
|
stack_ids = [stack["id"] for stack in stacks_resource_data]
|
|
assert stack_id in stack_ids, "Stack not found in stacks resource list"
|
|
logger.info("Stack found in stacks resource list")
|
|
|
|
# 2. Test listing cards resource
|
|
logger.info("Reading cards list via MCP resource for stack %s", stack_id)
|
|
cards_resource_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/stacks/{stack_id}/cards"
|
|
)
|
|
cards_resource_data = json.loads(cards_resource_result.contents[0].text)
|
|
assert isinstance(cards_resource_data, list)
|
|
|
|
# Verify our card is in the resource list
|
|
card_ids = [card["id"] for card in cards_resource_data]
|
|
assert card_data["id"] in card_ids, "Card not found in cards resource list"
|
|
logger.info("Card found in cards resource list")
|
|
|
|
# 3. Test listing labels resource
|
|
logger.info("Reading labels list via MCP resource for board %s", board_id)
|
|
labels_resource_result = await nc_mcp_client.read_resource(
|
|
f"nc://Deck/boards/{board_id}/labels"
|
|
)
|
|
labels_resource_data = json.loads(labels_resource_result.contents[0].text)
|
|
assert isinstance(labels_resource_data, list)
|
|
logger.info("Labels resource read successfully")
|