diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 0e4677f21..00b55316d 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -2684,3 +2684,104 @@ 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") + # 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" + ) + 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"(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 1fc651535..f4303582e 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( @@ -1153,3 +1160,124 @@ async def nc_webdav_create_comment( comment_id=comment_id, message=message, ) + + @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) -> ListTagsResponse: + """List all system tags available for tagging files.""" + client = await get_client(ctx) + 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", + annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), + ) + @require_scopes("files.read") + @instrument_tool + 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) + # 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", + 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. + + 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 FilesByTagResponse(tag=tag, files=[], total_count=0) + + files = await client.webdav.get_files_by_tag(found["id"]) + + # 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( + 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) -> TagFileResponse: + """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 _resolve_file_id(client, path) + resolved = await client.webdav.get_or_create_tag(tag) + 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( + 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 + ) -> TagFileResponse: + """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 _resolve_file_id(client, path) + found = await client.webdav.get_tag_by_name(tag) + if found is None or found.get("id") is None: + 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 new file mode 100644 index 000000000..c22ba717e --- /dev/null +++ b/tests/unit/client/test_webdav_tags.py @@ -0,0 +1,93 @@ +"""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)) == [] + + 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): + 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")) 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 ---------------------------------------