diff --git a/.codacy.yaml b/.codacy.yaml index cc6f359d..bac1d0aa 100644 --- a/.codacy.yaml +++ b/.codacy.yaml @@ -1,3 +1,12 @@ --- +# NOTE: Codacy reads this `.codacy.yaml` and ignores the sibling `.codacy.yml` +# (when both exist, `.yaml` wins). The exclusions below must therefore mirror the +# intent expressed in `.codacy.yml`, which is otherwise inert. Test files are +# excluded from static analysis here as well - without this, Codacy applies its +# default pydocstyle profile to tests and flags D203 (blank line before class +# docstring), which conflicts with the D211 convention Ruff enforces project-wide. exclude_paths: - "plugins/**" + - "tests/**" + - "**/test_*.py" + - "**/*_test.py" diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index ce9da7d4..5289d922 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -2,6 +2,7 @@ import logging import os +from datetime import datetime from typing import Any from urllib.parse import urlparse @@ -247,6 +248,7 @@ def update_ticket( priority: str | None = None, owner: str | None = None, group: str | None = None, + pending_time: datetime | str | None = None, time_unit: float | None = None, ) -> dict[str, Any]: """Update an existing ticket.""" @@ -264,6 +266,11 @@ def update_ticket( update_data["owner"] = owner if group is not None: update_data["group"] = group + if pending_time is not None: + # Zammad expects an ISO 8601 string; serialize datetimes for the JSON body. + update_data["pending_time"] = ( + pending_time.isoformat() if isinstance(pending_time, datetime) else pending_time + ) if time_unit is not None: update_data["time_unit"] = time_unit diff --git a/mcp_zammad/models.py b/mcp_zammad/models.py index 29d0a4fb..679ae8bb 100644 --- a/mcp_zammad/models.py +++ b/mcp_zammad/models.py @@ -181,6 +181,16 @@ class PriorityBrief(BaseModel): active: bool = True +class Attachment(BaseModel): + """Ticket article attachment information.""" + + id: int + filename: str + size: int | None = None + content_type: str | None = None + created_at: datetime | None = None + + class Article(BaseModel): """Ticket article (comment/note).""" @@ -201,6 +211,9 @@ class Article(BaseModel): updated_at: datetime created_by: UserBrief | str | None = None updated_by: UserBrief | str | None = None + attachments: list[Attachment] | None = Field( + None, description="Files attached to this article; download via zammad_download_attachment using their id" + ) class Ticket(BaseModel): @@ -306,16 +319,6 @@ class TicketSearchParams(StrictBaseModel): response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") -class Attachment(BaseModel): - """Ticket article attachment information.""" - - id: int - filename: str - size: int | None = None - content_type: str | None = None - created_at: datetime | None = None - - class ArticleCreate(StrictBaseModel): """Create article request with optional attachments.""" @@ -373,6 +376,13 @@ class TicketUpdateParams(StrictBaseModel): priority: str | None = Field(None, description="New priority name", max_length=100) owner: str | None = Field(None, description="New owner login/email", max_length=255) group: str | None = Field(None, description="New group name", max_length=100) + pending_time: datetime | None = Field( + None, + description=( + "Pending-until timestamp (ISO 8601, e.g. '2026-07-01T08:00:00Z'). " + "Required by Zammad when state is 'pending reminder' or 'pending close'." + ), + ) time_unit: float | None = Field( None, description="Time spent for time accounting (unit defined in Zammad admin settings)", gt=0 ) @@ -383,6 +393,13 @@ def sanitize_title(cls, v: str | None) -> str | None: """Escape HTML to prevent XSS attacks.""" return html.escape(v) if v else v + @model_validator(mode="after") + def require_pending_time_for_pending_states(self) -> "TicketUpdateParams": + """Fail fast when moving to a pending state without a pending_time.""" + if self.state is not None and "pending" in self.state.lower() and self.pending_time is None: + raise ValueError(f"state '{self.state}' requires 'pending_time' (the pending-until timestamp, ISO 8601).") + return self + class GetArticleAttachmentsParams(StrictBaseModel): """Get article attachments request parameters.""" diff --git a/mcp_zammad/server.py b/mcp_zammad/server.py index 4a5f5443..81cf3c70 100644 --- a/mcp_zammad/server.py +++ b/mcp_zammad/server.py @@ -205,6 +205,24 @@ def _escape_article_body(article: Article) -> str: return html.escape(article.body) if "html" in ct else article.body +def _sanitize_inline_text(value: object) -> str: + """Neutralize control characters and HTML in a value rendered inline in markdown. + + Attachment filenames originate from user uploads, so they may contain + newlines, control characters, or HTML/Markdown metacharacters that could + break out of the list item or inject markup. Strip non-printable characters + and HTML-escape the rest. + + Args: + value: The value to sanitize (coerced to str) + + Returns: + A single-line, HTML-escaped representation safe for inline rendering + """ + text = "".join(ch for ch in str(value) if ch.isprintable()) + return html.escape(text, quote=False) + + def _serialize_json(obj: dict[str, Any], *, use_compact: bool) -> str: """Serialize JSON object with appropriate formatting. @@ -597,16 +615,20 @@ def _format_ticket_detail_markdown(ticket: Ticket) -> str: type_field = article.get("type", "Unknown") created_at = article.get("created_at", "Unknown") body = article.get("body", "") + attachments = article.get("attachments") else: # Article object - use attribute access from_field = article.from_ or "Unknown" type_field = article.type created_at = article.created_at body = article.body + attachments = article.attachments lines.append(f"- **From**: {from_field}") lines.append(f"- **Type**: {type_field}") lines.append(f"- **Created**: {created_at}") + article_id = article.get("id") if isinstance(article, dict) else article.id + lines.extend(_format_article_attachments(attachments, article_id)) lines.append("") # Truncate very long bodies if len(body) > ARTICLE_BODY_TRUNCATE_LENGTH: @@ -617,6 +639,42 @@ def _format_ticket_detail_markdown(ticket: Ticket) -> str: return "\n".join(lines) +def _attachment_field(att: Attachment | dict, name: str) -> object: + """Read a field from an attachment in either dict or model form.""" + return att.get(name) if isinstance(att, dict) else getattr(att, name, None) + + +def _format_attachment_size(size: object) -> str: + """Render a trailing size suffix for genuine non-negative byte counts.""" + if isinstance(size, int) and not isinstance(size, bool) and size >= 0: + return f", {size} bytes" + return "" + + +def _format_attachment_line(att: Attachment | dict) -> str: + """Format a single attachment as a sanitized markdown bullet line.""" + filename = _attachment_field(att, "filename") + safe_id = _sanitize_inline_text(_attachment_field(att, "id")) + safe_filename = _sanitize_inline_text(filename) if filename is not None else "(unnamed)" + size_str = _format_attachment_size(_attachment_field(att, "size")) + return f" - id={safe_id}: {safe_filename}{size_str}" + + +def _format_article_attachments(attachments: list[Attachment] | list[dict] | None, article_id: int) -> list[str]: + """Render an article's attachment list as markdown lines. + + Surfaces attachment id/filename/size so the LLM knows files exist and can + fetch their content via zammad_download_attachment. + """ + if not attachments: + return [] + + safe_article_id = _sanitize_inline_text(article_id) + lines = [f"- **Attachments** (download via zammad_download_attachment, article_id={safe_article_id}):"] + lines.extend(_format_attachment_line(att) for att in attachments) + return lines + + def _format_user_contact_section(user: User) -> list[str]: """Build contact information section for user markdown.""" fields = [] @@ -1129,6 +1187,8 @@ def zammad_update_ticket(params: TicketUpdateParams) -> Ticket: - group (str | None): New group name - owner (str | None): New owner email/login - customer (str | None): New customer email/login + - pending_time (datetime | None): Pending-until timestamp (ISO 8601), + required when state is "pending reminder" or "pending close" - time_unit (float | None): Time spent for time accounting Returns: @@ -1148,6 +1208,8 @@ def zammad_update_ticket(params: TicketUpdateParams) -> Ticket: Examples: - Use when: "Change ticket 123 to high priority" -> ticket_id=123, priority="high" - Use when: "Close ticket 123" -> ticket_id=123, state="closed" + - Use when: "Set ticket 123 to pending until 2026-07-01" -> + ticket_id=123, state="pending reminder", pending_time="2026-07-01T08:00:00Z" - Use when: "Reassign ticket to Alice" -> ticket_id=123, owner="alice@company.com" - Don't use when: Adding comments (use zammad_add_article) - Don't use when: Adding tags (use zammad_add_ticket_tag) @@ -2508,6 +2570,7 @@ def get_ticket_resource(ticket_id: str) -> str: [ f"--- {article.created_at.isoformat()} by {created_by_email} ---", _escape_article_body(article), + *_format_article_attachments(article.attachments, article.id), "", ] ) diff --git a/tests/test_client_methods.py b/tests/test_client_methods.py index 1bf84876..d4c7f9e0 100644 --- a/tests/test_client_methods.py +++ b/tests/test_client_methods.py @@ -3,6 +3,7 @@ import os import pathlib from collections.abc import Generator +from datetime import datetime, timezone from unittest.mock import Mock, patch import pytest @@ -98,6 +99,33 @@ def test_update_ticket_without_time_unit_excludes_field(self, mock_zammad_api: M call_args = mock_instance.ticket.update.call_args[0][1] assert "time_unit" not in call_args + def test_update_ticket_serializes_pending_time(self, mock_zammad_api: Mock) -> None: + """Test that a datetime pending_time is serialized to an ISO 8601 string.""" + mock_instance = Mock() + mock_instance.ticket.update.return_value = {"id": 1} + mock_zammad_api.return_value = mock_instance + + client = ZammadClient(url="https://test.zammad.com/api/v1", http_token="test-token") + + client.update_ticket(1, state="pending reminder", pending_time=datetime(2026, 7, 1, 8, 0, tzinfo=timezone.utc)) + + call_args = mock_instance.ticket.update.call_args[0][1] + assert call_args["pending_time"] == "2026-07-01T08:00:00+00:00" + assert call_args["state"] == "pending reminder" + + def test_update_ticket_passes_pending_time_string_through(self, mock_zammad_api: Mock) -> None: + """Test that a string pending_time is forwarded unchanged.""" + mock_instance = Mock() + mock_instance.ticket.update.return_value = {"id": 1} + mock_zammad_api.return_value = mock_instance + + client = ZammadClient(url="https://test.zammad.com/api/v1", http_token="test-token") + + client.update_ticket(1, state="pending reminder", pending_time="2026-07-01T08:00:00Z") + + call_args = mock_instance.ticket.update.call_args[0][1] + assert call_args["pending_time"] == "2026-07-01T08:00:00Z" + @pytest.mark.parametrize("time_unit", [0, -5]) def test_update_ticket_rejects_invalid_time_unit(self, mock_zammad_api: Mock, time_unit: float) -> None: """Test update_ticket rejects non-positive time_unit values before API calls.""" diff --git a/tests/test_models.py b/tests/test_models.py index f9761bea..9eb64196 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -4,6 +4,7 @@ from pydantic import ValidationError from mcp_zammad.models import ( + Article, ArticleCreate, AttachmentUpload, DeleteAttachmentParams, @@ -14,6 +15,41 @@ TicketUpdate, ) +_BASE_ARTICLE = { + "id": 456, + "ticket_id": 123, + "type": "email", + "sender": "Customer", + "body": "See attached.", + "created_by_id": 2, + "updated_by_id": 2, + "created_at": "2026-05-30T10:00:00Z", + "updated_at": "2026-05-30T10:00:00Z", +} + + +class TestArticleAttachments: + """Tests for attachment metadata on read Article models.""" + + def test_article_parses_attachments(self): + """Article exposes attachment metadata returned by the Zammad API.""" + article = Article( + **_BASE_ARTICLE, + attachments=[ + {"id": 1, "filename": "kaufanfrage.pdf", "size": 20480}, + {"id": 2, "filename": "logo.png", "size": 2048}, + ], + ) + assert article.attachments is not None + assert [a.id for a in article.attachments] == [1, 2] + assert article.attachments[0].filename == "kaufanfrage.pdf" + assert article.attachments[0].size == 20480 + + def test_article_without_attachments_defaults_none(self): + """Articles with no attachments key default to None.""" + article = Article(**_BASE_ARTICLE) + assert article.attachments is None + class TestTicketCreate: """Test TicketCreate model validation.""" diff --git a/tests/test_server.py b/tests/test_server.py index e5e7e7be..620b8bcc 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -47,6 +47,7 @@ CHARACTER_LIMIT, AttachmentDeletionError, ZammadMCPServer, + _format_article_attachments, _format_ticket_detail_markdown, main, mcp, @@ -1042,6 +1043,43 @@ def test_update_ticket_valid_time_unit(): assert params_none.time_unit is None +def test_update_ticket_pending_state_requires_pending_time(): + """Moving to a pending state without pending_time is rejected up front.""" + with pytest.raises(ValidationError, match="pending_time"): + TicketUpdateParams(ticket_id=1, state="pending reminder") + + with pytest.raises(ValidationError, match="pending_time"): + TicketUpdateParams(ticket_id=1, state="pending close") + + +def test_update_ticket_pending_state_with_pending_time(): + """pending_time is accepted (and parsed) alongside a pending state.""" + params = TicketUpdateParams(ticket_id=1, state="pending reminder", pending_time="2026-07-01T08:00:00Z") + assert params.pending_time == datetime(2026, 7, 1, 8, 0, tzinfo=timezone.utc) + + +def test_update_ticket_forwards_pending_time_as_datetime(mock_zammad_client, sample_ticket_data, decorator_capturer): + """The tool forwards pending_time to the client as a datetime object.""" + mock_instance, _ = mock_zammad_client + mock_instance.update_ticket.return_value = sample_ticket_data + + server_inst = ZammadMCPServer() + server_inst.client = mock_instance + + test_tools, capture_tool = decorator_capturer(server_inst.mcp.tool) + server_inst.mcp.tool = capture_tool # type: ignore[method-assign, assignment] + server_inst.get_client = lambda: server_inst.client # type: ignore[method-assign, assignment, return-value] + server_inst._setup_tools() + + params = TicketUpdateParams(ticket_id=1, state="pending reminder", pending_time="2026-07-01T08:00:00Z") + test_tools["zammad_update_ticket"](params) + + _, kwargs = mock_instance.update_ticket.call_args + assert kwargs["ticket_id"] == 1 + assert kwargs["state"] == "pending reminder" + assert kwargs["pending_time"] == datetime(2026, 7, 1, 8, 0, tzinfo=timezone.utc) + + def test_get_organization_tool(mock_zammad_client, sample_organization_data): """Test get organization tool.""" mock_instance, _ = mock_zammad_client @@ -2987,6 +3025,61 @@ def test_format_ticket_detail_markdown_with_articles(sample_ticket_data, sample_ assert "Second article" in result +def test_format_ticket_detail_markdown_with_attachments(sample_ticket_data, sample_article_data): + """Attachments are surfaced with id/filename so they can be downloaded.""" + article = Article( + **{ + **sample_article_data, + "attachments": [ + {"id": 1, "filename": "kaufanfrage.pdf", "size": 20480}, + {"id": 2, "filename": "logo.png", "size": 2048}, + ], + } + ) + ticket_with_attachments = Ticket(**sample_ticket_data, articles=[article]) + + result = _format_ticket_detail_markdown(ticket_with_attachments) + + assert "**Attachments**" in result + assert "zammad_download_attachment" in result + assert f"article_id={article.id}" in result + assert "id=1: kaufanfrage.pdf, 20480 bytes" in result + assert "id=2: logo.png, 2048 bytes" in result + + +def test_format_article_attachments_sanitizes_filename(): + """Filenames with control characters or HTML are neutralized before rendering.""" + attachments = [ + Attachment(id=1, filename="evil\n- injected line", size=10), + Attachment(id=2, filename=".txt", size=None), + ] + + rendered = "\n".join(_format_article_attachments(attachments, article_id=5)) + + # A newline in the filename must not create an extra markdown line + assert "\n- injected line" not in rendered + assert "evil- injected line" in rendered + # HTML metacharacters are escaped + assert "