Refactor
This commit is contained in:
Chris Coutinho
@@ -155,7 +155,7 @@ class NextcloudClient:
|
||||
"modified": note.get("modified"),
|
||||
}
|
||||
)
|
||||
raise e
|
||||
return search_results
|
||||
|
||||
def delete_webdav_resource(self, *, path: str):
|
||||
"""Delete a resource (file or directory) via WebDAV DELETE."""
|
||||
|
||||
+91
-2
@@ -1,7 +1,9 @@
|
||||
import pytest
|
||||
import os
|
||||
import logging
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
import uuid
|
||||
import time
|
||||
from nextcloud_mcp_server.client import NextcloudClient, HTTPStatusError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -9,8 +11,95 @@ logger = logging.getLogger(__name__)
|
||||
def nc_client() -> NextcloudClient:
|
||||
"""
|
||||
Fixture to create a NextcloudClient instance for integration tests.
|
||||
Uses environment variables for configuration.
|
||||
"""
|
||||
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
|
||||
assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
|
||||
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
|
||||
return NextcloudClient.from_env()
|
||||
logger.info("Creating session-scoped NextcloudClient from environment variables.")
|
||||
client = NextcloudClient.from_env()
|
||||
# Optional: Perform a quick check like getting capabilities to ensure connection works
|
||||
try:
|
||||
client.capabilities()
|
||||
logger.info("NextcloudClient session fixture initialized and capabilities checked.")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize NextcloudClient session fixture: {e}")
|
||||
pytest.fail(f"Failed to connect to Nextcloud or get capabilities: {e}")
|
||||
return client
|
||||
|
||||
@pytest.fixture
|
||||
def temporary_note(nc_client: NextcloudClient):
|
||||
"""
|
||||
Fixture to create a temporary note for a test and ensure its deletion afterward.
|
||||
Yields the created note dictionary.
|
||||
"""
|
||||
note_id = None
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
note_title = f"Temporary Test Note {unique_suffix}"
|
||||
note_content = f"Content for temporary note {unique_suffix}"
|
||||
note_category = "TemporaryTesting"
|
||||
created_note_data = None
|
||||
|
||||
logger.info(f"Creating temporary note: {note_title}")
|
||||
try:
|
||||
created_note_data = nc_client.notes_create_note(
|
||||
title=note_title, content=note_content, category=note_category
|
||||
)
|
||||
note_id = created_note_data.get("id")
|
||||
if not note_id:
|
||||
pytest.fail("Failed to get ID from created temporary note.")
|
||||
|
||||
logger.info(f"Temporary note created with ID: {note_id}")
|
||||
yield created_note_data # Provide the created note data to the test
|
||||
|
||||
finally:
|
||||
if note_id:
|
||||
logger.info(f"Cleaning up temporary note ID: {note_id}")
|
||||
try:
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
logger.info(f"Successfully deleted temporary note ID: {note_id}")
|
||||
except HTTPStatusError as e:
|
||||
# Ignore 404 if note was already deleted by the test itself
|
||||
if e.response.status_code != 404:
|
||||
logger.error(f"HTTP error deleting temporary note {note_id}: {e}")
|
||||
else:
|
||||
logger.warning(f"Temporary note {note_id} already deleted (404).")
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error deleting temporary note {note_id}: {e}")
|
||||
|
||||
@pytest.fixture
|
||||
def temporary_note_with_attachment(nc_client: NextcloudClient, temporary_note: dict):
|
||||
"""
|
||||
Fixture that creates a temporary note, adds an attachment, and cleans up both.
|
||||
Yields a tuple: (note_data, attachment_filename, attachment_content).
|
||||
Depends on the temporary_note fixture.
|
||||
"""
|
||||
note_data = temporary_note
|
||||
note_id = note_data["id"]
|
||||
unique_suffix = uuid.uuid4().hex[:8]
|
||||
attachment_filename = f"temp_attach_{unique_suffix}.txt"
|
||||
attachment_content = f"Content for {attachment_filename}".encode('utf-8')
|
||||
attachment_mime = "text/plain"
|
||||
|
||||
logger.info(f"Adding attachment '{attachment_filename}' to temporary note ID: {note_id}")
|
||||
try:
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
mime_type=attachment_mime
|
||||
)
|
||||
assert upload_response.get("status_code") in [201, 204], f"Failed to upload attachment: {upload_response}"
|
||||
logger.info(f"Attachment '{attachment_filename}' added successfully.")
|
||||
|
||||
yield note_data, attachment_filename, attachment_content
|
||||
|
||||
# Cleanup for the attachment is handled by the notes_delete_note call
|
||||
# in the temporary_note fixture's finally block (which deletes the .attachments dir)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to add attachment in fixture: {e}")
|
||||
pytest.fail(f"Fixture setup failed during attachment upload: {e}")
|
||||
|
||||
# Note: The temporary_note fixture's finally block will handle note deletion,
|
||||
# which should also trigger the WebDAV directory deletion attempt.
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
import pytest
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
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
|
||||
|
||||
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"]
|
||||
|
||||
logger.info(f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}")
|
||||
retrieved_content, retrieved_mime = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
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.")
|
||||
|
||||
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}")
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
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}")
|
||||
retrieved_content, retrieved_mime = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
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
|
||||
|
||||
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"]
|
||||
|
||||
# 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.")
|
||||
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:
|
||||
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:
|
||||
nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
# Expect 404 because the parent directory (.attachments.NOTE_ID) should be gone
|
||||
assert excinfo_attach.value.response.status_code == 404
|
||||
logger.info(f"Attachment '{attachment_filename}' correctly not found (404) after note deletion.")
|
||||
|
||||
# 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,106 @@
|
||||
import pytest
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
import logging
|
||||
import tempfile
|
||||
from PIL import Image, ImageDraw
|
||||
from io import BytesIO
|
||||
from httpx import HTTPStatusError # Import if needed for specific error checks
|
||||
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
logger.info(f"Uploading image attachment '{attachment_filename}' to note {note_id}...")
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=image_content,
|
||||
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
|
||||
|
||||
# 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 = nc_client.notes_update_note(
|
||||
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 = 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}'...")
|
||||
retrieved_img_content, mime_type = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
assert retrieved_img_content == image_content
|
||||
assert mime_type.startswith("image/png")
|
||||
logger.info("Successfully retrieved and verified image attachment content and mime type.")
|
||||
|
||||
# Note cleanup is handled by the temporary_note fixture
|
||||
@@ -0,0 +1,114 @@
|
||||
import pytest
|
||||
import logging
|
||||
import time
|
||||
import uuid # Keep uuid if needed for generating unique data within tests
|
||||
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
|
||||
|
||||
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 = 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}")
|
||||
|
||||
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 = nc_client.notes_update_note(
|
||||
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
|
||||
time.sleep(1) # Allow potential propagation delay
|
||||
read_updated_note = 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}")
|
||||
|
||||
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 = nc_client.notes_update_note(
|
||||
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}")
|
||||
time.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:
|
||||
nc_client.notes_update_note(
|
||||
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.")
|
||||
|
||||
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:
|
||||
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.")
|
||||
|
||||
# --- Attachment tests moved to test_attachments.py ---
|
||||
@@ -1,394 +0,0 @@
|
||||
import pytest
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
from httpx import HTTPStatusError
|
||||
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
# Tests assume NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars are set
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def nc_client() -> NextcloudClient:
|
||||
"""
|
||||
Fixture to create a NextcloudClient instance for integration tests.
|
||||
Reads credentials from environment variables.
|
||||
Scope is 'module' so the client is reused for all tests in this file.
|
||||
"""
|
||||
# Basic check to ensure env vars seem present - tests will fail properly if not
|
||||
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
|
||||
assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
|
||||
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
|
||||
return NextcloudClient.from_env()
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_note_crud_integration(nc_client: NextcloudClient):
|
||||
"""
|
||||
Integration test for the complete CRUD (Create, Read, Update, Delete)
|
||||
lifecycle of a note.
|
||||
"""
|
||||
# --- Create ---
|
||||
unique_id = str(uuid.uuid4()) # To ensure note is unique for this test run
|
||||
create_title = f"Integration Test Note {unique_id}"
|
||||
create_content = f"Content for integration test {unique_id}"
|
||||
create_category = "IntegrationTesting"
|
||||
|
||||
created_note = (
|
||||
None # Initialize to ensure cleanup happens even if create fails mid-assert
|
||||
)
|
||||
try:
|
||||
logger.info(f"\nAttempting to create note: {create_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=create_title, content=create_content, category=create_category
|
||||
)
|
||||
logger.info(f"Note created: {created_note}")
|
||||
|
||||
assert created_note is not None
|
||||
assert "id" in created_note
|
||||
assert created_note["title"] == create_title
|
||||
assert created_note["content"] == create_content
|
||||
assert created_note["category"] == create_category
|
||||
assert "etag" in created_note
|
||||
note_id = created_note["id"]
|
||||
etag = created_note["etag"]
|
||||
|
||||
# Add a small delay to allow Nextcloud to process if needed
|
||||
time.sleep(1)
|
||||
|
||||
# --- Read (Verify Create) ---
|
||||
logger.info(f"Attempting to read note ID: {note_id}")
|
||||
read_note = nc_client.notes_get_note(note_id=note_id)
|
||||
logger.info(f"Note read: {read_note}")
|
||||
assert read_note["id"] == note_id
|
||||
assert read_note["title"] == create_title
|
||||
assert read_note["content"] == create_content
|
||||
assert read_note["category"] == create_category
|
||||
# Etag might change even on read in some systems, so don't assert etag equality here
|
||||
|
||||
# --- Update ---
|
||||
update_title = f"Updated Test Note {unique_id}"
|
||||
update_content = f"Updated content {unique_id}"
|
||||
# Use the etag from the *creation* for the update's If-Match header
|
||||
logger.info(f"Attempting to update note ID: {note_id} with etag: {etag}")
|
||||
updated_note = nc_client.notes_update_note(
|
||||
note_id=note_id,
|
||||
etag=etag,
|
||||
title=update_title,
|
||||
content=update_content,
|
||||
# category=create_category # Keep category same or update if needed
|
||||
)
|
||||
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"] == create_category # Category wasn't updated
|
||||
assert "etag" in updated_note
|
||||
assert updated_note["etag"] != etag # Etag must change on update
|
||||
new_etag = updated_note["etag"]
|
||||
|
||||
# Add a small delay
|
||||
time.sleep(1)
|
||||
|
||||
# --- Read (Verify Update) ---
|
||||
logger.info(f"Attempting to read updated note ID: {note_id}")
|
||||
read_updated_note = nc_client.notes_get_note(note_id=note_id)
|
||||
logger.info(f"Updated note read: {read_updated_note}")
|
||||
assert read_updated_note["id"] == note_id
|
||||
assert read_updated_note["title"] == update_title
|
||||
assert read_updated_note["content"] == update_content
|
||||
# Don't assert etag equality here either
|
||||
|
||||
# --- Test Update Conflict (Precondition Failed) ---
|
||||
logger.info(f"Attempting to update note ID: {note_id} with OLD etag: {etag}")
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
nc_client.notes_update_note(
|
||||
note_id=note_id,
|
||||
etag=etag, # Use the OLD etag
|
||||
title="This update should fail",
|
||||
)
|
||||
assert excinfo.value.response.status_code == 412 # Precondition Failed
|
||||
logger.info("Update with old etag correctly failed with 412.")
|
||||
|
||||
finally:
|
||||
# --- Delete ---
|
||||
if created_note and "id" in created_note:
|
||||
note_id_to_delete = created_note["id"]
|
||||
logger.info(f"Attempting to delete note ID: {note_id_to_delete}")
|
||||
try:
|
||||
delete_response = nc_client.notes_delete_note(note_id=note_id_to_delete)
|
||||
logger.info(f"Delete response: {delete_response}")
|
||||
# Check if delete returns the deleted object or just status
|
||||
# Assuming it returns the object based on previous tests
|
||||
assert delete_response["id"] == note_id_to_delete
|
||||
logger.info(f"Note ID: {note_id_to_delete} deleted successfully.")
|
||||
|
||||
# --- Verify Delete ---
|
||||
logger.info(f"Attempting to read deleted note ID: {note_id_to_delete}")
|
||||
with pytest.raises(HTTPStatusError) as excinfo_del:
|
||||
nc_client.notes_get_note(note_id=note_id_to_delete)
|
||||
assert excinfo_del.value.response.status_code == 404
|
||||
logger.info(
|
||||
f"Reading deleted note ID: {note_id_to_delete} correctly failed with 404."
|
||||
)
|
||||
|
||||
except HTTPStatusError as e:
|
||||
# If deletion fails unexpectedly, log it but don't fail the test here
|
||||
# as the primary goal was CRUD, and cleanup failure is secondary.
|
||||
logger.info(f"Error during cleanup (deleting note {note_id_to_delete}): {e}")
|
||||
except Exception as e:
|
||||
logger.info(f"Unexpected error during cleanup: {e}")
|
||||
else:
|
||||
logger.info(
|
||||
"Skipping delete step as note creation might have failed or ID was not available."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_delete_nonexistent_note(nc_client: NextcloudClient):
|
||||
"""Test deleting a note that doesn't exist."""
|
||||
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:
|
||||
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."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_note_attachment_integration(nc_client: NextcloudClient):
|
||||
"""
|
||||
Integration test for adding and retrieving a note attachment via WebDAV.
|
||||
This test is conditional on WebDAV permissions being available.
|
||||
"""
|
||||
# --- Create Note ---
|
||||
unique_id = str(uuid.uuid4())
|
||||
note_title = f"Attachment Test Note {unique_id}"
|
||||
note_content = "Note for testing attachments."
|
||||
note_category = "AttachmentTesting"
|
||||
created_note = None
|
||||
note_id = None
|
||||
|
||||
try:
|
||||
logger.info(f"\nCreating note for attachment test: {note_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=note_title, content=note_content, category=note_category
|
||||
)
|
||||
assert created_note and "id" in created_note
|
||||
note_id = created_note["id"]
|
||||
logger.info(f"Note created with ID: {note_id}")
|
||||
time.sleep(1) # Allow time for note creation
|
||||
|
||||
# --- Try to Add Attachment ---
|
||||
attachment_filename = f"test_attachment_{unique_id}.txt"
|
||||
attachment_content = f"This is the content of {attachment_filename}".encode('utf-8')
|
||||
attachment_mime = "text/plain"
|
||||
|
||||
logger.info(f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}")
|
||||
# Assuming WebDAV should work now, directly call add_note_attachment
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
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) # Allow time for upload processing
|
||||
|
||||
# --- Get and Verify Attachment ---
|
||||
logger.info(f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}")
|
||||
retrieved_content, retrieved_mime = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes")
|
||||
|
||||
# --- Verify Attachment ---
|
||||
assert retrieved_content == attachment_content
|
||||
# Check if the expected mime type is part of the retrieved one (to handle charset)
|
||||
assert attachment_mime in retrieved_mime
|
||||
logger.info("Retrieved attachment content and mime type verified successfully.")
|
||||
|
||||
finally:
|
||||
# --- Delete Note (Cleanup) ---
|
||||
if note_id:
|
||||
logger.info(f"Attempting cleanup: deleting note ID: {note_id}")
|
||||
try:
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
logger.info(f"Note ID: {note_id} deleted successfully.")
|
||||
# Verify deletion
|
||||
time.sleep(1)
|
||||
with pytest.raises(HTTPStatusError) as excinfo_del:
|
||||
nc_client.notes_get_note(note_id=note_id)
|
||||
assert excinfo_del.value.response.status_code == 404
|
||||
logger.info(f"Verified note {note_id} deletion (404 received).")
|
||||
except Exception as e:
|
||||
logger.info(f"Error during cleanup (deleting note {note_id}): {e}")
|
||||
else:
|
||||
logger.info("Skipping cleanup as note ID was not obtained.")
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_note_attachment_with_category_integration(nc_client: NextcloudClient):
|
||||
"""
|
||||
Explicitly tests adding/retrieving an attachment for a note WITH a category.
|
||||
Functionally similar to test_note_attachment_integration but emphasizes the category.
|
||||
"""
|
||||
# --- Create Note with Category ---
|
||||
unique_id = str(uuid.uuid4())
|
||||
note_title = f"Category Attachment Test Note {unique_id}"
|
||||
note_content = "Note with category for testing attachments."
|
||||
note_category = "CategoryTest" # Explicitly using a category
|
||||
created_note = None
|
||||
note_id = None
|
||||
|
||||
try:
|
||||
logger.info(f"\nCreating note with category '{note_category}' for attachment test: {note_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=note_title, content=note_content, category=note_category
|
||||
)
|
||||
assert created_note and "id" in created_note
|
||||
note_id = created_note["id"]
|
||||
logger.info(f"Note with category created with ID: {note_id}")
|
||||
time.sleep(1)
|
||||
|
||||
# --- Try to Add Attachment ---
|
||||
attachment_filename = f"category_test_attachment_{unique_id}.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}")
|
||||
# Assuming WebDAV should work now, directly call add_note_attachment
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
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}")
|
||||
retrieved_content, retrieved_mime = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
logger.info(f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes")
|
||||
|
||||
# --- Verify Attachment ---
|
||||
assert retrieved_content == attachment_content
|
||||
assert attachment_mime in retrieved_mime # Check if expected mime is part of retrieved
|
||||
logger.info("Retrieved attachment content and mime type verified successfully for note with category.")
|
||||
|
||||
finally:
|
||||
# --- Delete Note (Cleanup) ---
|
||||
if note_id:
|
||||
logger.info(f"Attempting cleanup: deleting note ID: {note_id}")
|
||||
try:
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
logger.info(f"Note ID: {note_id} deleted successfully.")
|
||||
time.sleep(1)
|
||||
with pytest.raises(HTTPStatusError) as excinfo_del:
|
||||
nc_client.notes_get_note(note_id=note_id)
|
||||
assert excinfo_del.value.response.status_code == 404
|
||||
logger.info(f"Verified note {note_id} deletion (404 received).")
|
||||
except Exception as e:
|
||||
logger.info(f"Error during cleanup (deleting note {note_id}): {e}")
|
||||
else:
|
||||
logger.info("Skipping cleanup as note ID was not obtained.")
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_attachment_cleanup_behavior(nc_client: NextcloudClient):
|
||||
"""
|
||||
Test to document the behavior regarding note attachment cleanup.
|
||||
|
||||
This test confirms that when a note is deleted, its attachments remain in the system.
|
||||
This matches the behavior of the official Nextcloud Notes app, which also leaves
|
||||
orphaned attachments when notes are deleted.
|
||||
"""
|
||||
# --- Create Note ---
|
||||
unique_id = str(uuid.uuid4())
|
||||
note_title = f"Attachment Cleanup Test {unique_id}"
|
||||
note_content = "Test note for attachments cleanup."
|
||||
note_category = "AttachmentCleanupTest"
|
||||
|
||||
logger.info(f"\nCreating test note: {note_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=note_title, content=note_content, category=note_category
|
||||
)
|
||||
assert created_note and "id" in created_note
|
||||
note_id = created_note["id"]
|
||||
logger.info(f"Test note created with ID: {note_id}")
|
||||
time.sleep(1)
|
||||
|
||||
# Check authentication type
|
||||
auth_type = type(nc_client._client.auth).__name__
|
||||
logger.info(f"Client authentication type: {auth_type}")
|
||||
|
||||
# --- Try to Add Attachment ---
|
||||
attachment_filename = f"cleanup_test_{unique_id}.txt"
|
||||
attachment_content = f"Content for cleanup test".encode('utf-8')
|
||||
|
||||
logger.info(f"Adding attachment '{attachment_filename}' to note ID: {note_id}")
|
||||
# Removed try block as we expect WebDAV to work or fail the test
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
mime_type="text/plain"
|
||||
)
|
||||
assert upload_response["status_code"] in [201, 204]
|
||||
logger.info(f"Attachment added successfully (Status: {upload_response['status_code']}).")
|
||||
time.sleep(1)
|
||||
|
||||
# --- Verify Attachment Exists ---
|
||||
retrieved_content, _ = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
assert retrieved_content == attachment_content
|
||||
logger.info("Verified attachment exists and can be retrieved")
|
||||
|
||||
# Attachment operations successful - continue with test
|
||||
# has_webdav_access = True # No longer needed as we expect it to work or fail
|
||||
# except HTTPStatusError as e: # Removed the try/except block that skipped on 401
|
||||
# if e.response.status_code == 401:
|
||||
# logger.info(f"WebDAV access denied (401 Unauthorized). Skipping attachment tests.")
|
||||
# pytest.skip("WebDAV access denied (401 Unauthorized)")
|
||||
# else:
|
||||
# raise # Re-raise other HTTP errors
|
||||
|
||||
# --- Delete Note ---
|
||||
logger.info(f"Deleting note ID: {note_id}")
|
||||
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:
|
||||
nc_client.notes_get_note(note_id=note_id)
|
||||
assert excinfo.value.response.status_code == 404
|
||||
logger.info(f"Verified note deletion (404 received)")
|
||||
|
||||
# --- Verify Attachment Is Deleted (New Behavior) ---
|
||||
logger.info(f"Verifying attachment '{attachment_filename}' is deleted for note ID: {note_id}")
|
||||
with pytest.raises(HTTPStatusError) as excinfo_attach_del:
|
||||
nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
assert excinfo_attach_del.value.response.status_code == 404
|
||||
logger.info(f"Attachment '{attachment_filename}' correctly not found (404) after note deletion.")
|
||||
@@ -1,129 +0,0 @@
|
||||
import pytest
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
import logging
|
||||
import tempfile
|
||||
from PIL import Image, ImageDraw
|
||||
from io import BytesIO
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def nc_client() -> NextcloudClient:
|
||||
"""
|
||||
Fixture to create a NextcloudClient instance for integration tests.
|
||||
"""
|
||||
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
|
||||
assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
|
||||
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
|
||||
return NextcloudClient.from_env()
|
||||
|
||||
@pytest.fixture
|
||||
def test_image():
|
||||
"""Generate a test image with embedded text for attachment tests"""
|
||||
# Create a temporary file to store the test image
|
||||
fd, image_path = tempfile.mkstemp(suffix='.png')
|
||||
os.close(fd)
|
||||
|
||||
# Create a test image with text
|
||||
img = Image.new('RGB', (300, 200), color=(255, 255, 255))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([(20, 20), (280, 180)], fill=(0, 120, 212))
|
||||
draw.text((50, 90), "Nextcloud Notes Test Image", fill=(255, 255, 255))
|
||||
img.save(image_path)
|
||||
|
||||
try:
|
||||
yield image_path
|
||||
finally:
|
||||
# Clean up the temporary image file
|
||||
if os.path.exists(image_path):
|
||||
os.unlink(image_path)
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_note_with_embedded_image(nc_client: NextcloudClient, test_image):
|
||||
"""
|
||||
Test creating a note with an embedded image and verify the process works end-to-end.
|
||||
This test documents how images should be embedded in Nextcloud Notes.
|
||||
"""
|
||||
# Generate a unique identifier for this test run
|
||||
unique_id = str(uuid.uuid4())[:8]
|
||||
note_title = f"Embedded Image Test {unique_id}"
|
||||
initial_content = "# Embedded Image Test\n\nThis note demonstrates how to properly embed images in Nextcloud Notes."
|
||||
|
||||
# Create the note
|
||||
logger.info(f"Creating test note: {note_title}")
|
||||
note = nc_client.notes_create_note(
|
||||
title=note_title,
|
||||
content=initial_content,
|
||||
category="Documentation"
|
||||
)
|
||||
note_id = note["id"]
|
||||
note_etag = note["etag"]
|
||||
logger.info(f"Note created with ID: {note_id}")
|
||||
|
||||
try:
|
||||
# Read the test image content
|
||||
with open(test_image, 'rb') as f:
|
||||
image_content = f.read()
|
||||
|
||||
# Generate a unique filename for the attachment
|
||||
attachment_filename = f"test_image_{unique_id}.png"
|
||||
|
||||
# Upload the image as an attachment
|
||||
logger.info(f"Uploading image attachment '{attachment_filename}' to note {note_id}...")
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=image_content,
|
||||
mime_type="image/png"
|
||||
)
|
||||
logger.info(f"Image uploaded: {upload_response}")
|
||||
|
||||
# Update the note content to include the embedded image using Markdown syntax
|
||||
# This is the correct syntax for embedding images in Nextcloud Notes
|
||||
updated_content = f"""# Embedded Image Test
|
||||
|
||||
This note demonstrates how to properly embed images in Nextcloud Notes.
|
||||
|
||||
## Method 1: Markdown Image Syntax
|
||||
|
||||
|
||||
## Method 2: HTML Image Tag
|
||||
<img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="300" />
|
||||
|
||||
## Notes on Image Embedding
|
||||
- Images must be stored in the .attachments.{note_id} directory
|
||||
- Images are referenced using relative paths
|
||||
- Both Markdown and HTML image tags work in Nextcloud Notes
|
||||
- The Nextcloud Notes UI will display these images inline when viewing the note
|
||||
"""
|
||||
|
||||
# Update the note with the image references
|
||||
logger.info("Updating note content with image references...")
|
||||
updated_note = nc_client.notes_update_note(
|
||||
note_id=note_id,
|
||||
etag=note_etag,
|
||||
content=updated_content
|
||||
)
|
||||
|
||||
# Verify the updated note has the correct content
|
||||
retrieved_note = nc_client.notes_get_note(note_id=note_id)
|
||||
assert ".attachments." in retrieved_note["content"], "Image reference not found in note content"
|
||||
logger.info("Note updated successfully with image references")
|
||||
|
||||
# Verify we can retrieve the image attachment
|
||||
retrieved_content, mime_type = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
assert len(retrieved_content) > 0, "Retrieved image content is empty"
|
||||
assert mime_type.startswith("image/"), f"Expected image mime type, got {mime_type}"
|
||||
|
||||
logger.info("Test completed successfully - image was embedded in the note and can be retrieved")
|
||||
|
||||
finally:
|
||||
# Clean up - delete the test note
|
||||
logger.info(f"Cleaning up - deleting test note {note_id}")
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
@@ -1,92 +0,0 @@
|
||||
import pytest
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
import uuid
|
||||
from httpx import HTTPStatusError
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Tests assume NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars are set
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_attachment_deleted_after_note_deletion(nc_client: NextcloudClient):
|
||||
"""
|
||||
Test to verify that when a note is deleted, its attachments are also deleted
|
||||
by the MCP client's modified notes_delete_note method.
|
||||
"""
|
||||
# --- Create Note ---
|
||||
unique_id = str(uuid.uuid4())
|
||||
note_title = f"Attachment Cleanup Test {unique_id}"
|
||||
note_content = f"# Test for attachment cleanup behavior\n\nThis note and its attachments should be deleted."
|
||||
note_category = "CleanupTests"
|
||||
|
||||
created_note = None
|
||||
note_id = None
|
||||
|
||||
try:
|
||||
# Create the note
|
||||
logger.info(f"Creating note: {note_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=note_title,
|
||||
content=note_content,
|
||||
category=note_category
|
||||
)
|
||||
assert created_note and "id" in created_note
|
||||
note_id = created_note["id"]
|
||||
logger.info(f"Note created with ID: {note_id}")
|
||||
time.sleep(1)
|
||||
|
||||
# Create a simple text attachment
|
||||
attachment_filename = f"cleanup_test_{unique_id}.txt"
|
||||
attachment_content = f"This is a test attachment for note {note_id}".encode('utf-8')
|
||||
|
||||
# Attach the file to the note
|
||||
logger.info(f"Attaching text file to note {note_id}...")
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=attachment_content,
|
||||
mime_type="text/plain"
|
||||
)
|
||||
|
||||
assert upload_response["status_code"] in [201, 204]
|
||||
logger.info(f"Attachment added successfully (Status: {upload_response['status_code']}).")
|
||||
time.sleep(1)
|
||||
|
||||
# Verify the attachment exists before deletion
|
||||
logger.info(f"Verifying attachment exists before deletion...")
|
||||
content, mime_type = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
assert content == attachment_content, "Attachment content mismatch before deletion"
|
||||
logger.info("Attachment verified before deletion")
|
||||
|
||||
# Now delete the note (which should also delete the attachment directory)
|
||||
logger.info(f"Deleting note ID: {note_id}")
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
logger.info(f"Note deleted successfully.")
|
||||
time.sleep(1)
|
||||
|
||||
# Verify the note is deleted
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
nc_client.notes_get_note(note_id=note_id)
|
||||
assert excinfo.value.response.status_code == 404
|
||||
logger.info(f"Verified note deletion (404 Not Found)")
|
||||
|
||||
# Now check if the attachment is deleted (expected behavior: it should be)
|
||||
logger.info(f"Checking if attachment is deleted after note deletion...")
|
||||
with pytest.raises(HTTPStatusError) as excinfo:
|
||||
nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
# We expect a 404 because the attachment (and its directory) should be gone
|
||||
assert excinfo.value.response.status_code == 404
|
||||
logger.info("CONFIRMED: Attachment is deleted after note deletion (404 Not Found)")
|
||||
|
||||
finally:
|
||||
# No cleanup needed as the test itself cleans up the note and attachment
|
||||
pass
|
||||
@@ -1,136 +0,0 @@
|
||||
import pytest
|
||||
import os
|
||||
import time
|
||||
import uuid
|
||||
import logging
|
||||
import tempfile
|
||||
from httpx import HTTPStatusError
|
||||
from PIL import Image, ImageDraw
|
||||
from nextcloud_mcp_server.client import NextcloudClient
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Tests assume NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars are set
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def nc_client() -> NextcloudClient:
|
||||
"""
|
||||
Fixture to create a NextcloudClient instance for integration tests.
|
||||
"""
|
||||
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
|
||||
assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
|
||||
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
|
||||
return NextcloudClient.from_env()
|
||||
|
||||
@pytest.fixture
|
||||
def test_image():
|
||||
"""Generate a test image for attachment tests"""
|
||||
# Create a temporary file to store the test image
|
||||
fd, image_path = tempfile.mkstemp(suffix='.png')
|
||||
os.close(fd)
|
||||
|
||||
# Create a simple test image
|
||||
img = Image.new('RGB', (200, 200), color = (255, 255, 255))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.rectangle([(20, 20), (180, 180)], fill=(255, 0, 0))
|
||||
draw.text((40, 100), "Nextcloud MCP Test", fill=(255, 255, 255))
|
||||
img.save(image_path)
|
||||
|
||||
try:
|
||||
yield image_path
|
||||
finally:
|
||||
# Clean up the temporary image file
|
||||
if os.path.exists(image_path):
|
||||
os.unlink(image_path)
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_note_with_image_attachment(nc_client: NextcloudClient, test_image):
|
||||
"""
|
||||
Test creating a note with an image attachment and properly embedding it
|
||||
in the note content using Nextcloud Notes' syntax.
|
||||
"""
|
||||
# --- Create Note ---
|
||||
unique_id = str(uuid.uuid4())
|
||||
note_title = f"Note with Embedded Image {unique_id}"
|
||||
note_content = "# Note with Embedded Image\n\nThis note contains an embedded image."
|
||||
note_category = "ImageTests"
|
||||
|
||||
created_note = None
|
||||
note_id = None
|
||||
|
||||
try:
|
||||
# Create the note
|
||||
logger.info(f"Creating note: {note_title}")
|
||||
created_note = nc_client.notes_create_note(
|
||||
title=note_title,
|
||||
content=note_content,
|
||||
category=note_category
|
||||
)
|
||||
assert created_note and "id" in created_note
|
||||
note_id = created_note["id"]
|
||||
logger.info(f"Note created with ID: {note_id}")
|
||||
time.sleep(1)
|
||||
|
||||
# Read the test image
|
||||
with open(test_image, 'rb') as f:
|
||||
image_content = f.read()
|
||||
|
||||
# Attach the image to the note
|
||||
attachment_filename = f"test_image_{unique_id}.png"
|
||||
logger.info(f"Attaching image to note {note_id}...")
|
||||
upload_response = nc_client.add_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename,
|
||||
content=image_content,
|
||||
mime_type="image/png"
|
||||
)
|
||||
|
||||
assert upload_response["status_code"] in [201, 204]
|
||||
logger.info(f"Image attached successfully (Status: {upload_response['status_code']}).")
|
||||
time.sleep(1)
|
||||
|
||||
# Update the note content to include a reference to the attached image
|
||||
# Try embedding using Markdown image syntax
|
||||
updated_content = f"""# Note with Embedded Image
|
||||
|
||||
This note contains an embedded image.
|
||||
|
||||
## Embedded Image (Markdown Syntax)
|
||||
|
||||
|
||||
## WebDAV URL
|
||||
Files path: `/Notes/.attachments.{note_id}/{attachment_filename}`
|
||||
"""
|
||||
|
||||
# Update the note content
|
||||
logger.info("Updating note content to include image reference...")
|
||||
updated_note = nc_client.notes_update_note(
|
||||
note_id=note_id,
|
||||
etag=created_note["etag"],
|
||||
content=updated_content
|
||||
)
|
||||
|
||||
# Retrieve the note to verify content
|
||||
retrieved_note = nc_client.notes_get_note(note_id=note_id)
|
||||
logger.info("Retrieved note content:")
|
||||
logger.info(retrieved_note["content"])
|
||||
|
||||
# Verify the image attachment can be retrieved
|
||||
content, mime_type = nc_client.get_note_attachment(
|
||||
note_id=note_id,
|
||||
filename=attachment_filename
|
||||
)
|
||||
|
||||
assert content == image_content, "Attachment content mismatch"
|
||||
assert mime_type.startswith("image/"), f"Expected image mime type, got {mime_type}"
|
||||
logger.info("Image attachment verified")
|
||||
|
||||
finally:
|
||||
# Cleanup
|
||||
if note_id:
|
||||
logger.info(f"Cleaning up - deleting note ID: {note_id}")
|
||||
try:
|
||||
nc_client.notes_delete_note(note_id=note_id)
|
||||
logger.info(f"Note {note_id} deleted")
|
||||
except Exception as e:
|
||||
logger.info(f"Error during cleanup: {e}")
|
||||
Reference in New Issue
Block a user