admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: Update tests with async

Browse Source
This commit is contained in:
Chris Coutinho
2025-06-06 19:10:10 +02:00
parent 110df3d7b9
commit 5ee9435741
7 changed files with 116 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client.py
+4 -4
View File
@@ -8,7 +8,7 @@ from httpx import (
Request, Request,
Response, Response,
HTTPStatusError, HTTPStatusError,
) # Import HTTPStatusError )
import logging import logging
@@ -172,8 +172,8 @@ class NextcloudClient:
async def notes_append_content(self, *, note_id: int, content: str): async def notes_append_content(self, *, note_id: int, content: str):
"""Append content to an existing note. """Append content to an existing note.
The content will be separated by a newline, delimiter `---`, and The content will be separated by a newline and a delimiter `---`, so
timestemp so callers do not need to append metadata themselves. one will not be required in the content provided to this tool
""" """
logger.info(f"Appending content to note {note_id}") logger.info(f"Appending content to note {note_id}")
@@ -181,7 +181,7 @@ class NextcloudClient:
current_note = await self.notes_get_note(note_id=note_id) current_note = await self.notes_get_note(note_id=note_id)
# Use fixed separator for consistency # Use fixed separator for consistency
separator = f"\n---\n## Content appended: {dt.datetime.now():%Y-%m-%d %H:%M}\n" separator = f"\n---\n"
# Combine content # Combine content
existing_content = current_note.get("content", "") existing_content = current_note.get("content", "")
pyproject.toml
+2
View File
@@ -18,6 +18,8 @@ nc-mcp-server = "nextcloud_mcp_server.server:run"
[tool.pytest.ini_options] [tool.pytest.ini_options]
asyncio_mode = "auto" asyncio_mode = "auto"
asyncio_default_test_loop_scope = "session"
asyncio_default_fixture_loop_scope = "session"
log_cli = 1 log_cli = 1
log_cli_level = "WARN" log_cli_level = "WARN"
log_level = "WARN" log_level = "WARN"
tests/conftest.py
+9
View File
@@ -3,9 +3,13 @@ import os
import logging import logging
import uuid import uuid
from nextcloud_mcp_server.client import NextcloudClient, HTTPStatusError from nextcloud_mcp_server.client import NextcloudClient, HTTPStatusError
import asyncio
import pytest_asyncio
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# pytestmark = pytest.mark.asyncio(loop_scope="package")
@pytest.fixture(scope="session") @pytest.fixture(scope="session")
async def nc_client() -> NextcloudClient: async def nc_client() -> NextcloudClient:
@@ -13,6 +17,7 @@ async def nc_client() -> NextcloudClient:
Fixture to create a NextcloudClient instance for integration tests. Fixture to create a NextcloudClient instance for integration tests.
Uses environment variables for configuration. Uses environment variables for configuration.
""" """
assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set" 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_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set" assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
@@ -36,6 +41,8 @@ async def temporary_note(nc_client: NextcloudClient):
Fixture to create a temporary note for a test and ensure its deletion afterward. Fixture to create a temporary note for a test and ensure its deletion afterward.
Yields the created note dictionary. Yields the created note dictionary.
""" """
asyncio.new_event_loop()
note_id = None note_id = None
unique_suffix = uuid.uuid4().hex[:8] unique_suffix = uuid.uuid4().hex[:8]
note_title = f"Temporary Test Note {unique_suffix}" note_title = f"Temporary Test Note {unique_suffix}"
@@ -80,6 +87,8 @@ async def temporary_note_with_attachment(
Yields a tuple: (note_data, attachment_filename, attachment_content). Yields a tuple: (note_data, attachment_filename, attachment_content).
Depends on the temporary_note fixture. Depends on the temporary_note fixture.
""" """
asyncio.new_event_loop()
note_data = temporary_note note_data = temporary_note
note_id = note_data["id"] note_id = note_data["id"]
note_category = note_data.get("category") # Get category from the note data note_category = note_data.get("category") # Get category from the note data
tests/integration/test_attachments.py
+45 -45
View File
@@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
def test_attachments_add_and_get( async def test_attachments_add_and_get(
nc_client: NextcloudClient, temporary_note_with_attachment: tuple nc_client: NextcloudClient, temporary_note_with_attachment: tuple
): ):
""" """
@@ -29,7 +29,7 @@ def test_attachments_add_and_get(
f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}" f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}"
) )
# Pass category to get_note_attachment # Pass category to get_note_attachment
retrieved_content, retrieved_mime = nc_client.get_note_attachment( retrieved_content, retrieved_mime = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=note_category note_id=note_id, filename=attachment_filename, category=note_category
) )
logger.info( logger.info(
@@ -41,7 +41,7 @@ def test_attachments_add_and_get(
logger.info("Retrieved attachment content and mime type verified successfully.") logger.info("Retrieved attachment content and mime type verified successfully.")
def test_attachments_add_to_note_with_category( async def test_attachments_add_to_note_with_category(
nc_client: NextcloudClient, temporary_note: dict nc_client: NextcloudClient, temporary_note: dict
): ):
""" """
@@ -67,7 +67,7 @@ def test_attachments_add_to_note_with_category(
f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}" f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}"
) )
# Pass category to add_note_attachment # Pass category to add_note_attachment
upload_response = nc_client.add_note_attachment( upload_response = await nc_client.add_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
content=attachment_content, content=attachment_content,
@@ -86,7 +86,7 @@ def test_attachments_add_to_note_with_category(
f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}" f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}"
) )
# Pass category to get_note_attachment # Pass category to get_note_attachment
retrieved_content, retrieved_mime = nc_client.get_note_attachment( retrieved_content, retrieved_mime = await nc_client.get_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
category=note_category, # Pass the note's category category=note_category, # Pass the note's category
@@ -103,7 +103,7 @@ def test_attachments_add_to_note_with_category(
# Cleanup is handled by the temporary_note fixture # Cleanup is handled by the temporary_note fixture
def test_attachments_cleanup_on_note_delete( async def test_attachments_cleanup_on_note_delete(
nc_client: NextcloudClient, temporary_note_with_attachment: tuple nc_client: NextcloudClient, temporary_note_with_attachment: tuple
): ):
""" """
@@ -127,13 +127,13 @@ def test_attachments_cleanup_on_note_delete(
# Manually delete the note # Manually delete the note
logger.info(f"Manually deleting note ID: {note_id} within the test.") logger.info(f"Manually deleting note ID: {note_id} within the test.")
nc_client.notes_delete_note(note_id=note_id) await nc_client.notes_delete_note(note_id=note_id)
logger.info(f"Note ID: {note_id} deleted successfully.") logger.info(f"Note ID: {note_id} deleted successfully.")
time.sleep(1) time.sleep(1)
# Verify Note Is Deleted # Verify Note Is Deleted
with pytest.raises(HTTPStatusError) as excinfo_note: with pytest.raises(HTTPStatusError) as excinfo_note:
nc_client.notes_get_note(note_id=note_id) await nc_client.notes_get_note(note_id=note_id)
assert excinfo_note.value.response.status_code == 404 assert excinfo_note.value.response.status_code == 404
logger.info(f"Verified note {note_id} deletion (404 received).") logger.info(f"Verified note {note_id} deletion (404 received).")
@@ -145,7 +145,7 @@ def test_attachments_cleanup_on_note_delete(
# Pass category to get_note_attachment - although it should fail anyway # Pass category to get_note_attachment - although it should fail anyway
# because the note (and thus details) are gone. # because the note (and thus details) are gone.
# The client method will raise 404 from the initial notes_get_note call. # The client method will raise 404 from the initial notes_get_note call.
nc_client.get_note_attachment( await nc_client.get_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
category=note_category, # Pass category, though note fetch should fail first category=note_category, # Pass category, though note fetch should fail first
@@ -165,7 +165,7 @@ def test_attachments_cleanup_on_note_delete(
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", attachment_dir_path, headers=propfind_headers "PROPFIND", attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -173,13 +173,13 @@ def test_attachments_cleanup_on_note_delete(
logger.error( logger.error(
f"Attachment directory still exists! PROPFIND returned {status}" f"Attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected attachment directory to be gone, but PROPFIND returned {status}!" False
) ), f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified attachment directory does not exist via PROPFIND (404 received)" "Verified attachment directory does not exist via PROPFIND (404 received)"
) )
@@ -188,7 +188,7 @@ def test_attachments_cleanup_on_note_delete(
# but it will find the note already deleted (404) and handle it gracefully. # but it will find the note already deleted (404) and handle it gracefully.
def test_attachments_category_change_handling(nc_client: NextcloudClient): async def test_attachments_category_change_handling(nc_client: NextcloudClient):
""" """
Tests attachment handling when a note's category is changed. Tests attachment handling when a note's category is changed.
Verifies attachment retrieval works before and after category change, Verifies attachment retrieval works before and after category change,
@@ -205,7 +205,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
try: try:
# 1. Create note with initial category # 1. Create note with initial category
logger.info(f"Creating note '{note_title}' in category '{initial_category}'") logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
created_note = nc_client.notes_create_note( created_note = await nc_client.notes_create_note(
title=note_title, content="Initial content", category=initial_category title=note_title, content="Initial content", category=initial_category
) )
note_id = created_note["id"] note_id = created_note["id"]
@@ -217,7 +217,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info( logger.info(
f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})" f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
) )
upload_response = nc_client.add_note_attachment( upload_response = await nc_client.add_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
content=attachment_content, content=attachment_content,
@@ -232,7 +232,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info( logger.info(
f"Verifying attachment retrieval from initial category '{initial_category}'" f"Verifying attachment retrieval from initial category '{initial_category}'"
) )
retrieved_content1, _ = nc_client.get_note_attachment( retrieved_content1, _ = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category note_id=note_id, filename=attachment_filename, category=initial_category
) )
assert retrieved_content1 == attachment_content assert retrieved_content1 == attachment_content
@@ -243,9 +243,9 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"Updating note {note_id} category from '{initial_category}' to '{new_category}'" 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) # Need to fetch the latest etag after attachment add (WebDAV ops don't update note etag)
current_note_data = nc_client.notes_get_note(note_id=note_id) current_note_data = await nc_client.notes_get_note(note_id=note_id)
current_etag = current_note_data["etag"] current_etag = current_note_data["etag"]
updated_note = nc_client.notes_update_note( updated_note = await nc_client.notes_update_note(
note_id=note_id, note_id=note_id,
etag=current_etag, etag=current_etag,
category=new_category, category=new_category,
@@ -261,7 +261,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.info( logger.info(
f"Verifying attachment retrieval from new category '{new_category}'" f"Verifying attachment retrieval from new category '{new_category}'"
) )
retrieved_content2, _ = nc_client.get_note_attachment( retrieved_content2, _ = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=new_category note_id=note_id, filename=attachment_filename, category=new_category
) )
assert retrieved_content2 == attachment_content assert retrieved_content2 == attachment_content
@@ -275,7 +275,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", old_attachment_dir_path, headers=propfind_headers "PROPFIND", old_attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -283,13 +283,13 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.error( logger.error(
f"Old attachment directory still exists! PROPFIND returned {status}" f"Old attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected old directory to be gone, but PROPFIND returned {status} - directory still exists!" False
) ), f"Expected old directory to be gone, but PROPFIND returned {status} - directory still exists!"
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified old attachment directory does not exist via PROPFIND (404 received)" "Verified old attachment directory does not exist via PROPFIND (404 received)"
) )
@@ -300,7 +300,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}" f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
) )
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", new_attachment_dir_path, headers=propfind_headers "PROPFIND", new_attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -315,9 +315,9 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
logger.error( logger.error(
f"New attachment directory not found! PROPFIND failed with {e.response.status_code}" f"New attachment directory not found! PROPFIND failed with {e.response.status_code}"
) )
assert False, ( assert (
f"Expected new attachment directory to exist, but PROPFIND failed with {e.response.status_code}" False
) ), f"Expected new attachment directory to exist, but PROPFIND failed with {e.response.status_code}"
finally: finally:
# 6. Cleanup: Delete the note (client should use the *final* category for cleanup path) # 6. Cleanup: Delete the note (client should use the *final* category for cleanup path)
@@ -326,18 +326,18 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"Cleaning up note ID: {note_id} (last known category: '{new_category}')" f"Cleaning up note ID: {note_id} (last known category: '{new_category}')"
) )
try: try:
nc_client.notes_delete_note(note_id=note_id) await nc_client.notes_delete_note(note_id=note_id)
logger.info(f"Note {note_id} deleted.") logger.info(f"Note {note_id} deleted.")
time.sleep(1) time.sleep(1)
# Verify note deletion # Verify note deletion
with pytest.raises(HTTPStatusError) as excinfo_note_del: with pytest.raises(HTTPStatusError) as excinfo_note_del:
nc_client.notes_get_note(note_id=note_id) await nc_client.notes_get_note(note_id=note_id)
assert excinfo_note_del.value.response.status_code == 404 assert excinfo_note_del.value.response.status_code == 404
logger.info("Verified note deleted (404).") logger.info("Verified note deleted (404).")
# Verify attachment deletion (should fail with 404 on the initial note fetch) # Verify attachment deletion (should fail with 404 on the initial note fetch)
with pytest.raises(HTTPStatusError) as excinfo_attach_del: with pytest.raises(HTTPStatusError) as excinfo_attach_del:
# Pass the *last known* category, although the note fetch should fail first # Pass the *last known* category, although the note fetch should fail first
nc_client.get_note_attachment( await nc_client.get_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
category=new_category, category=new_category,
@@ -359,7 +359,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
resp = nc_client._client.request( resp = await nc_client._client.request(
"PROPFIND", new_attachment_dir_path, headers=propfind_headers "PROPFIND", new_attachment_dir_path, headers=propfind_headers
) )
if resp.status_code in [ if resp.status_code in [
@@ -368,9 +368,9 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
]: # Successful PROPFIND means directory exists ]: # Successful PROPFIND means directory exists
assert False, "New category attachment directory still exists!" assert False, "New category attachment directory still exists!"
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified new category attachment directory is gone via PROPFIND" "Verified new category attachment directory is gone via PROPFIND"
) )
@@ -380,7 +380,7 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}" f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
) )
try: try:
resp = nc_client._client.request( resp = await nc_client._client.request(
"PROPFIND", old_attachment_dir_path, headers=propfind_headers "PROPFIND", old_attachment_dir_path, headers=propfind_headers
) )
if resp.status_code in [ if resp.status_code in [
@@ -389,9 +389,9 @@ def test_attachments_category_change_handling(nc_client: NextcloudClient):
]: # Successful PROPFIND means directory exists ]: # Successful PROPFIND means directory exists
assert False, "Old category attachment directory still exists!" assert False, "Old category attachment directory still exists!"
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified old category attachment directory is gone via PROPFIND" "Verified old category attachment directory is gone via PROPFIND"
) )
tests/integration/test_embedded_images.py
+18 -18
View File
@@ -40,7 +40,7 @@ def test_image_data() -> tuple[bytes, str]:
return image_bytes, suggested_filename return image_bytes, suggested_filename
def test_note_with_embedded_image( async def test_note_with_embedded_image(
nc_client: NextcloudClient, temporary_note: dict, test_image_data: tuple nc_client: NextcloudClient, temporary_note: dict, test_image_data: tuple
): ):
""" """
@@ -62,7 +62,7 @@ def test_note_with_embedded_image(
logger.info( logger.info(
f"Uploading image attachment '{attachment_filename}' to note {note_id} (category: '{note_category or ''}')..." f"Uploading image attachment '{attachment_filename}' to note {note_id} (category: '{note_category or ''}')..."
) )
upload_response = nc_client.add_note_attachment( upload_response = await nc_client.add_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
content=image_content, content=image_content,
@@ -84,7 +84,7 @@ def test_note_with_embedded_image(
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", attachment_dir_path, headers=propfind_headers "PROPFIND", attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -99,9 +99,9 @@ def test_note_with_embedded_image(
logger.error( logger.error(
f"Attachment directory not found! PROPFIND failed with {e.response.status_code}" f"Attachment directory not found! PROPFIND failed with {e.response.status_code}"
) )
assert False, ( assert (
f"Expected attachment directory to exist, but PROPFIND failed with {e.response.status_code}" 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 # 2. Update the note content to include the embedded image references
updated_content = f"""{note_data["content"]} updated_content = f"""{note_data["content"]}
@@ -115,7 +115,7 @@ def test_note_with_embedded_image(
<img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="150" /> <img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="150" />
""" """
logger.info("Updating note content with image references...") logger.info("Updating note content with image references...")
updated_note = nc_client.notes_update_note( updated_note = await nc_client.notes_update_note(
note_id=note_id, note_id=note_id,
etag=note_etag, # Use etag from the created note etag=note_etag, # Use etag from the created note
content=updated_content, content=updated_content,
@@ -128,7 +128,7 @@ def test_note_with_embedded_image(
time.sleep(1) time.sleep(1)
# 3. Verify the updated note content # 3. Verify the updated note content
retrieved_note = nc_client.notes_get_note(note_id=note_id) retrieved_note = await nc_client.notes_get_note(note_id=note_id)
assert f".attachments.{note_id}/{attachment_filename}" in retrieved_note["content"] assert f".attachments.{note_id}/{attachment_filename}" in retrieved_note["content"]
logger.info("Verified image reference exists in updated note content.") logger.info("Verified image reference exists in updated note content.")
@@ -137,7 +137,7 @@ def test_note_with_embedded_image(
f"Retrieving image attachment '{attachment_filename}' (category: '{note_category or ''}')..." f"Retrieving image attachment '{attachment_filename}' (category: '{note_category or ''}')..."
) )
# Pass category to get_note_attachment # Pass category to get_note_attachment
retrieved_img_content, mime_type = nc_client.get_note_attachment( retrieved_img_content, mime_type = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=note_category note_id=note_id, filename=attachment_filename, category=note_category
) )
assert retrieved_img_content == image_content assert retrieved_img_content == image_content
@@ -150,20 +150,20 @@ def test_note_with_embedded_image(
logger.info( logger.info(
f"Manually deleting note ID: {note_id} to verify proper attachment cleanup" f"Manually deleting note ID: {note_id} to verify proper attachment cleanup"
) )
nc_client.notes_delete_note(note_id=note_id) await nc_client.notes_delete_note(note_id=note_id)
logger.info(f"Note ID: {note_id} deleted successfully.") logger.info(f"Note ID: {note_id} deleted successfully.")
time.sleep(1) time.sleep(1)
# 6. Verify note is deleted # 6. Verify note is deleted
with pytest.raises(HTTPStatusError) as excinfo_note: with pytest.raises(HTTPStatusError) as excinfo_note:
nc_client.notes_get_note(note_id=note_id) await nc_client.notes_get_note(note_id=note_id)
assert excinfo_note.value.response.status_code == 404 assert excinfo_note.value.response.status_code == 404
logger.info(f"Verified note {note_id} deletion (404 received).") logger.info(f"Verified note {note_id} deletion (404 received).")
# 7. Verify attachment directory is deleted via WebDAV PROPFIND # 7. Verify attachment directory is deleted via WebDAV PROPFIND
logger.info("Directly verifying attachment directory doesn't exist via PROPFIND") logger.info("Directly verifying attachment directory doesn't exist via PROPFIND")
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", attachment_dir_path, headers=propfind_headers "PROPFIND", attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -171,13 +171,13 @@ def test_note_with_embedded_image(
logger.error( logger.error(
f"Attachment directory still exists! PROPFIND returned {status}" f"Attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected attachment directory to be gone, but PROPFIND returned {status}!" False
) ), f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified attachment directory does not exist via PROPFIND (404 received)" "Verified attachment directory does not exist via PROPFIND (404 received)"
) )
tests/integration/test_notes_api.py
-8
View File
@@ -14,7 +14,6 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
@pytest.mark.asyncio
async def test_notes_api_create_and_read( async def test_notes_api_create_and_read(
nc_client: NextcloudClient, temporary_note: dict nc_client: NextcloudClient, temporary_note: dict
): ):
@@ -34,7 +33,6 @@ async def test_notes_api_create_and_read(
logger.info(f"Successfully read and verified note ID: {note_id}") logger.info(f"Successfully read and verified note ID: {note_id}")
@pytest.mark.asyncio
async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict): async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict):
""" """
Tests updating a note created by the fixture. Tests updating a note created by the fixture.
@@ -74,7 +72,6 @@ async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict
logger.info(f"Successfully updated and verified note ID: {note_id}") logger.info(f"Successfully updated and verified note ID: {note_id}")
@pytest.mark.asyncio
async def test_notes_api_update_conflict( async def test_notes_api_update_conflict(
nc_client: NextcloudClient, temporary_note: dict nc_client: NextcloudClient, temporary_note: dict
): ):
@@ -115,7 +112,6 @@ async def test_notes_api_update_conflict(
logger.info("Update with old etag correctly failed with 412 Precondition Failed.") logger.info("Update with old etag correctly failed with 412 Precondition Failed.")
@pytest.mark.asyncio
async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient): async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
""" """
Tests deleting a note that doesn't exist fails with 404. Tests deleting a note that doesn't exist fails with 404.
@@ -130,7 +126,6 @@ async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
) )
@pytest.mark.asyncio
async def test_notes_api_append_content_to_existing_note( async def test_notes_api_append_content_to_existing_note(
nc_client: NextcloudClient, temporary_note: dict nc_client: NextcloudClient, temporary_note: dict
): ):
@@ -165,7 +160,6 @@ async def test_notes_api_append_content_to_existing_note(
logger.info(f"Successfully appended content to note ID: {note_id}") logger.info(f"Successfully appended content to note ID: {note_id}")
@pytest.mark.asyncio
async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient): 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). Tests appending content to an empty note (no separator should be added).
@@ -208,7 +202,6 @@ async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient
logger.warning(f"Failed to clean up test note ID: {note_id}: {e}") logger.warning(f"Failed to clean up test note ID: {note_id}: {e}")
@pytest.mark.asyncio
async def test_notes_api_append_content_multiple_times( async def test_notes_api_append_content_multiple_times(
nc_client: NextcloudClient, temporary_note: dict nc_client: NextcloudClient, temporary_note: dict
): ):
@@ -249,7 +242,6 @@ async def test_notes_api_append_content_multiple_times(
logger.info(f"Successfully performed multiple appends to note ID: {note_id}") logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
@pytest.mark.asyncio
async def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient): async def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient):
""" """
Tests that appending to a non-existent note fails with 404. Tests that appending to a non-existent note fails with 404.
tests/integration/test_webdav_cleanup.py
+38 -38
View File
@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
pytestmark = pytest.mark.integration pytestmark = pytest.mark.integration
def test_category_change_cleans_up_old_attachments_directory( async def test_category_change_cleans_up_old_attachments_directory(
nc_client: NextcloudClient, nc_client: NextcloudClient,
): ):
""" """
@@ -29,7 +29,7 @@ def test_category_change_cleans_up_old_attachments_directory(
try: try:
# 1. Create note with initial category # 1. Create note with initial category
logger.info(f"Creating note '{note_title}' in category '{initial_category}'") logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
created_note = nc_client.notes_create_note( created_note = await nc_client.notes_create_note(
title=note_title, content="Initial content", category=initial_category title=note_title, content="Initial content", category=initial_category
) )
note_id = created_note["id"] note_id = created_note["id"]
@@ -41,7 +41,7 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.info( logger.info(
f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})" f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
) )
upload_response = nc_client.add_note_attachment( upload_response = await nc_client.add_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
content=attachment_content, content=attachment_content,
@@ -56,7 +56,7 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.info( logger.info(
f"Verifying attachment retrieval from initial category '{initial_category}'" f"Verifying attachment retrieval from initial category '{initial_category}'"
) )
retrieved_content1, _ = nc_client.get_note_attachment( retrieved_content1, _ = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category note_id=note_id, filename=attachment_filename, category=initial_category
) )
assert retrieved_content1 == attachment_content assert retrieved_content1 == attachment_content
@@ -72,9 +72,9 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.info( logger.info(
f"Updating note {note_id} category from '{initial_category}' to '{new_category}'" f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
) )
current_note_data = nc_client.notes_get_note(note_id=note_id) current_note_data = await nc_client.notes_get_note(note_id=note_id)
current_etag = current_note_data["etag"] current_etag = current_note_data["etag"]
updated_note = nc_client.notes_update_note( updated_note = await nc_client.notes_update_note(
note_id=note_id, note_id=note_id,
etag=current_etag, etag=current_etag,
category=new_category, category=new_category,
@@ -90,7 +90,7 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.info( logger.info(
f"Verifying attachment retrieval from new category '{new_category}'" f"Verifying attachment retrieval from new category '{new_category}'"
) )
retrieved_content2, _ = nc_client.get_note_attachment( retrieved_content2, _ = await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=new_category note_id=note_id, filename=attachment_filename, category=new_category
) )
assert retrieved_content2 == attachment_content assert retrieved_content2 == attachment_content
@@ -101,24 +101,24 @@ def test_category_change_cleans_up_old_attachments_directory(
f"Trying to retrieve attachment from old category '{initial_category}' - should fail" f"Trying to retrieve attachment from old category '{initial_category}' - should fail"
) )
try: try:
nc_client.get_note_attachment( await nc_client.get_note_attachment(
note_id=note_id, filename=attachment_filename, category=initial_category note_id=note_id, filename=attachment_filename, category=initial_category
) )
# If we get here, it means the old directory still exists (a problem) # If we get here, it means the old directory still exists (a problem)
logger.error( logger.error(
"ISSUE DETECTED: Was able to retrieve attachment from old category path!" "ISSUE DETECTED: Was able to retrieve attachment from old category path!"
) )
assert False, ( assert (
"Old category attachment directory still exists and accessible!" False
) ), "Old category attachment directory still exists and accessible!"
except HTTPStatusError as e: except HTTPStatusError as e:
# This is the expected outcome - old directory should be gone # This is the expected outcome - old directory should be gone
logger.info( logger.info(
f"Correctly got error accessing old category path: {e.response.status_code}" f"Correctly got error accessing old category path: {e.response.status_code}"
) )
assert e.response.status_code == 404, ( assert (
f"Expected 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified old category attachment directory is not accessible (good!)" "Verified old category attachment directory is not accessible (good!)"
) )
@@ -133,7 +133,7 @@ def test_category_change_cleans_up_old_attachments_directory(
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", old_attachment_dir_path, headers=propfind_headers "PROPFIND", old_attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -144,18 +144,18 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.error( logger.error(
f"Old attachment directory still exists! PROPFIND returned {status}" f"Old attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected old attachment directory to be gone, but it still exists (PROPFIND returned {status})!" False
) ), f"Expected old attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
# If we got another status code (like 404), it's also good - the directory doesn't exist # If we got another status code (like 404), it's also good - the directory doesn't exist
logger.info( logger.info(
f"Verified old attachment directory does not exist (PROPFIND returned {status})" f"Verified old attachment directory does not exist (PROPFIND returned {status})"
) )
except HTTPStatusError as e: except HTTPStatusError as e:
# 404 is expected - directory should not exist # 404 is expected - directory should not exist
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified old attachment directory does not exist via PROPFIND (404 received)" "Verified old attachment directory does not exist via PROPFIND (404 received)"
) )
@@ -165,14 +165,14 @@ def test_category_change_cleans_up_old_attachments_directory(
if note_id: if note_id:
logger.info(f"Cleaning up note ID: {note_id}") logger.info(f"Cleaning up note ID: {note_id}")
try: try:
nc_client.notes_delete_note(note_id=note_id) await nc_client.notes_delete_note(note_id=note_id)
logger.info(f"Note {note_id} deleted.") logger.info(f"Note {note_id} deleted.")
time.sleep(1) time.sleep(1)
# 9. Verify both old and new attachment paths are gone # 9. Verify both old and new attachment paths are gone
logger.info("Verifying all attachment paths are gone") logger.info("Verifying all attachment paths are gone")
with pytest.raises(HTTPStatusError) as excinfo_new: with pytest.raises(HTTPStatusError) as excinfo_new:
nc_client.get_note_attachment( await nc_client.get_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
category=new_category, category=new_category,
@@ -180,7 +180,7 @@ def test_category_change_cleans_up_old_attachments_directory(
assert excinfo_new.value.response.status_code == 404 assert excinfo_new.value.response.status_code == 404
with pytest.raises(HTTPStatusError) as excinfo_old: with pytest.raises(HTTPStatusError) as excinfo_old:
nc_client.get_note_attachment( await nc_client.get_note_attachment(
note_id=note_id, note_id=note_id,
filename=attachment_filename, filename=attachment_filename,
category=initial_category, category=initial_category,
@@ -199,7 +199,7 @@ def test_category_change_cleans_up_old_attachments_directory(
) )
propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"} propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", new_attachment_dir_path, headers=propfind_headers "PROPFIND", new_attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -210,17 +210,17 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.error( logger.error(
f"New category attachment directory still exists! PROPFIND returned {status}" f"New category attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected new category attachment directory to be gone, but it still exists (PROPFIND returned {status})!" False
) ), f"Expected new category attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
# If we got another status code (like 404), it's also good - the directory doesn't exist # If we got another status code (like 404), it's also good - the directory doesn't exist
logger.info( logger.info(
f"Verified new category attachment directory does not exist (PROPFIND returned {status})" f"Verified new category attachment directory does not exist (PROPFIND returned {status})"
) )
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified new category attachment directory is gone via PROPFIND" "Verified new category attachment directory is gone via PROPFIND"
) )
@@ -230,7 +230,7 @@ def test_category_change_cleans_up_old_attachments_directory(
f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}" f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
) )
try: try:
propfind_resp = nc_client._client.request( propfind_resp = await nc_client._client.request(
"PROPFIND", old_attachment_dir_path, headers=propfind_headers "PROPFIND", old_attachment_dir_path, headers=propfind_headers
) )
status = propfind_resp.status_code status = propfind_resp.status_code
@@ -241,17 +241,17 @@ def test_category_change_cleans_up_old_attachments_directory(
logger.error( logger.error(
f"Old category attachment directory still exists! PROPFIND returned {status}" f"Old category attachment directory still exists! PROPFIND returned {status}"
) )
assert False, ( assert (
f"Expected old category attachment directory to be gone, but it still exists (PROPFIND returned {status})!" False
) ), f"Expected old category attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
# If we got another status code (like 404), it's also good - the directory doesn't exist # If we got another status code (like 404), it's also good - the directory doesn't exist
logger.info( logger.info(
f"Verified old category attachment directory does not exist (PROPFIND returned {status})" f"Verified old category attachment directory does not exist (PROPFIND returned {status})"
) )
except HTTPStatusError as e: except HTTPStatusError as e:
assert e.response.status_code == 404, ( assert (
f"Expected PROPFIND to fail with 404, got {e.response.status_code}" e.response.status_code == 404
) ), f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
logger.info( logger.info(
"Verified old category attachment directory is gone via PROPFIND" "Verified old category attachment directory is gone via PROPFIND"
) )