From 9441c19a72005e35a360efb00b46de57fb5ad209 Mon Sep 17 00:00:00 2001 From: IchbinkeinReh Date: Wed, 26 Aug 2026 22:50:39 +0200 Subject: [PATCH 1/3] feat(webdav): expose file tags The WebDAV client already covered system tags -- create_tag, get_or_create_tag, assign_tag_to_file, remove_tag_from_file, get_tag_by_name, get_files_by_tag, find_by_tag -- but not one tool exposed any of it, so tagging was unreachable through the server. Adds the two missing read paths on the client (list_tags, get_file_tags) and five tools: nc_webdav_list_tags every tag on the instance nc_webdav_get_file_tags tags of one file nc_webdav_find_by_tag_name files carrying a tag nc_webdav_tag_file attach a tag, creating it if needed nc_webdav_untag_file detach a tag, leaving the tag itself They take tag *names* rather than ids: an id is unguessable for a language model, a name is not. get_file_tags additionally fetches the full list because Nextcloud only reports ids on the file itself. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01GomNRo5jTc6rDNNLYrRVRX --- nextcloud_mcp_server/client/webdav.py | 97 +++++++++++++++++++++++ nextcloud_mcp_server/server/webdav.py | 110 ++++++++++++++++++++++++++ tests/unit/client/test_webdav_tags.py | 78 ++++++++++++++++++ 3 files changed, 285 insertions(+) create mode 100644 tests/unit/client/test_webdav_tags.py diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 0e4677f21..91cea83d3 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -2684,3 +2684,100 @@ async def remove_tag_from_file(self, file_id: int, tag_id: int) -> bool: response.raise_for_status() return True + + # -- File tags (systemtags) --------------------------------------------- + # + # The client already covered tags (create/assign/remove/search); only two + # read paths were missing: list every tag, and list the tags of one file. + + _TAG_PROPFIND = """ + + + + + + + +""" + + async def list_tags(self) -> List[Dict[str, Any]]: + """Every system tag defined on this instance.""" + response = await self._make_request( + "PROPFIND", + "/remote.php/dav/systemtags/", + headers={ + "Depth": "1", + "Content-Type": "text/xml", + "OCS-APIRequest": "true", + }, + content=self._TAG_PROPFIND, + ) + + return self._tags_from_multistatus(ET.fromstring(response.content)) + + @staticmethod + def _tags_from_multistatus(root: Any) -> List[Dict[str, Any]]: + """Extract tags from a systemtags PROPFIND response. + + The collection itself comes back as a response element and is skipped. + An entry without an id or display name cannot be used for anything, so + it is dropped rather than surfaced with placeholder values. + """ + tags: List[Dict[str, Any]] = [] + for response_elem in root.findall("{DAV:}response"): + href = response_elem.find("{DAV:}href") + if href is None or href.text == "/remote.php/dav/systemtags/": + continue + name_elem = response_elem.find(".//{http://owncloud.org/ns}display-name") + id_elem = response_elem.find(".//{http://owncloud.org/ns}id") + if name_elem is None or id_elem is None or not id_elem.text: + continue + assignable = response_elem.find( + ".//{http://owncloud.org/ns}user-assignable" + ) + tags.append( + { + "id": int(id_elem.text), + "name": name_elem.text or "", + "assignable": (assignable is None or assignable.text != "false"), + } + ) + return sorted(tags, key=lambda t: t["name"].lower()) + + async def get_file_tags(self, path: str) -> Dict[str, Any]: + """The tags assigned to one file. + + Nextcloud only reports tag *ids* on the file, so the names are looked + up from the full list. That is one extra call, but ids alone are of + little use to the caller -- and unguessable for a language model. + """ + file_id = await self.get_fileid(path) + if not file_id: + raise ValueError(f"No file id for path: {path}") + + response = await self._make_request( + "PROPFIND", + f"/remote.php/dav/systemtags-relations/files/{file_id}", + headers={ + "Depth": "1", + "Content-Type": "text/xml", + "OCS-APIRequest": "true", + }, + content=self._TAG_PROPFIND, + ) + + root = ET.fromstring(response.content) + ids: List[int] = [] + for id_elem in root.findall(".//{http://owncloud.org/ns}id"): + if id_elem.text and id_elem.text.isdigit(): + ids.append(int(id_elem.text)) + + names = {t["id"]: t["name"] for t in await self.list_tags()} + return { + "path": path, + "file_id": file_id, + "tags": [ + {"id": i, "name": names.get(i, f"(unbekannt: {i})")} + for i in sorted(set(ids)) + ], + } diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 1fc651535..a2ff319f9 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -1153,3 +1153,113 @@ async def nc_webdav_create_comment( comment_id=comment_id, message=message, ) + + # -- File tags ----------------------------------------------------------- + # + # These work with tag *names* rather than ids: an id is unguessable for a + # language model, a name is not. + + @mcp.tool( + title="List Tags", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_list_tags(ctx: Context) -> dict: + """List all system tags available for tagging files.""" + client = await get_client(ctx) + tags = await client.webdav.list_tags() + return {"tags": tags, "total_count": len(tags)} + + @mcp.tool( + title="Get File Tags", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_get_file_tags(path: str, ctx: Context) -> dict: + """List the tags assigned to one file. + + Args: + path: Path to the file, relative to the user's files root. + """ + client = await get_client(ctx) + return await client.webdav.get_file_tags(path) + + @mcp.tool( + title="Find Files By Tag", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> dict: + """Find all files carrying a given tag. + + Args: + tag: Tag name, case-sensitive. + """ + client = await get_client(ctx) + found = await client.webdav.get_tag_by_name(tag) + if found is None or found.get("id") is None: + return { + "tag": tag, + "files": [], + "total_count": 0, + "message": f"No tag named {tag!r} exists.", + } + files = await client.webdav.get_files_by_tag(found["id"]) + return { + "tag": tag, + "tag_id": found["id"], + "files": files, + "total_count": len(files), + } + + @mcp.tool( + title="Tag File", + annotations=ToolAnnotations(readOnlyHint=False, openWorldHint=True), + ) + @require_scopes("files.write") + @instrument_tool + async def nc_webdav_tag_file(path: str, tag: str, ctx: Context) -> dict: + """Attach a tag to a file, creating the tag if it does not exist yet. + + Args: + path: Path to the file, relative to the user's files root. + tag: Tag name. + """ + client = await get_client(ctx) + file_id = await client.webdav.get_fileid(path) + if not file_id: + raise ValueError(f"No file id for path: {path}") + resolved = await client.webdav.get_or_create_tag(tag) + await client.webdav.assign_tag_to_file(int(file_id), resolved["id"]) + return {"path": path, "tag": tag, "tag_id": resolved["id"], "assigned": True} + + @mcp.tool( + title="Untag File", + annotations=ToolAnnotations(readOnlyHint=False, openWorldHint=True), + ) + @require_scopes("files.write") + @instrument_tool + async def nc_webdav_untag_file(path: str, tag: str, ctx: Context) -> dict: + """Remove a tag from a file. The tag itself keeps existing. + + Args: + path: Path to the file, relative to the user's files root. + tag: Tag name. + """ + client = await get_client(ctx) + file_id = await client.webdav.get_fileid(path) + if not file_id: + raise ValueError(f"No file id for path: {path}") + found = await client.webdav.get_tag_by_name(tag) + if found is None or found.get("id") is None: + return { + "path": path, + "tag": tag, + "removed": False, + "message": f"No tag named {tag!r} exists.", + } + await client.webdav.remove_tag_from_file(int(file_id), found["id"]) + return {"path": path, "tag": tag, "tag_id": found["id"], "removed": True} diff --git a/tests/unit/client/test_webdav_tags.py b/tests/unit/client/test_webdav_tags.py new file mode 100644 index 000000000..622e7061e --- /dev/null +++ b/tests/unit/client/test_webdav_tags.py @@ -0,0 +1,78 @@ +"""Parsing of the systemtags collection. + +Tags come back as a PROPFIND multistatus. The collection itself appears as a +response element and must be skipped, and an entry without an id or display +name is unusable and dropped rather than surfaced with placeholder values. +""" + +import pytest + +from nextcloud_mcp_server.client.webdav import WebDAVClient + +pytestmark = pytest.mark.unit + +NS = 'xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns"' + + +def _multistatus(*entries: str) -> bytes: + return ( + f'' + "/remote.php/dav/systemtags/" + + "".join(entries) + + "" + ).encode("utf-8") + + +def _tag(tag_id: str, name: str, assignable: str = "true") -> str: + return ( + f"/remote.php/dav/systemtags/{tag_id}" + f"{tag_id}" + f"{name}" + f"{assignable}" + "" + ) + + +def _parse(payload: bytes) -> list[dict]: + """Run the client's own extraction over a canned response body.""" + from lxml import etree + + return WebDAVClient._tags_from_multistatus(etree.fromstring(payload)) + + +class TestTagParsing: + def test_collection_itself_is_skipped(self): + assert _parse(_multistatus()) == [] + + def test_tags_are_returned(self): + tags = _parse(_multistatus(_tag("7", "Invoice"), _tag("8", "Tax"))) + assert [t["id"] for t in tags] == [7, 8] + + def test_sorted_case_insensitively(self): + tags = _parse(_multistatus(_tag("1", "zebra"), _tag("2", "Apple"))) + assert [t["name"] for t in tags] == ["Apple", "zebra"] + + def test_assignable_flag(self): + tags = _parse(_multistatus(_tag("1", "Locked", assignable="false"))) + assert tags[0]["assignable"] is False + + def test_entry_without_id_is_dropped(self): + broken = ( + "/remote.php/dav/systemtags/9" + "" + "Nameless" + "" + ) + assert _parse(_multistatus(broken)) == [] + + +class TestPropfindBody: + def test_requests_the_tag_properties(self): + body = WebDAVClient._TAG_PROPFIND + for prop in ("oc:id", "oc:display-name", "oc:user-assignable"): + assert prop in body + + def test_is_well_formed(self): + from lxml import etree + + etree.fromstring(WebDAVClient._TAG_PROPFIND.encode("utf-8")) From 59c66852f553b89eaebc85a4d175f0926368cc09 Mon Sep 17 00:00:00 2001 From: IchbinkeinReh Date: Wed, 26 Aug 2026 23:10:45 +0200 Subject: [PATCH 2/3] refactor(webdav): typed responses and exclusion guards for the tag tools Review feedback: - Return BaseResponse models (ListTagsResponse, FileTagsResponse, FilesByTagResponse, TagFileResponse) with SystemTag as the item model, instead of raw dicts. - Route path resolution through the shared helper so EXCLUDED_TAGS is enforced and a missing or non-numeric fileid becomes a ToolError rather than a ValueError. The helper is renamed _resolve_file_id, since it is no longer specific to comments. find_by_tag additionally filters excluded paths out of its results -- a tag can be attached to an excluded file. - Annotate the mutating tools with destructiveHint/idempotentHint per ADR-017. Tagging and untagging are both idempotent. - Drop tags whose display name is empty or whitespace: the check only verified the element existed, so such a tag was surfaced with an empty name and an ambiguous sort key, contradicting the docstring. - Replace the German placeholder for an unmapped tag id with English. Adds tests/server/test_webdav_tags_mcp.py for the tag lifecycle and the refusal cases, plus unit tests for the empty-name handling. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01GomNRo5jTc6rDNNLYrRVRX --- nextcloud_mcp_server/client/webdav.py | 8 +- nextcloud_mcp_server/models/webdav.py | 43 ++++++++++ nextcloud_mcp_server/server/webdav.py | 117 +++++++++++++++----------- tests/server/test_webdav_tags_mcp.py | 117 ++++++++++++++++++++++++++ tests/unit/client/test_webdav_tags.py | 15 ++++ 5 files changed, 248 insertions(+), 52 deletions(-) create mode 100644 tests/server/test_webdav_tags_mcp.py diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 91cea83d3..00b55316d 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -2730,7 +2730,11 @@ def _tags_from_multistatus(root: Any) -> List[Dict[str, Any]]: continue name_elem = response_elem.find(".//{http://owncloud.org/ns}display-name") id_elem = response_elem.find(".//{http://owncloud.org/ns}id") - if name_elem is None or id_elem is None or not id_elem.text: + # An entry without a usable id or name cannot be acted on, and an + # empty display name would also make the sort key ambiguous. + if id_elem is None or not id_elem.text: + continue + if name_elem is None or not (name_elem.text or "").strip(): continue assignable = response_elem.find( ".//{http://owncloud.org/ns}user-assignable" @@ -2777,7 +2781,7 @@ async def get_file_tags(self, path: str) -> Dict[str, Any]: "path": path, "file_id": file_id, "tags": [ - {"id": i, "name": names.get(i, f"(unbekannt: {i})")} + {"id": i, "name": names.get(i, f"(unknown tag {i})")} for i in sorted(set(ids)) ], } diff --git a/nextcloud_mcp_server/models/webdav.py b/nextcloud_mcp_server/models/webdav.py index c677abc42..c0dce693c 100644 --- a/nextcloud_mcp_server/models/webdav.py +++ b/nextcloud_mcp_server/models/webdav.py @@ -255,3 +255,46 @@ class SearchFilesResponse(BaseResponse): filters_applied: Optional[dict] = Field( None, description="Filters that were applied to the search" ) + + +class SystemTag(BaseModel): + """A system tag defined on the instance.""" + + id: int = Field(description="Tag id") + name: str = Field(description="Tag display name") + assignable: bool = Field( + default=True, description="Whether users may assign this tag" + ) + + +class ListTagsResponse(BaseResponse): + """Every system tag on the instance.""" + + tags: List[SystemTag] = Field(default_factory=list, description="Tags, by name") + total_count: int = Field(description="Number of tags") + + +class FileTagsResponse(BaseResponse): + """Tags assigned to one file.""" + + path: str = Field(description="Path the tags belong to") + file_id: str = Field(description="Nextcloud file id") + tags: List[SystemTag] = Field(default_factory=list, description="Assigned tags") + + +class FilesByTagResponse(BaseResponse): + """Files carrying a given tag.""" + + tag: str = Field(description="Tag name that was searched for") + tag_id: Optional[int] = Field(None, description="Tag id, when the tag exists") + files: List[FileInfo] = Field(default_factory=list, description="Matching files") + total_count: int = Field(description="Number of matching files") + + +class TagFileResponse(BaseResponse): + """Outcome of attaching or detaching a tag.""" + + path: str = Field(description="Path that was tagged or untagged") + tag: str = Field(description="Tag name") + tag_id: int = Field(description="Tag id") + assigned: bool = Field(description="True after tagging, False after untagging") diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index a2ff319f9..322654d4d 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -27,7 +27,14 @@ SearchFilesResponse, WriteFileResponse, ) -from nextcloud_mcp_server.models.webdav import ParseStatus +from nextcloud_mcp_server.models.webdav import ( + FilesByTagResponse, + FileTagsResponse, + ListTagsResponse, + ParseStatus, + SystemTag, + TagFileResponse, +) from nextcloud_mcp_server.observability.metrics import instrument_tool from nextcloud_mcp_server.server.tag_exclusion import ( get_excluded_file_paths, @@ -164,11 +171,11 @@ def _read_capped() -> bytes | None: ) -async def _resolve_commented_file(client: "NextcloudClient", path: str) -> int: - """Resolve ``path`` to the file ID the comments collection is keyed by. +async def _resolve_file_id(client: "NextcloudClient", path: str) -> int: + """Resolve ``path`` to the numeric file ID the DAV collections are keyed by. - Shared by both comment tools so the excluded-tag guard and the - does-it-exist check cannot drift between reading and writing comments. + Shared by the comment and tag tools so the excluded-tag guard and the + does-it-exist check cannot drift between them. Raises: ToolError: If the path is excluded by tag, resolves to nothing, or @@ -1083,7 +1090,7 @@ async def nc_webdav_list_comments( raise ValueError(f"offset must not be negative, got {offset}") client = await get_client(ctx) - file_id = await _resolve_commented_file(client, path) + file_id = await _resolve_file_id(client, path) comments = await client.webdav.list_comments( file_id, limit=limit, offset=offset @@ -1144,7 +1151,7 @@ async def nc_webdav_create_comment( ) client = await get_client(ctx) - file_id = await _resolve_commented_file(client, path) + file_id = await _resolve_file_id(client, path) comment_id = await client.webdav.create_comment(file_id, message) return CreateFileCommentResponse( @@ -1154,22 +1161,17 @@ async def nc_webdav_create_comment( message=message, ) - # -- File tags ----------------------------------------------------------- - # - # These work with tag *names* rather than ids: an id is unguessable for a - # language model, a name is not. - @mcp.tool( title="List Tags", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("files.read") @instrument_tool - async def nc_webdav_list_tags(ctx: Context) -> dict: + async def nc_webdav_list_tags(ctx: Context) -> ListTagsResponse: """List all system tags available for tagging files.""" client = await get_client(ctx) - tags = await client.webdav.list_tags() - return {"tags": tags, "total_count": len(tags)} + tags = [SystemTag(**t) for t in await client.webdav.list_tags()] + return ListTagsResponse(tags=tags, total_count=len(tags)) @mcp.tool( title="Get File Tags", @@ -1177,14 +1179,23 @@ async def nc_webdav_list_tags(ctx: Context) -> dict: ) @require_scopes("files.read") @instrument_tool - async def nc_webdav_get_file_tags(path: str, ctx: Context) -> dict: + async def nc_webdav_get_file_tags(path: str, ctx: Context) -> FileTagsResponse: """List the tags assigned to one file. Args: path: Path to the file, relative to the user's files root. """ client = await get_client(ctx) - return await client.webdav.get_file_tags(path) + # Resolving through the shared helper applies the excluded-tag guard, + # so an excluded path cannot be probed for existence via its tags. + await _resolve_file_id(client, path) + + data = await client.webdav.get_file_tags(path) + return FileTagsResponse( + path=data["path"], + file_id=str(data["file_id"]), + tags=[SystemTag(id=t["id"], name=t["name"]) for t in data["tags"]], + ) @mcp.tool( title="Find Files By Tag", @@ -1192,7 +1203,7 @@ async def nc_webdav_get_file_tags(path: str, ctx: Context) -> dict: ) @require_scopes("files.read") @instrument_tool - async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> dict: + async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> FilesByTagResponse: """Find all files carrying a given tag. Args: @@ -1201,27 +1212,36 @@ async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> dict: client = await get_client(ctx) found = await client.webdav.get_tag_by_name(tag) if found is None or found.get("id") is None: - return { - "tag": tag, - "files": [], - "total_count": 0, - "message": f"No tag named {tag!r} exists.", - } + return FilesByTagResponse(tag=tag, files=[], total_count=0) + files = await client.webdav.get_files_by_tag(found["id"]) - return { - "tag": tag, - "tag_id": found["id"], - "files": files, - "total_count": len(files), - } + + # A tag can be attached to an excluded file; the listing must not + # surface it any more than a directory listing would. + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + files = [ + f for f in files if not is_path_excluded(f.get("path", ""), excluded) + ] + + return FilesByTagResponse( + tag=tag, + tag_id=found["id"], + files=[FileInfo(**f) for f in files], + total_count=len(files), + ) @mcp.tool( title="Tag File", - annotations=ToolAnnotations(readOnlyHint=False, openWorldHint=True), + annotations=ToolAnnotations( + destructiveHint=False, + idempotentHint=True, # Same tag twice = same end state + openWorldHint=True, + ), ) @require_scopes("files.write") @instrument_tool - async def nc_webdav_tag_file(path: str, tag: str, ctx: Context) -> dict: + async def nc_webdav_tag_file(path: str, tag: str, ctx: Context) -> TagFileResponse: """Attach a tag to a file, creating the tag if it does not exist yet. Args: @@ -1229,20 +1249,24 @@ async def nc_webdav_tag_file(path: str, tag: str, ctx: Context) -> dict: tag: Tag name. """ client = await get_client(ctx) - file_id = await client.webdav.get_fileid(path) - if not file_id: - raise ValueError(f"No file id for path: {path}") + file_id = await _resolve_file_id(client, path) resolved = await client.webdav.get_or_create_tag(tag) - await client.webdav.assign_tag_to_file(int(file_id), resolved["id"]) - return {"path": path, "tag": tag, "tag_id": resolved["id"], "assigned": True} + await client.webdav.assign_tag_to_file(file_id, resolved["id"]) + return TagFileResponse(path=path, tag=tag, tag_id=resolved["id"], assigned=True) @mcp.tool( title="Untag File", - annotations=ToolAnnotations(readOnlyHint=False, openWorldHint=True), + annotations=ToolAnnotations( + destructiveHint=False, # The tag itself survives + idempotentHint=True, # Removing an absent tag = same end state + openWorldHint=True, + ), ) @require_scopes("files.write") @instrument_tool - async def nc_webdav_untag_file(path: str, tag: str, ctx: Context) -> dict: + async def nc_webdav_untag_file( + path: str, tag: str, ctx: Context + ) -> TagFileResponse: """Remove a tag from a file. The tag itself keeps existing. Args: @@ -1250,16 +1274,9 @@ async def nc_webdav_untag_file(path: str, tag: str, ctx: Context) -> dict: tag: Tag name. """ client = await get_client(ctx) - file_id = await client.webdav.get_fileid(path) - if not file_id: - raise ValueError(f"No file id for path: {path}") + file_id = await _resolve_file_id(client, path) found = await client.webdav.get_tag_by_name(tag) if found is None or found.get("id") is None: - return { - "path": path, - "tag": tag, - "removed": False, - "message": f"No tag named {tag!r} exists.", - } - await client.webdav.remove_tag_from_file(int(file_id), found["id"]) - return {"path": path, "tag": tag, "tag_id": found["id"], "removed": True} + raise ToolError(f"No tag named {tag!r} exists") + await client.webdav.remove_tag_from_file(file_id, found["id"]) + return TagFileResponse(path=path, tag=tag, tag_id=found["id"], assigned=False) diff --git a/tests/server/test_webdav_tags_mcp.py b/tests/server/test_webdav_tags_mcp.py new file mode 100644 index 000000000..c7054823a --- /dev/null +++ b/tests/server/test_webdav_tags_mcp.py @@ -0,0 +1,117 @@ +"""End-to-end tests for the file-tag MCP tools. + +Exercises the real systemtags collections: attach a tag to a file, read it +back, find the file through it, and detach it again. +""" + +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 + + +def _payload(tool_result) -> dict: + """Return the JSON-decoded text content of an MCP tool result.""" + return json.loads(tool_result.content[0].text) + + +@pytest.fixture +async def tagged_file(nc_client: NextcloudClient): + """A file to hang tags off, removed afterwards.""" + path = f"mcp_tags_{uuid.uuid4().hex[:8]}.txt" + await nc_client.webdav.write_file(path, b"tag me", "text/plain") + yield path + try: + await nc_client.webdav.delete_resource(path) + except Exception as e: + logger.warning("Failed to cleanup %s: %s", path, e) + + +async def test_tag_lifecycle(nc_mcp_client: ClientSession, tagged_file: str): + """Tag a file, read it back, find it by tag, then untag it.""" + tag = f"mcp-test-{uuid.uuid4().hex[:8]}" + + tagged = await nc_mcp_client.call_tool( + "nc_webdav_tag_file", {"path": tagged_file, "tag": tag} + ) + assert tagged.isError is False + assert _payload(tagged)["assigned"] is True + + read_back = await nc_mcp_client.call_tool( + "nc_webdav_get_file_tags", {"path": tagged_file} + ) + assert read_back.isError is False + assert tag in [t["name"] for t in _payload(read_back)["tags"]] + + listed = await nc_mcp_client.call_tool("nc_webdav_list_tags", {}) + assert listed.isError is False + assert tag in [t["name"] for t in _payload(listed)["tags"]] + + found = await nc_mcp_client.call_tool("nc_webdav_find_by_tag_name", {"tag": tag}) + assert found.isError is False + by_tag = _payload(found) + assert by_tag["total_count"] >= 1 + assert by_tag["tag_id"] is not None + + untagged = await nc_mcp_client.call_tool( + "nc_webdav_untag_file", {"path": tagged_file, "tag": tag} + ) + assert untagged.isError is False + assert _payload(untagged)["assigned"] is False + + after = await nc_mcp_client.call_tool( + "nc_webdav_get_file_tags", {"path": tagged_file} + ) + assert tag not in [t["name"] for t in _payload(after)["tags"]] + + +async def test_tagging_is_idempotent(nc_mcp_client: ClientSession, tagged_file: str): + """Applying the same tag twice leaves the same end state.""" + tag = f"mcp-test-{uuid.uuid4().hex[:8]}" + for _ in range(2): + result = await nc_mcp_client.call_tool( + "nc_webdav_tag_file", {"path": tagged_file, "tag": tag} + ) + assert result.isError is False + + read_back = await nc_mcp_client.call_tool( + "nc_webdav_get_file_tags", {"path": tagged_file} + ) + names = [t["name"] for t in _payload(read_back)["tags"]] + assert names.count(tag) == 1 + + +async def test_unknown_tag_yields_empty_result(nc_mcp_client: ClientSession): + """Searching for a tag that does not exist is not an error.""" + result = await nc_mcp_client.call_tool( + "nc_webdav_find_by_tag_name", {"tag": f"nope-{uuid.uuid4().hex}"} + ) + assert result.isError is False + payload = _payload(result) + assert payload["total_count"] == 0 + assert payload["tag_id"] is None + + +async def test_untagging_an_unknown_tag_is_refused( + nc_mcp_client: ClientSession, tagged_file: str +): + result = await nc_mcp_client.call_tool( + "nc_webdav_untag_file", + {"path": tagged_file, "tag": f"nope-{uuid.uuid4().hex}"}, + ) + assert result.isError is True + + +async def test_tagging_a_missing_file_is_refused(nc_mcp_client: ClientSession): + result = await nc_mcp_client.call_tool( + "nc_webdav_tag_file", + {"path": f"nope_{uuid.uuid4().hex}.txt", "tag": "whatever"}, + ) + assert result.isError is True diff --git a/tests/unit/client/test_webdav_tags.py b/tests/unit/client/test_webdav_tags.py index 622e7061e..c22ba717e 100644 --- a/tests/unit/client/test_webdav_tags.py +++ b/tests/unit/client/test_webdav_tags.py @@ -65,6 +65,21 @@ def test_entry_without_id_is_dropped(self): ) assert _parse(_multistatus(broken)) == [] + def test_entry_without_display_name_is_dropped(self): + broken = ( + "/remote.php/dav/systemtags/9" + "9" + "" + ) + assert _parse(_multistatus(broken)) == [] + + def test_entry_with_empty_display_name_is_dropped(self): + """An empty name would be kept and make the sort key ambiguous.""" + assert _parse(_multistatus(_tag("9", ""))) == [] + + def test_entry_with_whitespace_display_name_is_dropped(self): + assert _parse(_multistatus(_tag("9", " "))) == [] + class TestPropfindBody: def test_requests_the_tag_properties(self): From f42814ce71d1fadf49d0e29166ce4010548f67cb Mon Sep 17 00:00:00 2001 From: IchbinkeinReh Date: Mon, 31 Aug 2026 10:48:45 +0200 Subject: [PATCH 3/3] fix(webdav): apply @with_links to nc_webdav_find_by_tag_name CI caught it: this tool returns FilesByTagResponse.files (FileInfo entries), which is a registered linkable model, but the tool never applied the decorator that fills in each entry's url field -- so url was always None, same class of bug test_links_tool_coverage.py exists to catch (see its docstring re: the Deck card tools). Added a direct test alongside the existing DirectoryListing one, since the coverage test only proves the decorator is present, not that it actually populates anything for this tool's specific response shape. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01GomNRo5jTc6rDNNLYrRVRX --- nextcloud_mcp_server/server/webdav.py | 1 + tests/unit/test_links.py | 23 ++++++++++++++++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 322654d4d..f4303582e 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -1202,6 +1202,7 @@ async def nc_webdav_get_file_tags(path: str, ctx: Context) -> FileTagsResponse: annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("files.read") + @with_links @instrument_tool async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> FilesByTagResponse: """Find all files carrying a given tag. diff --git a/tests/unit/test_links.py b/tests/unit/test_links.py index 9beb0224e..c0c101a34 100644 --- a/tests/unit/test_links.py +++ b/tests/unit/test_links.py @@ -33,7 +33,11 @@ NoteSearchResult, SearchNotesResponse, ) -from nextcloud_mcp_server.models.webdav import DirectoryListing, FileInfo +from nextcloud_mcp_server.models.webdav import ( + DirectoryListing, + FileInfo, + FilesByTagResponse, +) pytestmark = pytest.mark.unit @@ -196,6 +200,23 @@ def test_directory_listing_links_each_entry_that_can_be_linked(): assert listing.files[1].url is None +def test_files_by_tag_links_each_entry_that_can_be_linked(): + """nc_webdav_find_by_tag_name's results link the same way a directory listing's do - + both hold plain FileInfo entries, so one registry entry covers both tools.""" + found = FilesByTagResponse( + tag="invoice", + tag_id=7, + files=[ + FileInfo(name="a", path="/a", is_directory=False, file_id=1), + FileInfo(name="b", path="/b", is_directory=False, file_id=None), + ], + total_count=2, + ) + _attach(found) + assert found.files[0].url == f"{BASE}/index.php/f/1" + assert found.files[1].url is None + + # --- absent or unusable configuration ---------------------------------------