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 <noreply@anthropic.com>
This commit is contained in:
Neovasky
@@ -249,10 +249,10 @@ class WebDAVClient(BaseNextcloudClient):
|
||||
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 @@ class WebDAVClient(BaseNextcloudClient):
|
||||
<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,19 +345,23 @@ class WebDAVClient(BaseNextcloudClient):
|
||||
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
|
||||
@@ -358,29 +369,32 @@ class WebDAVClient(BaseNextcloudClient):
|
||||
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 @@ class WebDAVClient(BaseNextcloudClient):
|
||||
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}")
|
||||
|
||||
@@ -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")
|
||||
"""
|
||||
|
||||
@@ -0,0 +1,269 @@
|
||||
"""Integration tests for WebDAV operations."""
|
||||
|
||||
import pytest
|
||||
import logging
|
||||
import uuid
|
||||
from httpx import HTTPStatusError
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mark all tests in this module as integration tests
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def test_base_path():
|
||||
"""Base path for test files/directories."""
|
||||
return f"mcp_test_{uuid.uuid4().hex[:8]}"
|
||||
|
||||
|
||||
async def test_create_and_delete_directory(nc_client: NextcloudClient, test_base_path: str):
|
||||
"""Test creating and deleting directories."""
|
||||
test_dir = f"{test_base_path}/test_directory"
|
||||
|
||||
try:
|
||||
# Create directory
|
||||
result = await nc_client.webdav.create_directory(test_dir)
|
||||
assert result["status_code"] == 201 # Created
|
||||
logger.info(f"Created directory: {test_dir}")
|
||||
|
||||
# Verify directory exists by listing parent
|
||||
parent_listing = await nc_client.webdav.list_directory(test_base_path)
|
||||
dir_names = [item["name"] for item in parent_listing]
|
||||
assert "test_directory" in dir_names
|
||||
|
||||
# Delete directory
|
||||
delete_result = await nc_client.webdav.delete_resource(test_dir)
|
||||
assert delete_result["status_code"] in [204, 404] # No Content or Not Found
|
||||
logger.info(f"Deleted directory: {test_dir}")
|
||||
|
||||
finally:
|
||||
# Cleanup: ensure directory is deleted
|
||||
try:
|
||||
await nc_client.webdav.delete_resource(test_dir)
|
||||
await nc_client.webdav.delete_resource(test_base_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_write_read_delete_file(nc_client: NextcloudClient, test_base_path: str):
|
||||
"""Test writing, reading, and deleting files."""
|
||||
test_file = f"{test_base_path}/test_file.txt"
|
||||
test_content = f"Test content {uuid.uuid4().hex}"
|
||||
|
||||
try:
|
||||
# Create base directory first
|
||||
await nc_client.webdav.create_directory(test_base_path)
|
||||
|
||||
# Write file
|
||||
write_result = await nc_client.webdav.write_file(
|
||||
test_file,
|
||||
test_content.encode('utf-8'),
|
||||
content_type="text/plain"
|
||||
)
|
||||
assert write_result["status_code"] in [200, 201, 204] # Success codes
|
||||
logger.info(f"Wrote file: {test_file}")
|
||||
|
||||
# Read file back
|
||||
content, content_type = await nc_client.webdav.read_file(test_file)
|
||||
assert content.decode('utf-8') == test_content
|
||||
assert content_type == "text/plain"
|
||||
logger.info(f"Read file: {test_file}")
|
||||
|
||||
# Verify file appears in directory listing
|
||||
listing = await nc_client.webdav.list_directory(test_base_path)
|
||||
file_names = [item["name"] for item in listing]
|
||||
assert "test_file.txt" in file_names
|
||||
|
||||
# Delete file
|
||||
delete_result = await nc_client.webdav.delete_resource(test_file)
|
||||
assert delete_result["status_code"] in [204, 404] # No Content or Not Found
|
||||
logger.info(f"Deleted file: {test_file}")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
try:
|
||||
await nc_client.webdav.delete_resource(test_file)
|
||||
await nc_client.webdav.delete_resource(test_base_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_list_directory_empty_and_populated(nc_client: NextcloudClient, test_base_path: str):
|
||||
"""Test listing empty and populated directories."""
|
||||
try:
|
||||
# Create base directory
|
||||
await nc_client.webdav.create_directory(test_base_path)
|
||||
|
||||
# List empty directory
|
||||
empty_listing = await nc_client.webdav.list_directory(test_base_path)
|
||||
assert isinstance(empty_listing, list)
|
||||
assert len(empty_listing) == 0
|
||||
logger.info(f"Empty directory listing: {len(empty_listing)} items")
|
||||
|
||||
# Add some files and directories
|
||||
await nc_client.webdav.create_directory(f"{test_base_path}/subdir1")
|
||||
await nc_client.webdav.create_directory(f"{test_base_path}/subdir2")
|
||||
await nc_client.webdav.write_file(
|
||||
f"{test_base_path}/file1.txt",
|
||||
b"content1",
|
||||
content_type="text/plain"
|
||||
)
|
||||
await nc_client.webdav.write_file(
|
||||
f"{test_base_path}/file2.md",
|
||||
b"# Markdown content",
|
||||
content_type="text/markdown"
|
||||
)
|
||||
|
||||
# List populated directory
|
||||
populated_listing = await nc_client.webdav.list_directory(test_base_path)
|
||||
assert len(populated_listing) == 4 # 2 dirs + 2 files
|
||||
|
||||
# Check that we have both files and directories
|
||||
names = [item["name"] for item in populated_listing]
|
||||
assert "subdir1" in names
|
||||
assert "subdir2" in names
|
||||
assert "file1.txt" in names
|
||||
assert "file2.md" in names
|
||||
|
||||
# Check metadata is present
|
||||
for item in populated_listing:
|
||||
assert "name" in item
|
||||
assert "path" in item
|
||||
assert "is_directory" in item
|
||||
assert "size" in item
|
||||
assert "content_type" in item
|
||||
assert "last_modified" in item
|
||||
|
||||
logger.info(f"Populated directory listing: {len(populated_listing)} items")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
try:
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/file1.txt")
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/file2.md")
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/subdir1")
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/subdir2")
|
||||
await nc_client.webdav.delete_resource(test_base_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_read_nonexistent_file(nc_client: NextcloudClient):
|
||||
"""Test reading a file that doesn't exist."""
|
||||
nonexistent_file = f"nonexistent_{uuid.uuid4().hex}.txt"
|
||||
|
||||
with pytest.raises(HTTPStatusError) as exc_info:
|
||||
await nc_client.webdav.read_file(nonexistent_file)
|
||||
|
||||
assert exc_info.value.response.status_code == 404
|
||||
logger.info(f"Correctly got 404 for nonexistent file: {nonexistent_file}")
|
||||
|
||||
|
||||
async def test_delete_nonexistent_resource(nc_client: NextcloudClient):
|
||||
"""Test deleting a resource that doesn't exist."""
|
||||
nonexistent_resource = f"nonexistent_{uuid.uuid4().hex}"
|
||||
|
||||
result = await nc_client.webdav.delete_resource(nonexistent_resource)
|
||||
assert result["status_code"] == 404
|
||||
logger.info(f"Correctly got 404 for nonexistent resource: {nonexistent_resource}")
|
||||
|
||||
|
||||
async def test_create_nested_directories(nc_client: NextcloudClient, test_base_path: str):
|
||||
"""Test creating nested directory structures."""
|
||||
nested_path = f"{test_base_path}/level1/level2/level3"
|
||||
|
||||
try:
|
||||
# Create nested directories (should create parent directories automatically)
|
||||
result = await nc_client.webdav.create_directory(nested_path)
|
||||
assert result["status_code"] == 201
|
||||
|
||||
# Verify the structure was created
|
||||
level1_listing = await nc_client.webdav.list_directory(f"{test_base_path}/level1")
|
||||
assert len(level1_listing) == 1
|
||||
assert level1_listing[0]["name"] == "level2"
|
||||
assert level1_listing[0]["is_directory"] is True
|
||||
|
||||
level2_listing = await nc_client.webdav.list_directory(f"{test_base_path}/level1/level2")
|
||||
assert len(level2_listing) == 1
|
||||
assert level2_listing[0]["name"] == "level3"
|
||||
assert level2_listing[0]["is_directory"] is True
|
||||
|
||||
logger.info(f"Created nested directory structure: {nested_path}")
|
||||
|
||||
finally:
|
||||
# Cleanup - delete from deepest to shallowest
|
||||
try:
|
||||
await nc_client.webdav.delete_resource(nested_path)
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/level1/level2")
|
||||
await nc_client.webdav.delete_resource(f"{test_base_path}/level1")
|
||||
await nc_client.webdav.delete_resource(test_base_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_overwrite_existing_file(nc_client: NextcloudClient, test_base_path: str):
|
||||
"""Test overwriting an existing file."""
|
||||
test_file = f"{test_base_path}/overwrite_test.txt"
|
||||
original_content = "Original content"
|
||||
new_content = "New content after overwrite"
|
||||
|
||||
try:
|
||||
# Create base directory
|
||||
await nc_client.webdav.create_directory(test_base_path)
|
||||
|
||||
# Write original file
|
||||
await nc_client.webdav.write_file(
|
||||
test_file,
|
||||
original_content.encode('utf-8'),
|
||||
content_type="text/plain"
|
||||
)
|
||||
|
||||
# Verify original content
|
||||
content, _ = await nc_client.webdav.read_file(test_file)
|
||||
assert content.decode('utf-8') == original_content
|
||||
|
||||
# Overwrite with new content
|
||||
overwrite_result = await nc_client.webdav.write_file(
|
||||
test_file,
|
||||
new_content.encode('utf-8'),
|
||||
content_type="text/plain"
|
||||
)
|
||||
assert overwrite_result["status_code"] in [200, 204] # OK or No Content
|
||||
|
||||
# Verify new content
|
||||
content, _ = await nc_client.webdav.read_file(test_file)
|
||||
assert content.decode('utf-8') == new_content
|
||||
|
||||
logger.info(f"Successfully overwrote file: {test_file}")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
try:
|
||||
await nc_client.webdav.delete_resource(test_file)
|
||||
await nc_client.webdav.delete_resource(test_base_path)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
async def test_list_root_directory(nc_client: NextcloudClient):
|
||||
"""Test listing the root directory."""
|
||||
root_listing = await nc_client.webdav.list_directory("")
|
||||
|
||||
# Root directory should exist and be listable
|
||||
assert isinstance(root_listing, list)
|
||||
# Should have at least some default folders/files
|
||||
assert len(root_listing) >= 0
|
||||
|
||||
# Check structure of items
|
||||
for item in root_listing:
|
||||
assert "name" in item
|
||||
assert "path" in item
|
||||
assert "is_directory" in item
|
||||
assert "size" in item
|
||||
assert "content_type" in item
|
||||
assert "last_modified" in item
|
||||
|
||||
logger.info(f"Root directory contains {len(root_listing)} items")
|
||||
Reference in New Issue
Block a user