ci: [skip ci] Move tests to subdirs
This commit is contained in:
Chris Coutinho
@@ -0,0 +1,404 @@
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from httpx import HTTPStatusError
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
# Note: nc_client fixture is session-scoped in conftest.py
|
||||
# Note: temporary_note and temporary_note_with_attachment fixtures are function-scoped in conftest.py
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mark all tests in this module as integration tests
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
async def test_attachments_add_and_get(
|
||||
nc_client: NextcloudClient, temporary_note_with_attachment: tuple
|
||||
):
|
||||
"""
|
||||
Tests adding an attachment (via fixture) and retrieving it.
|
||||
"""
|
||||
note_data, attachment_filename, attachment_content = temporary_note_with_attachment
|
||||
note_id = note_data["id"]
|
||||
note_category = note_data.get("category") # Get category from fixture data
|
||||
|
||||
logger.info(
|
||||
f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}"
|
||||
)
|
||||
# Pass category to get_note_attachment
|
||||
retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id, filename=attachment_filename, category=note_category
|
||||
)
|
||||
logger.info(
|
||||
f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes"
|
||||
)
|
||||
|
||||
assert retrieved_content == attachment_content
|
||||
assert "text/plain" in retrieved_mime # Fixture uses text/plain
|
||||
logger.info("Retrieved attachment content and mime type verified successfully.")
|
||||
|
||||
|
||||
async def test_attachments_add_to_note_with_category(
|
||||
nc_client: NextcloudClient, temporary_note: dict
|
||||
):
|
||||
"""
|
||||
Tests adding and retrieving an attachment specifically for a note that has a category.
|
||||
Uses temporary_note fixture and adds attachment manually within the test.
|
||||
"""
|
||||
note_data = (
|
||||
temporary_note # Note created by fixture (has category 'TemporaryTesting')
|
||||
)
|
||||
note_id = note_data["id"]
|
||||
note_category = note_data["category"]
|
||||
logger.info(
|
||||
f"Using note ID: {note_id} with category '{note_category}' for attachment test."
|
||||
)
|
||||
|
||||
# Add attachment within the test
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
attachment_filename = f"category_attach_{unique_suffix}.txt"
|
||||
attachment_content = f"Content for {attachment_filename}".encode("utf-8")
|
||||
attachment_mime = "text/plain"
|
||||
|
||||
logger.info(
|
||||
f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}"
|
||||
)
|
||||
# Pass category to add_note_attachment
|
||||
upload_response = await nc_client.webdav.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
category=note_category, # Pass the note's category
|
||||
mime_type=attachment_mime,
|
||||
)
|
||||
assert upload_response and "status_code" in upload_response
|
||||
assert upload_response["status_code"] in [201, 204]
|
||||
logger.info(
|
||||
f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']})."
|
||||
)
|
||||
time.sleep(1)
|
||||
|
||||
# Get and Verify Attachment
|
||||
logger.info(
|
||||
f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}"
|
||||
)
|
||||
# Pass category to get_note_attachment
|
||||
retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
category=note_category, # Pass the note's category
|
||||
)
|
||||
logger.info(
|
||||
f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes"
|
||||
)
|
||||
|
||||
assert retrieved_content == attachment_content
|
||||
assert attachment_mime in retrieved_mime
|
||||
logger.info(
|
||||
"Retrieved attachment content and mime type verified successfully for note with category."
|
||||
)
|
||||
# Cleanup is handled by the temporary_note fixture
|
||||
|
||||
|
||||
async def test_attachments_cleanup_on_note_delete(
|
||||
nc_client: NextcloudClient, temporary_note_with_attachment: tuple
|
||||
):
|
||||
"""
|
||||
Tests that the attachment (and its directory) are deleted when the parent note is deleted.
|
||||
Relies on the cleanup mechanism within notes_delete_note and the temporary_note fixture.
|
||||
"""
|
||||
note_data, attachment_filename, _ = temporary_note_with_attachment
|
||||
note_id = note_data["id"]
|
||||
note_category = note_data.get("category") # Get category from fixture data
|
||||
|
||||
# Fixture setup already added the attachment.
|
||||
# Fixture teardown (from temporary_note) will delete the note.
|
||||
# We just need to verify the attachment is gone *after* the test finishes
|
||||
# and the fixture cleanup runs. However, pytest fixtures don't easily allow
|
||||
# checking state *after* cleanup.
|
||||
# Instead, we will manually delete the note here and verify the attachment is gone.
|
||||
|
||||
logger.info(
|
||||
f"Attachment '{attachment_filename}' exists for note {note_id} (added by fixture)."
|
||||
)
|
||||
|
||||
# Manually delete the note
|
||||
logger.info(f"Manually deleting note ID: {note_id} within the test.")
|
||||
await nc_client.notes.delete_note(note_id=note_id)
|
||||
logger.info(f"Note ID: {note_id} deleted successfully.")
|
||||
time.sleep(1)
|
||||
|
||||
# Verify Note Is Deleted
|
||||
with pytest.raises(HTTPStatusError) as excinfo_note:
|
||||
await nc_client.notes.get_note(note_id=note_id)
|
||||
assert excinfo_note.value.response.status_code == 404
|
||||
logger.info(f"Verified note {note_id} deletion (404 received).")
|
||||
|
||||
# Verify Attachment Is Deleted (via 404 on GET)
|
||||
logger.info(
|
||||
f"Verifying attachment '{attachment_filename}' is deleted for note ID: {note_id}"
|
||||
)
|
||||
with pytest.raises(HTTPStatusError) as excinfo_attach:
|
||||
# Pass category to get_note_attachment - although it should fail anyway
|
||||
# because the note (and thus details) are gone.
|
||||
# The client method will raise 404 from the initial notes_get_note call.
|
||||
await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
category=note_category, # Pass category, though note fetch should fail first
|
||||
)
|
||||
# Expect 404 because the note itself is gone
|
||||
assert excinfo_attach.value.response.status_code == 404
|
||||
logger.info(
|
||||
f"Attachment '{attachment_filename}' correctly not found (404) after note deletion."
|
||||
)
|
||||
|
||||
# Directly verify attachment directory doesn't exist using WebDAV PROPFIND
|
||||
logger.info("Directly verifying attachment directory doesn't exist via PROPFIND")
|
||||
webdav_base = nc_client._get_webdav_base_path()
|
||||
category_path_part = f"{note_category}/" if note_category else ""
|
||||
attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{category_path_part}.attachments.{note_id}"
|
||||
)
|
||||
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
|
||||
try:
|
||||
propfind_resp = await nc_client._client.request(
|
||||
"PROPFIND", attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
status = propfind_resp.status_code
|
||||
if status in [200, 207]: # Successful PROPFIND means directory exists
|
||||
logger.error(
|
||||
f"Attachment directory still exists! PROPFIND returned {status}"
|
||||
)
|
||||
assert False, (
|
||||
f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
|
||||
)
|
||||
except HTTPStatusError as e:
|
||||
assert e.response.status_code == 404, (
|
||||
f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
|
||||
)
|
||||
logger.info(
|
||||
"Verified attachment directory does not exist via PROPFIND (404 received)"
|
||||
)
|
||||
|
||||
# Note: The temporary_note fixture will still run its cleanup,
|
||||
# but it will find the note already deleted (404) and handle it gracefully.
|
||||
|
||||
|
||||
async def test_attachments_category_change_handling(nc_client: NextcloudClient):
|
||||
"""
|
||||
Tests attachment handling when a note's category is changed.
|
||||
Verifies attachment retrieval works before and after category change,
|
||||
and that cleanup targets the correct final location.
|
||||
"""
|
||||
note_id = None
|
||||
initial_category = "CategoryA"
|
||||
new_category = "CategoryB"
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
note_title = f"Category Change Test {unique_suffix}"
|
||||
attachment_filename = f"cat_change_{unique_suffix}.txt"
|
||||
attachment_content = f"Content for {attachment_filename}".encode("utf-8")
|
||||
|
||||
try:
|
||||
# 1. Create note with initial category
|
||||
logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
|
||||
created_note = await nc_client.notes.create_note(
|
||||
title=note_title, content="Initial content", category=initial_category
|
||||
)
|
||||
note_id = created_note["id"]
|
||||
etag1 = created_note["etag"]
|
||||
logger.info(f"Note created with ID: {note_id}, Etag: {etag1}")
|
||||
time.sleep(1)
|
||||
|
||||
# 2. Add attachment (passing initial category)
|
||||
logger.info(
|
||||
f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
|
||||
)
|
||||
upload_response = await nc_client.webdav.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
category=initial_category,
|
||||
mime_type="text/plain",
|
||||
)
|
||||
assert upload_response["status_code"] in [201, 204]
|
||||
logger.info("Attachment added successfully.")
|
||||
time.sleep(1)
|
||||
|
||||
# 3. Verify attachment retrieval from initial category (passing initial category)
|
||||
logger.info(
|
||||
f"Verifying attachment retrieval from initial category '{initial_category}'"
|
||||
)
|
||||
retrieved_content1, _ = await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id, filename=attachment_filename, category=initial_category
|
||||
)
|
||||
assert retrieved_content1 == attachment_content
|
||||
logger.info("Attachment retrieved successfully from initial category.")
|
||||
|
||||
# 4. Update note category
|
||||
logger.info(
|
||||
f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
|
||||
)
|
||||
# Need to fetch the latest etag after attachment add (WebDAV ops don't update note etag)
|
||||
current_note_data = await nc_client.notes.get_note(note_id=note_id)
|
||||
current_etag = current_note_data["etag"]
|
||||
updated_note = await nc_client.notes.update(
|
||||
note_id=note_id,
|
||||
etag=current_etag,
|
||||
category=new_category,
|
||||
title=note_title,
|
||||
content="Updated content", # Pass required fields
|
||||
)
|
||||
etag3 = updated_note["etag"]
|
||||
assert updated_note["category"] == new_category
|
||||
logger.info(f"Note category updated successfully. New Etag: {etag3}")
|
||||
time.sleep(1)
|
||||
|
||||
# 5. Verify attachment retrieval from *new* category (passing new category)
|
||||
logger.info(
|
||||
f"Verifying attachment retrieval from new category '{new_category}'"
|
||||
)
|
||||
retrieved_content2, _ = await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id, filename=attachment_filename, category=new_category
|
||||
)
|
||||
assert retrieved_content2 == attachment_content
|
||||
logger.info("Attachment retrieved successfully from new category.")
|
||||
|
||||
# 5.1 Verify old category attachment directory is gone via WebDAV PROPFIND
|
||||
logger.info("Directly checking if old attachment directory exists in WebDAV")
|
||||
webdav_base = nc_client._get_webdav_base_path()
|
||||
old_attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
|
||||
)
|
||||
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
|
||||
try:
|
||||
propfind_resp = await nc_client._client.request(
|
||||
"PROPFIND", old_attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
status = propfind_resp.status_code
|
||||
if status in [200, 207]: # Successful PROPFIND means directory exists
|
||||
logger.error(
|
||||
f"Old attachment directory still exists! PROPFIND returned {status}"
|
||||
)
|
||||
assert False, (
|
||||
f"Expected old directory to be gone, but PROPFIND returned {status} - directory still exists!"
|
||||
)
|
||||
except HTTPStatusError as e:
|
||||
assert e.response.status_code == 404, (
|
||||
f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
|
||||
)
|
||||
logger.info(
|
||||
"Verified old attachment directory does not exist via PROPFIND (404 received)"
|
||||
)
|
||||
|
||||
# 5.2 Verify new category attachment directory exists via WebDAV PROPFIND
|
||||
logger.info("Directly checking if new attachment directory exists in WebDAV")
|
||||
new_attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
|
||||
)
|
||||
try:
|
||||
propfind_resp = await nc_client._client.request(
|
||||
"PROPFIND", new_attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
status = propfind_resp.status_code
|
||||
assert status in [
|
||||
207,
|
||||
200,
|
||||
], f"Expected PROPFIND to return success (207/200), got {status}"
|
||||
logger.info(
|
||||
f"Verified new attachment directory exists via PROPFIND ({status} received)"
|
||||
)
|
||||
except HTTPStatusError as e:
|
||||
logger.error(
|
||||
f"New attachment directory not found! PROPFIND failed with {e.response.status_code}"
|
||||
)
|
||||
assert False, (
|
||||
f"Expected new attachment directory to exist, but PROPFIND failed with {e.response.status_code}"
|
||||
)
|
||||
|
||||
finally:
|
||||
# 6. Cleanup: Delete the note (client should use the *final* category for cleanup path)
|
||||
if note_id:
|
||||
logger.info(
|
||||
f"Cleaning up note ID: {note_id} (last known category: '{new_category}')"
|
||||
)
|
||||
try:
|
||||
await nc_client.notes.delete_note(note_id=note_id)
|
||||
logger.info(f"Note {note_id} deleted.")
|
||||
time.sleep(1)
|
||||
# Verify note deletion
|
||||
with pytest.raises(HTTPStatusError) as excinfo_note_del:
|
||||
await nc_client.notes.get_note(note_id=note_id)
|
||||
assert excinfo_note_del.value.response.status_code == 404
|
||||
logger.info("Verified note deleted (404).")
|
||||
# Verify attachment deletion (should fail with 404 on the initial note fetch)
|
||||
with pytest.raises(HTTPStatusError) as excinfo_attach_del:
|
||||
# Pass the *last known* category, although the note fetch should fail first
|
||||
await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
category=new_category,
|
||||
)
|
||||
assert excinfo_attach_del.value.response.status_code == 404
|
||||
logger.info(
|
||||
"Verified attachment cannot be retrieved after note deletion (404)."
|
||||
)
|
||||
|
||||
# 6.1 Verify both old and new attachment directories are gone via WebDAV PROPFIND
|
||||
logger.info(
|
||||
"Directly verifying attachment directories don't exist via PROPFIND"
|
||||
)
|
||||
webdav_base = nc_client._get_webdav_base_path()
|
||||
|
||||
# Check new category attachment directory
|
||||
new_attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
|
||||
)
|
||||
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
|
||||
try:
|
||||
resp = await nc_client._client.request(
|
||||
"PROPFIND", new_attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
if resp.status_code in [
|
||||
200,
|
||||
207,
|
||||
]: # Successful PROPFIND means directory exists
|
||||
assert False, "New category attachment directory still exists!"
|
||||
except HTTPStatusError as e:
|
||||
assert e.response.status_code == 404, (
|
||||
f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
|
||||
)
|
||||
logger.info(
|
||||
"Verified new category attachment directory is gone via PROPFIND"
|
||||
)
|
||||
|
||||
# Check old category attachment directory
|
||||
old_attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
|
||||
)
|
||||
try:
|
||||
resp = await nc_client._client.request(
|
||||
"PROPFIND", old_attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
if resp.status_code in [
|
||||
200,
|
||||
207,
|
||||
]: # Successful PROPFIND means directory exists
|
||||
assert False, "Old category attachment directory still exists!"
|
||||
except HTTPStatusError as e:
|
||||
assert e.response.status_code == 404, (
|
||||
f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
|
||||
)
|
||||
logger.info(
|
||||
"Verified old category attachment directory is gone via PROPFIND"
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Verified all attachment directories are properly cleaned up."
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during cleanup for note {note_id}: {e}")
|
||||
@@ -0,0 +1,187 @@
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from io import BytesIO
|
||||
|
||||
import pytest
|
||||
from httpx import HTTPStatusError # Import if needed for specific error checks
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
# Note: nc_client fixture is session-scoped in conftest.py
|
||||
# Note: temporary_note fixture is function-scoped in conftest.py
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mark all tests in this module as integration tests
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
# Keep the test_image fixture as it's specific to generating image data
|
||||
@pytest.fixture(scope="module") # Keep module scope if image generation is slow
|
||||
def test_image_data() -> tuple[bytes, str]:
|
||||
"""
|
||||
Generate test image data (bytes) and suggest a filename.
|
||||
Returns (image_bytes, suggested_filename).
|
||||
"""
|
||||
logger.info("Generating test image data in memory.")
|
||||
img = Image.new("RGB", (300, 200), color=(255, 255, 255))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([(20, 20), (280, 180)], fill=(0, 120, 212)) # Blue rectangle
|
||||
draw.text(
|
||||
(50, 90), "Nextcloud Notes Test Image", fill=(255, 255, 255)
|
||||
) # White text
|
||||
|
||||
img_byte_arr = BytesIO()
|
||||
img.save(img_byte_arr, format="PNG")
|
||||
image_bytes = img_byte_arr.getvalue()
|
||||
suggested_filename = "test_image.png"
|
||||
logger.info(f"Generated test image data ({len(image_bytes)} bytes).")
|
||||
return image_bytes, suggested_filename
|
||||
|
||||
|
||||
async def test_note_with_embedded_image(
|
||||
nc_client: NextcloudClient, temporary_note: dict, test_image_data: tuple
|
||||
):
|
||||
"""
|
||||
Tests creating a note, attaching an image, embedding it in the content,
|
||||
and verifying the attachment can be retrieved.
|
||||
"""
|
||||
note_data = temporary_note # Use fixture for note creation/cleanup
|
||||
note_id = note_data["id"]
|
||||
note_etag = note_data["etag"]
|
||||
image_content, suggested_filename = test_image_data # Get image data from fixture
|
||||
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
attachment_filename = (
|
||||
f"test_image_{unique_suffix}.png" # Make filename unique per run
|
||||
)
|
||||
|
||||
# 1. Upload the image as an attachment
|
||||
note_category = note_data.get("category") # Get category from fixture data
|
||||
logger.info(
|
||||
f"Uploading image attachment '{attachment_filename}' to note {note_id} (category: '{note_category or ''}')..."
|
||||
)
|
||||
upload_response = await nc_client.webdav.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=image_content,
|
||||
category=note_category, # Pass the category
|
||||
mime_type="image/png",
|
||||
)
|
||||
assert upload_response and upload_response.get("status_code") in [201, 204]
|
||||
logger.info(
|
||||
f"Image uploaded successfully (Status: {upload_response.get('status_code')})."
|
||||
)
|
||||
time.sleep(1) # Allow potential processing time
|
||||
|
||||
# 1.1 Verify attachment directory exists via WebDAV PROPFIND
|
||||
logger.info("Directly checking if attachment directory exists in WebDAV")
|
||||
webdav_base = nc_client._get_webdav_base_path()
|
||||
category_path_part = f"{note_category}/" if note_category else ""
|
||||
attachment_dir_path = (
|
||||
f"{webdav_base}/Notes/{category_path_part}.attachments.{note_id}"
|
||||
)
|
||||
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
|
||||
try:
|
||||
propfind_resp = await nc_client._client.request(
|
||||
"PROPFIND", attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
status = propfind_resp.status_code
|
||||
assert status in [
|
||||
207,
|
||||
200,
|
||||
], f"Expected PROPFIND to return success (207/200), got {status}"
|
||||
logger.info(
|
||||
f"Verified attachment directory exists via PROPFIND ({status} received)"
|
||||
)
|
||||
except HTTPStatusError as e:
|
||||
logger.error(
|
||||
f"Attachment directory not found! PROPFIND failed with {e.response.status_code}"
|
||||
)
|
||||
assert False, (
|
||||
f"Expected attachment directory to exist, but PROPFIND failed with {e.response.status_code}"
|
||||
)
|
||||
|
||||
# 2. Update the note content to include the embedded image references
|
||||
updated_content = f"""{note_data["content"]}
|
||||
|
||||
## Image Embedding Test
|
||||
|
||||
### Markdown Syntax
|
||||
|
||||
|
||||
### HTML Syntax
|
||||
<img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="150" />
|
||||
"""
|
||||
logger.info("Updating note content with image references...")
|
||||
updated_note = await nc_client.notes.update(
|
||||
note_id=note_id,
|
||||
etag=note_etag, # Use etag from the created note
|
||||
content=updated_content,
|
||||
title=note_data["title"], # Pass required fields
|
||||
category=note_data["category"], # Pass required fields
|
||||
)
|
||||
new_etag = updated_note["etag"]
|
||||
assert new_etag != note_etag
|
||||
logger.info("Note content updated with image references.")
|
||||
time.sleep(1)
|
||||
|
||||
# 3. Verify the updated note content
|
||||
retrieved_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
assert f".attachments.{note_id}/{attachment_filename}" in retrieved_note["content"]
|
||||
logger.info("Verified image reference exists in updated note content.")
|
||||
|
||||
# 4. Verify the image attachment can be retrieved
|
||||
logger.info(
|
||||
f"Retrieving image attachment '{attachment_filename}' (category: '{note_category or ''}')..."
|
||||
)
|
||||
# Pass category to get_note_attachment
|
||||
retrieved_img_content, mime_type = await nc_client.webdav.get_note_attachment(
|
||||
note_id=note_id, filename=attachment_filename, category=note_category
|
||||
)
|
||||
assert retrieved_img_content == image_content
|
||||
assert mime_type.startswith("image/png")
|
||||
logger.info(
|
||||
"Successfully retrieved and verified image attachment content and mime type."
|
||||
)
|
||||
|
||||
# 5. Manually trigger deletion to verify cleanup (instead of waiting for fixture teardown)
|
||||
logger.info(
|
||||
f"Manually deleting note ID: {note_id} to verify proper attachment cleanup"
|
||||
)
|
||||
await nc_client.notes.delete_note(note_id=note_id)
|
||||
logger.info(f"Note ID: {note_id} deleted successfully.")
|
||||
time.sleep(1)
|
||||
|
||||
# 6. Verify note is deleted
|
||||
with pytest.raises(HTTPStatusError) as excinfo_note:
|
||||
await nc_client.notes.get_note(note_id=note_id)
|
||||
assert excinfo_note.value.response.status_code == 404
|
||||
logger.info(f"Verified note {note_id} deletion (404 received).")
|
||||
|
||||
# 7. Verify attachment directory is deleted via WebDAV PROPFIND
|
||||
logger.info("Directly verifying attachment directory doesn't exist via PROPFIND")
|
||||
try:
|
||||
propfind_resp = await nc_client._client.request(
|
||||
"PROPFIND", attachment_dir_path, headers=propfind_headers
|
||||
)
|
||||
status = propfind_resp.status_code
|
||||
if status in [200, 207]: # Successful PROPFIND means directory exists
|
||||
logger.error(
|
||||
f"Attachment directory still exists! PROPFIND returned {status}"
|
||||
)
|
||||
assert False, (
|
||||
f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
|
||||
)
|
||||
except HTTPStatusError as e:
|
||||
assert e.response.status_code == 404, (
|
||||
f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
|
||||
)
|
||||
logger.info(
|
||||
"Verified attachment directory does not exist via PROPFIND (404 received)"
|
||||
)
|
||||
|
||||
# Note: The temporary_note fixture will still run its cleanup,
|
||||
# but it will find the note already deleted (404) and handle it gracefully.
|
||||
@@ -0,0 +1,260 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import uuid # Keep uuid if needed for generating unique data within tests
|
||||
|
||||
import pytest
|
||||
from httpx import HTTPStatusError
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
# Note: nc_client fixture is now session-scoped in conftest.py
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Mark all tests in this module as integration tests
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
async def test_notes_api_create_and_read(
|
||||
nc_client: NextcloudClient, temporary_note: dict
|
||||
):
|
||||
"""
|
||||
Tests creating a note via the API (using fixture) and then reading it back.
|
||||
"""
|
||||
created_note_data = temporary_note # Get data from fixture
|
||||
note_id = created_note_data["id"]
|
||||
|
||||
logger.info(f"Reading note created by fixture, ID: {note_id}")
|
||||
read_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
|
||||
assert read_note["id"] == note_id
|
||||
assert read_note["title"] == created_note_data["title"]
|
||||
assert read_note["content"] == created_note_data["content"]
|
||||
assert read_note["category"] == created_note_data["category"]
|
||||
logger.info(f"Successfully read and verified note ID: {note_id}")
|
||||
|
||||
|
||||
async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict):
|
||||
"""
|
||||
Tests updating a note created by the fixture.
|
||||
"""
|
||||
created_note_data = temporary_note
|
||||
note_id = created_note_data["id"]
|
||||
original_etag = created_note_data["etag"]
|
||||
original_category = created_note_data["category"]
|
||||
|
||||
update_title = f"Updated Title {uuid.uuid4().hex[:8]}"
|
||||
update_content = f"Updated Content {uuid.uuid4().hex[:8]}"
|
||||
|
||||
logger.info(f"Attempting to update note ID: {note_id} with etag: {original_etag}")
|
||||
updated_note = await nc_client.notes.update(
|
||||
note_id=note_id,
|
||||
etag=original_etag,
|
||||
title=update_title,
|
||||
content=update_content,
|
||||
# category=original_category # Explicitly pass category if required by update
|
||||
)
|
||||
logger.info(f"Note updated: {updated_note}")
|
||||
|
||||
assert updated_note["id"] == note_id
|
||||
assert updated_note["title"] == update_title
|
||||
assert updated_note["content"] == update_content
|
||||
assert (
|
||||
updated_note["category"] == original_category
|
||||
) # Verify category didn't change
|
||||
assert "etag" in updated_note
|
||||
assert updated_note["etag"] != original_etag # Etag must change
|
||||
|
||||
# Optional: Verify update by reading again
|
||||
await asyncio.sleep(1) # Allow potential propagation delay
|
||||
read_updated_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
assert read_updated_note["title"] == update_title
|
||||
assert read_updated_note["content"] == update_content
|
||||
logger.info(f"Successfully updated and verified note ID: {note_id}")
|
||||
|
||||
|
||||
async def test_notes_api_update_conflict(
|
||||
nc_client: NextcloudClient, temporary_note: dict
|
||||
):
|
||||
"""
|
||||
Tests that attempting to update with an old etag fails with 412.
|
||||
"""
|
||||
created_note_data = temporary_note
|
||||
note_id = created_note_data["id"]
|
||||
original_etag = created_note_data["etag"]
|
||||
|
||||
# Perform a first update to change the etag
|
||||
first_update_title = f"First Update {uuid.uuid4().hex[:8]}"
|
||||
logger.info(f"Performing first update on note ID: {note_id} to change etag.")
|
||||
first_updated_note = await nc_client.notes.update(
|
||||
note_id=note_id,
|
||||
etag=original_etag,
|
||||
title=first_update_title,
|
||||
content="First update content",
|
||||
# category=created_note_data["category"] # Pass category if required
|
||||
)
|
||||
new_etag = first_updated_note["etag"]
|
||||
assert new_etag != original_etag
|
||||
logger.info(f"Note ID: {note_id} updated, new etag: {new_etag}")
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Now attempt update with the *original* etag
|
||||
logger.info(
|
||||
f"Attempting second update on note ID: {note_id} with OLD etag: {original_etag}"
|
||||
)
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
await nc_client.notes.update(
|
||||
note_id=note_id,
|
||||
etag=original_etag, # Use the stale etag
|
||||
title="This update should fail due to conflict",
|
||||
# category=created_note_data["category"] # Pass category if required
|
||||
)
|
||||
assert excinfo.value.response.status_code == 412 # Precondition Failed
|
||||
logger.info("Update with old etag correctly failed with 412 Precondition Failed.")
|
||||
|
||||
|
||||
async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
|
||||
"""
|
||||
Tests deleting a note that doesn't exist fails with 404.
|
||||
"""
|
||||
non_existent_id = 999999999 # Use an ID highly unlikely to exist
|
||||
logger.info(f"\nAttempting to delete non-existent note ID: {non_existent_id}")
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
await nc_client.notes.delete_note(note_id=non_existent_id)
|
||||
assert excinfo.value.response.status_code == 404
|
||||
logger.info(
|
||||
f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404."
|
||||
)
|
||||
|
||||
|
||||
async def test_notes_api_append_content_to_existing_note(
|
||||
nc_client: NextcloudClient, temporary_note: dict
|
||||
):
|
||||
"""
|
||||
Tests appending content to an existing note using the new append functionality.
|
||||
"""
|
||||
created_note_data = temporary_note
|
||||
note_id = created_note_data["id"]
|
||||
original_content = created_note_data["content"]
|
||||
|
||||
append_text = f"Appended content {uuid.uuid4().hex[:8]}"
|
||||
|
||||
logger.info(f"Appending content to note ID: {note_id}")
|
||||
updated_note = await nc_client.notes.append_content(
|
||||
note_id=note_id, content=append_text
|
||||
)
|
||||
logger.info(f"Note after append: {updated_note}")
|
||||
|
||||
# Verify the note was updated
|
||||
assert updated_note["id"] == note_id
|
||||
assert "etag" in updated_note
|
||||
assert updated_note["etag"] != created_note_data["etag"] # Etag must change
|
||||
|
||||
# Verify content has the separator and appended text
|
||||
expected_content = original_content + "\n---\n" + append_text
|
||||
assert updated_note["content"] == expected_content
|
||||
|
||||
# Verify by reading the note again
|
||||
await asyncio.sleep(1) # Allow potential propagation delay
|
||||
read_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
assert read_note["content"] == expected_content
|
||||
logger.info(f"Successfully appended content to note ID: {note_id}")
|
||||
|
||||
|
||||
async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient):
|
||||
"""
|
||||
Tests appending content to an empty note (no separator should be added).
|
||||
"""
|
||||
# Create an empty note
|
||||
test_title = f"Empty Note {uuid.uuid4().hex[:8]}"
|
||||
test_category = "Test"
|
||||
|
||||
logger.info("Creating empty note for append test")
|
||||
empty_note = await nc_client.notes.create_note(
|
||||
title=test_title,
|
||||
content="",
|
||||
category=test_category, # Empty content
|
||||
)
|
||||
note_id = empty_note["id"]
|
||||
|
||||
try:
|
||||
append_text = f"First content {uuid.uuid4().hex[:8]}"
|
||||
|
||||
logger.info(f"Appending content to empty note ID: {note_id}")
|
||||
updated_note = await nc_client.notes.append_content(
|
||||
note_id=note_id, content=append_text
|
||||
)
|
||||
|
||||
# For empty notes, content should just be the appended text (no separator)
|
||||
assert updated_note["content"] == append_text
|
||||
|
||||
# Verify by reading the note again
|
||||
await asyncio.sleep(1)
|
||||
read_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
assert read_note["content"] == append_text
|
||||
logger.info(f"Successfully appended content to empty note ID: {note_id}")
|
||||
|
||||
finally:
|
||||
# Clean up the test note
|
||||
try:
|
||||
await nc_client.notes.delete_note(note_id=note_id)
|
||||
logger.info(f"Cleaned up test note ID: {note_id}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to clean up test note ID: {note_id}: {e}")
|
||||
|
||||
|
||||
async def test_notes_api_append_content_multiple_times(
|
||||
nc_client: NextcloudClient, temporary_note: dict
|
||||
):
|
||||
"""
|
||||
Tests appending content multiple times to verify separator behavior.
|
||||
"""
|
||||
created_note_data = temporary_note
|
||||
note_id = created_note_data["id"]
|
||||
original_content = created_note_data["content"]
|
||||
|
||||
first_append = f"First append {uuid.uuid4().hex[:8]}"
|
||||
second_append = f"Second append {uuid.uuid4().hex[:8]}"
|
||||
|
||||
logger.info(f"Performing multiple appends to note ID: {note_id}")
|
||||
|
||||
# First append
|
||||
updated_note = await nc_client.notes.append_content(
|
||||
note_id=note_id, content=first_append
|
||||
)
|
||||
|
||||
expected_content_after_first = original_content + "\n---\n" + first_append
|
||||
assert updated_note["content"] == expected_content_after_first
|
||||
|
||||
# Second append
|
||||
updated_note = await nc_client.notes.append_content(
|
||||
note_id=note_id, content=second_append
|
||||
)
|
||||
|
||||
expected_content_after_second = (
|
||||
expected_content_after_first + "\n---\n" + second_append
|
||||
)
|
||||
assert updated_note["content"] == expected_content_after_second
|
||||
|
||||
# Verify by reading the note again
|
||||
await asyncio.sleep(1)
|
||||
read_note = await nc_client.notes.get_note(note_id=note_id)
|
||||
assert read_note["content"] == expected_content_after_second
|
||||
logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
|
||||
|
||||
|
||||
async def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient):
|
||||
"""
|
||||
Tests that appending to a non-existent note fails with 404.
|
||||
"""
|
||||
non_existent_id = 999999999
|
||||
|
||||
logger.info(f"Attempting to append to non-existent note ID: {non_existent_id}")
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
await nc_client.notes.append_content(
|
||||
note_id=non_existent_id, content="This should fail"
|
||||
)
|
||||
assert excinfo.value.response.status_code == 404
|
||||
logger.info(
|
||||
f"Appending to non-existent note ID: {non_existent_id} correctly failed with 404."
|
||||
)
|
||||
Reference in New Issue
Block a user