test: add comprehensive WebDAV integration tests

- Add 8 core WebDAV operation tests covering CRUD operations
- Add complex attachment cleanup test for category changes
- Fix ruff formatting violations in webdav.py and server.py
- Address PR feedback requirements for expanded WebDAV functionality

Tests focus on WebDAV client functionality and run locally with docker-compose.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: neovasky

nextcloud_mcp_server/client/webdav.py

@@ -249,10 +249,10 @@ async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
         webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
         if not webdav_path.endswith("/"):
             webdav_path += "/"
-        
+
         logger.info(f"Listing directory: {webdav_path}")
-        
-        propfind_body = '''<?xml version="1.0"?>
+
+        propfind_body = """<?xml version="1.0"?>
         <d:propfind xmlns:d="DAV:">
             <d:prop>
                 <d:displayname/>
@@ -261,73 +261,80 @@ async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
                 <d:getlastmodified/>
                 <d:resourcetype/>
             </d:prop>
-        </d:propfind>'''
-        
-        headers = {
-            "Depth": "1",
-            "Content-Type": "text/xml",
-            "OCS-APIRequest": "true"
-        }
-        
+        </d:propfind>"""
+
+        headers = {"Depth": "1", "Content-Type": "text/xml", "OCS-APIRequest": "true"}
+
         try:
             response = await self._client.request(
                 "PROPFIND", webdav_path, content=propfind_body, headers=headers
             )
             response.raise_for_status()
-            
+
             # Parse the XML response
             root = ET.fromstring(response.content)
             items = []
-            
+
             # Skip the first response (the directory itself)
             responses = root.findall(".//{DAV:}response")[1:]
-            
+
             for response_elem in responses:
                 href = response_elem.find(".//{DAV:}href")
                 if href is None:
                     continue
-                    
+
                 # Extract file/directory name from href
                 href_text = href.text or ""
                 name = href_text.rstrip("/").split("/")[-1]
                 if not name:
                     continue
-                
+
                 # Get properties
                 propstat = response_elem.find(".//{DAV:}propstat")
                 if propstat is None:
                     continue
-                    
+
                 prop = propstat.find(".//{DAV:}prop")
                 if prop is None:
                     continue
-                
+
                 # Determine if it's a directory
                 resourcetype = prop.find(".//{DAV:}resourcetype")
-                is_directory = resourcetype is not None and resourcetype.find(".//{DAV:}collection") is not None
-                
+                is_directory = (
+                    resourcetype is not None
+                    and resourcetype.find(".//{DAV:}collection") is not None
+                )
+
                 # Get other properties
                 size_elem = prop.find(".//{DAV:}getcontentlength")
-                size = int(size_elem.text) if size_elem is not None and size_elem.text else 0
-                
+                size = (
+                    int(size_elem.text)
+                    if size_elem is not None and size_elem.text
+                    else 0
+                )
+
                 content_type_elem = prop.find(".//{DAV:}getcontenttype")
-                content_type = content_type_elem.text if content_type_elem is not None else None
-                
+                content_type = (
+                    content_type_elem.text if content_type_elem is not None else None
+                )
+
                 modified_elem = prop.find(".//{DAV:}getlastmodified")
                 modified = modified_elem.text if modified_elem is not None else None
-                
-                items.append({
-                    "name": name,
-                    "path": f"{path.rstrip('/')}/{name}" if path else name,
-                    "is_directory": is_directory,
-                    "size": size if not is_directory else None,
-                    "content_type": content_type,
-                    "last_modified": modified
-                })
-            
+
+                items.append(
+                    {
+                        "name": name,
+                        "path": f"{path.rstrip('/')}/{name}" if path else name,
+                        "is_directory": is_directory,
+                        "size": size if not is_directory else None,
+                        "content_type": content_type,
+                        "last_modified": modified,
+                    }
+                )
+
             logger.info(f"Found {len(items)} items in directory: {webdav_path}")
             return items
-            
+
         except HTTPStatusError as e:
             logger.error(f"HTTP error listing directory '{webdav_path}': {e}")
             raise e
@@ -338,49 +345,56 @@ async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
     async def read_file(self, path: str) -> Tuple[bytes, str]:
         """Read a file's content via WebDAV GET."""
         webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
