fix: address PR #574 second review round
- Enrich single-calendar event dicts with calendar_name before mapping to CalendarEventSummary (list_events and upcoming_events paths) - Extract _raw_contact_to_model() from inline mapping in contacts.py, fix custom_fields type annotation to dict[str, Any] - Add unit tests for _event_dict_to_summary covering categories parsing, falsy coercion, and calendar name passthrough - Replace duplicated test helper with import of production function Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
76e305006c
commit
76e6c12b56
@@ -240,6 +240,10 @@ def configure_calendar_tools(mcp: FastMCP):
|
|||||||
limit=limit,
|
limit=limit,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Enrich events with calendar context for per-event mapping
|
||||||
|
for event in events:
|
||||||
|
event["calendar_name"] = calendar_name
|
||||||
|
|
||||||
# Apply filters if provided
|
# Apply filters if provided
|
||||||
if filters:
|
if filters:
|
||||||
events = client.calendar._apply_event_filters(events, filters)
|
events = client.calendar._apply_event_filters(events, filters)
|
||||||
@@ -459,6 +463,8 @@ def configure_calendar_tools(mcp: FastMCP):
|
|||||||
end_datetime=end_datetime,
|
end_datetime=end_datetime,
|
||||||
limit=limit,
|
limit=limit,
|
||||||
)
|
)
|
||||||
|
for event in events:
|
||||||
|
event["calendar_name"] = calendar_name
|
||||||
else:
|
else:
|
||||||
# Get events from all calendars
|
# Get events from all calendars
|
||||||
all_calendars = await client.calendar.list_calendars()
|
all_calendars = await client.calendar.list_calendars()
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import logging
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from mcp.server.fastmcp import Context, FastMCP
|
from mcp.server.fastmcp import Context, FastMCP
|
||||||
from mcp.types import ToolAnnotations
|
from mcp.types import ToolAnnotations
|
||||||
@@ -17,6 +18,34 @@ from nextcloud_mcp_server.observability.metrics import instrument_tool
|
|||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def _raw_contact_to_model(raw: dict) -> Contact:
|
||||||
|
"""Convert a raw contact dict from the contacts client to a Contact model."""
|
||||||
|
contact_info = raw.get("contact", {})
|
||||||
|
|
||||||
|
# Convert email field (str, list, or None) to list[ContactField]
|
||||||
|
raw_email = contact_info.get("email")
|
||||||
|
emails: list[ContactField] = []
|
||||||
|
if isinstance(raw_email, list):
|
||||||
|
emails = [ContactField(type="email", value=e) for e in raw_email if e]
|
||||||
|
elif isinstance(raw_email, str) and raw_email:
|
||||||
|
emails = [ContactField(type="email", value=raw_email)]
|
||||||
|
|
||||||
|
# Nickname goes into custom_fields (no dedicated model field)
|
||||||
|
custom_fields: dict[str, Any] = {}
|
||||||
|
nickname = contact_info.get("nickname")
|
||||||
|
if nickname:
|
||||||
|
custom_fields["nickname"] = nickname
|
||||||
|
|
||||||
|
return Contact(
|
||||||
|
uid=raw["vcard_id"],
|
||||||
|
fn=contact_info.get("fullname", ""),
|
||||||
|
etag=raw.get("getetag"),
|
||||||
|
birthday=contact_info.get("birthday"),
|
||||||
|
emails=emails,
|
||||||
|
custom_fields=custom_fields,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def configure_contacts_tools(mcp: FastMCP):
|
def configure_contacts_tools(mcp: FastMCP):
|
||||||
# Contacts tools
|
# Contacts tools
|
||||||
@mcp.tool(
|
@mcp.tool(
|
||||||
@@ -53,34 +82,7 @@ def configure_contacts_tools(mcp: FastMCP):
|
|||||||
"""List all contacts in the specified addressbook."""
|
"""List all contacts in the specified addressbook."""
|
||||||
client = await get_client(ctx)
|
client = await get_client(ctx)
|
||||||
contacts_data = await client.contacts.list_contacts(addressbook=addressbook)
|
contacts_data = await client.contacts.list_contacts(addressbook=addressbook)
|
||||||
contacts = []
|
contacts = [_raw_contact_to_model(c) for c in contacts_data]
|
||||||
for c in contacts_data:
|
|
||||||
contact_info = c.get("contact", {})
|
|
||||||
|
|
||||||
# Convert email field (str, list, or None) to list[ContactField]
|
|
||||||
raw_email = contact_info.get("email")
|
|
||||||
emails: list[ContactField] = []
|
|
||||||
if isinstance(raw_email, list):
|
|
||||||
emails = [ContactField(type="email", value=e) for e in raw_email if e]
|
|
||||||
elif isinstance(raw_email, str) and raw_email:
|
|
||||||
emails = [ContactField(type="email", value=raw_email)]
|
|
||||||
|
|
||||||
# Nickname goes into custom_fields (no dedicated model field)
|
|
||||||
custom_fields: dict[str, str] = {}
|
|
||||||
nickname = contact_info.get("nickname")
|
|
||||||
if nickname:
|
|
||||||
custom_fields["nickname"] = nickname
|
|
||||||
|
|
||||||
contacts.append(
|
|
||||||
Contact(
|
|
||||||
uid=c["vcard_id"],
|
|
||||||
fn=contact_info.get("fullname", ""),
|
|
||||||
etag=c.get("getetag"),
|
|
||||||
birthday=contact_info.get("birthday"),
|
|
||||||
emails=emails,
|
|
||||||
custom_fields=custom_fields,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
return ListContactsResponse(
|
return ListContactsResponse(
|
||||||
contacts=contacts, addressbook=addressbook, total_count=len(contacts)
|
contacts=contacts, addressbook=addressbook, total_count=len(contacts)
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import pytest
|
|||||||
|
|
||||||
from nextcloud_mcp_server.models.contacts import (
|
from nextcloud_mcp_server.models.contacts import (
|
||||||
Contact,
|
Contact,
|
||||||
ContactField,
|
|
||||||
ListContactsResponse,
|
ListContactsResponse,
|
||||||
)
|
)
|
||||||
from nextcloud_mcp_server.models.notes import (
|
from nextcloud_mcp_server.models.notes import (
|
||||||
@@ -17,6 +16,8 @@ from nextcloud_mcp_server.models.semantic import (
|
|||||||
SamplingSearchResponse,
|
SamplingSearchResponse,
|
||||||
SemanticSearchResult,
|
SemanticSearchResult,
|
||||||
)
|
)
|
||||||
|
from nextcloud_mcp_server.server.calendar import _event_dict_to_summary
|
||||||
|
from nextcloud_mcp_server.server.contacts import _raw_contact_to_model
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
@@ -275,29 +276,8 @@ def test_sampling_search_response_serialization():
|
|||||||
|
|
||||||
|
|
||||||
def _map_contact(raw: dict) -> Contact:
|
def _map_contact(raw: dict) -> Contact:
|
||||||
"""Replicate the mapping logic from server/contacts.py for testing."""
|
"""Thin wrapper around the production mapping function for test readability."""
|
||||||
contact_info = raw.get("contact", {})
|
return _raw_contact_to_model(raw)
|
||||||
|
|
||||||
raw_email = contact_info.get("email")
|
|
||||||
emails: list[ContactField] = []
|
|
||||||
if isinstance(raw_email, list):
|
|
||||||
emails = [ContactField(type="email", value=e) for e in raw_email if e]
|
|
||||||
elif isinstance(raw_email, str) and raw_email:
|
|
||||||
emails = [ContactField(type="email", value=raw_email)]
|
|
||||||
|
|
||||||
custom_fields: dict[str, str] = {}
|
|
||||||
nickname = contact_info.get("nickname")
|
|
||||||
if nickname:
|
|
||||||
custom_fields["nickname"] = nickname
|
|
||||||
|
|
||||||
return Contact(
|
|
||||||
uid=raw["vcard_id"],
|
|
||||||
fn=contact_info.get("fullname", ""),
|
|
||||||
etag=raw.get("getetag"),
|
|
||||||
birthday=contact_info.get("birthday"),
|
|
||||||
emails=emails,
|
|
||||||
custom_fields=custom_fields,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
@@ -394,3 +374,117 @@ def test_list_contacts_response_wraps_contacts():
|
|||||||
assert c["birthday"] == "2000-01-01"
|
assert c["birthday"] == "2000-01-01"
|
||||||
assert c["emails"][0]["value"] == "alice@test.com"
|
assert c["emails"][0]["value"] == "alice@test.com"
|
||||||
assert c["custom_fields"]["nickname"] == "Ali"
|
assert c["custom_fields"]["nickname"] == "Ali"
|
||||||
|
|
||||||
|
|
||||||
|
# ============= _event_dict_to_summary tests =============
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_basic():
|
||||||
|
"""Test basic mapping with all fields populated."""
|
||||||
|
event = {
|
||||||
|
"uid": "evt-001",
|
||||||
|
"title": "Team Standup",
|
||||||
|
"start_datetime": "2025-07-28T09:00:00",
|
||||||
|
"end_datetime": "2025-07-28T09:30:00",
|
||||||
|
"all_day": False,
|
||||||
|
"location": "Room 42",
|
||||||
|
"description": "Daily sync",
|
||||||
|
"categories": ["work", "meeting"],
|
||||||
|
"status": "CONFIRMED",
|
||||||
|
"calendar_name": "office",
|
||||||
|
"calendar_display_name": "Office Calendar",
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.uid == "evt-001"
|
||||||
|
assert summary.summary == "Team Standup"
|
||||||
|
assert summary.start == "2025-07-28T09:00:00"
|
||||||
|
assert summary.end == "2025-07-28T09:30:00"
|
||||||
|
assert summary.all_day is False
|
||||||
|
assert summary.location == "Room 42"
|
||||||
|
assert summary.description == "Daily sync"
|
||||||
|
assert summary.categories == ["work", "meeting"]
|
||||||
|
assert summary.status == "CONFIRMED"
|
||||||
|
assert summary.calendar_name == "office"
|
||||||
|
assert summary.calendar_display_name == "Office Calendar"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_categories_string():
|
||||||
|
"""Test that comma-separated category string is split into a list."""
|
||||||
|
event = {
|
||||||
|
"uid": "evt-002",
|
||||||
|
"title": "Review",
|
||||||
|
"categories": "work, meeting, important",
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.categories == ["work", "meeting", "important"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_categories_list_passthrough():
|
||||||
|
"""Test that a list of categories passes through unchanged."""
|
||||||
|
event = {
|
||||||
|
"uid": "evt-003",
|
||||||
|
"title": "Review",
|
||||||
|
"categories": ["personal", "health"],
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.categories == ["personal", "health"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_falsy_location_description():
|
||||||
|
"""Test that empty/falsy location and description are coerced to None."""
|
||||||
|
event = {
|
||||||
|
"uid": "evt-004",
|
||||||
|
"title": "Quick Chat",
|
||||||
|
"location": "",
|
||||||
|
"description": "",
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.location is None
|
||||||
|
assert summary.description is None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_missing_optional_fields():
|
||||||
|
"""Test mapping with only required fields present."""
|
||||||
|
event = {"uid": "evt-005", "title": "Minimal Event"}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.uid == "evt-005"
|
||||||
|
assert summary.summary == "Minimal Event"
|
||||||
|
assert summary.start == ""
|
||||||
|
assert summary.end is None
|
||||||
|
assert summary.all_day is False
|
||||||
|
assert summary.location is None
|
||||||
|
assert summary.description is None
|
||||||
|
assert summary.categories == []
|
||||||
|
assert summary.status is None
|
||||||
|
assert summary.calendar_name is None
|
||||||
|
assert summary.calendar_display_name is None
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
def test_event_dict_to_summary_calendar_name_without_display_name():
|
||||||
|
"""Test single-calendar path: calendar_name set, display_name absent."""
|
||||||
|
event = {
|
||||||
|
"uid": "evt-006",
|
||||||
|
"title": "Personal Errand",
|
||||||
|
"calendar_name": "personal",
|
||||||
|
}
|
||||||
|
|
||||||
|
summary = _event_dict_to_summary(event)
|
||||||
|
|
||||||
|
assert summary.calendar_name == "personal"
|
||||||
|
assert summary.calendar_display_name is None
|
||||||
|
|||||||
Reference in New Issue
Block a user