-        
+
         logger.info(f"Reading file: {webdav_path}")
-        
+
         try:
             response = await self._client.get(webdav_path)
             response.raise_for_status()
-            
+
             content = response.content
-            content_type = response.headers.get("content-type", "application/octet-stream")
-            
-            logger.info(f"Successfully read file '{path}' ({content_type}, {len(content)} bytes)")
+            content_type = response.headers.get(
+                "content-type", "application/octet-stream"
+            )
+
+            logger.info(
+                f"Successfully read file '{path}' ({content_type}, {len(content)} bytes)"
+            )
             return content, content_type
-            
+
         except HTTPStatusError as e:
             logger.error(f"HTTP error reading file '{path}': {e}")
             raise e
         except Exception as e:
             logger.error(f"Unexpected error reading file '{path}': {e}")
             raise e
 
-    async def write_file(self, path: str, content: bytes, content_type: Optional[str] = None) -> Dict[str, Any]:
+    async def write_file(
+        self, path: str, content: bytes, content_type: Optional[str] = None
+    ) -> Dict[str, Any]:
         """Write content to a file via WebDAV PUT."""
         webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
-        
+
         logger.info(f"Writing file: {webdav_path}")
-        
+
         if not content_type:
             content_type, _ = mimetypes.guess_type(path)
             if not content_type:
                 content_type = "application/octet-stream"
-        
-        headers = {
-            "Content-Type": content_type,
-            "OCS-APIRequest": "true"
-        }
-        
+
+        headers = {"Content-Type": content_type, "OCS-APIRequest": "true"}
+
         try:
-            response = await self._client.put(webdav_path, content=content, headers=headers)
+            response = await self._client.put(
+                webdav_path, content=content, headers=headers
+            )
             response.raise_for_status()
-            
-            logger.info(f"Successfully wrote file '{path}' (Status: {response.status_code})")
+
+            logger.info(
+                f"Successfully wrote file '{path}' (Status: {response.status_code})"
+            )
             return {"status_code": response.status_code}
-            
+
         except HTTPStatusError as e:
             logger.error(f"HTTP error writing file '{path}': {e}")
             raise e
@@ -393,20 +407,24 @@ async def create_directory(self, path: str) -> Dict[str, Any]:
         webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
         if not webdav_path.endswith("/"):
             webdav_path += "/"
-        
+
         logger.info(f"Creating directory: {webdav_path}")
-        
+
         headers = {"OCS-APIRequest": "true"}
-        
+
         try:
             response = await self._client.request("MKCOL", webdav_path, headers=headers)
             response.raise_for_status()
-            
-            logger.info(f"Successfully created directory '{path}' (Status: {response.status_code})")
+
+            logger.info(
+                f"Successfully created directory '{path}' (Status: {response.status_code})"
+            )
             return {"status_code": response.status_code}
-            
+
         except HTTPStatusError as e:
-            if e.response.status_code == 405:  # Method Not Allowed - directory already exists
+            if (
+                e.response.status_code == 405
+            ):  # Method Not Allowed - directory already exists
                 logger.info(f"Directory '{path}' already exists")
                 return {"status_code": 405, "message": "Directory already exists"}
             logger.error(f"HTTP error creating directory '{path}': {e}")

nextcloud_mcp_server/server.py

@@ -195,17 +195,17 @@ async def nc_notes_get_attachment(note_id: int, attachment_filename: str):
 @mcp.tool()
 async def nc_webdav_list_directory(ctx: Context, path: str = ""):
     """List files and directories in the specified NextCloud path.
-    
+
     Args:
         path: Directory path to list (empty string for root directory)
-        
+
     Returns:
         List of items with metadata including name, path, is_directory, size, content_type, last_modified
-        
+
     Examples:
         # List root directory
         await nc_webdav_list_directory("")
-        
+
         # List a specific folder
         await nc_webdav_list_directory("Documents/Projects")
     """
@@ -216,26 +216,26 @@ async def nc_webdav_list_directory(ctx: Context, path: str = ""):
 @mcp.tool()
 async def nc_webdav_read_file(path: str, ctx: Context):
     """Read the content of a file from NextCloud.
-    
+
     Args:
         path: Full path to the file to read
-        
+
     Returns:
         Dict with path, content, content_type, size, and encoding (if binary)
         Text files are decoded to UTF-8, binary files are base64 encoded
-        
+
     Examples:
         # Read a text file
         result = await nc_webdav_read_file("Documents/readme.txt")
         print(result['content'])  # Decoded text content
-        
+
         # Read a binary file
         result = await nc_webdav_read_file("Images/photo.jpg")
         print(result['encoding'])  # 'base64'
     """
     client: NextcloudClient = ctx.request_context.lifespan_context.client
     content, content_type = await client.webdav.read_file(path)
-    
+
     # For text files, decode content for easier viewing
     if content_type and content_type.startswith("text/"):
         try:
@@ -244,68 +244,72 @@ async def nc_webdav_read_file(path: str, ctx: Context):
                 "path": path,
                 "content": decoded_content,
                 "content_type": content_type,
-                "size": len(content)
+                "size": len(content),
             }
         except UnicodeDecodeError:
             pass
-    
+
     # For binary files, return metadata and base64 encoded content
     import base64
+
     return {
         "path": path,
         "content": base64.b64encode(content).decode("ascii"),
         "content_type": content_type,
         "size": len(content),
-        "encoding": "base64"
+        "encoding": "base64",
     }
 
 
 @mcp.tool()
-async def nc_webdav_write_file(path: str, content: str, ctx: Context, content_type: str | None = None):
+async def nc_webdav_write_file(
+    path: str, content: str, ctx: Context, content_type: str | None = None
+):
     """Write content to a file in NextCloud.
-    
+
     Args:
         path: Full path where to write the file
         content: File content (text or base64 for binary)
         content_type: MIME type (auto-detected if not provided, use 'type;base64' for binary)
-        
+
     Returns:
         Dict with status_code indicating success
-        
+
     Examples:
         # Write a text file
         await nc_webdav_write_file("Documents/notes.md", "# My Notes\nContent here...")
-        
+
         # Write binary data (base64 encoded)
         await nc_webdav_write_file("files/data.bin", base64_content, "application/octet-stream;base64")
     """
     client: NextcloudClient = ctx.request_context.lifespan_context.client
-    
+
     # Handle base64 encoded content
     if content_type and "base64" in content_type.lower():
         import base64
+
         content_bytes = base64.b64decode(content)
         content_type = content_type.replace(";base64", "")
     else:
         content_bytes = content.encode("utf-8")
-    
+
     return await client.webdav.write_file(path, content_bytes, content_type)
 
 
 @mcp.tool()
 async def nc_webdav_create_directory(path: str, ctx: Context):
     """Create a directory in NextCloud.
-    
+
     Args:
         path: Full path of the directory to create
-        
+
     Returns:
         Dict with status_code (201 for created, 405 if already exists)
-        
+
     Examples:
         # Create a single directory
         await nc_webdav_create_directory("NewProject")
-        
+
         # Create nested directories (parent must exist)
         await nc_webdav_create_directory("Projects/MyApp/docs")
     """
@@ -316,17 +320,17 @@ async def nc_webdav_create_directory(path: str, ctx: Context):
 @mcp.tool()
 async def nc_webdav_delete_resource(path: str, ctx: Context):
     """Delete a file or directory in NextCloud.
-    
+
     Args:
         path: Full path of the file or directory to delete
-        
+
     Returns:
         Dict with status_code indicating result (404 if not found)
-        
+
     Examples:
         # Delete a file
         await nc_webdav_delete_resource("old_document.txt")
-        
+
         # Delete a directory (will delete all contents)
         await nc_webdav_delete_resource("temp_folder")
     """