From cdb3dda3e4aab3946c23f923ae5eb35101e0ca02 Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Sun, 27 Apr 2025 20:06:51 +0530 Subject: [PATCH 1/7] feat(recyclebin): add REST API services for recyclebin management --- src/plone/restapi/services/configure.zcml | 1 + .../restapi/services/recyclebin/__init__.py | 1 + .../services/recyclebin/configure.zcml | 53 ++++ .../restapi/services/recyclebin/delete.py | 82 +++++ src/plone/restapi/services/recyclebin/get.py | 76 +++++ src/plone/restapi/services/recyclebin/post.py | 47 +++ .../restapi/tests/test_services_recyclebin.py | 282 ++++++++++++++++++ 7 files changed, 542 insertions(+) create mode 100644 src/plone/restapi/services/recyclebin/__init__.py create mode 100644 src/plone/restapi/services/recyclebin/configure.zcml create mode 100644 src/plone/restapi/services/recyclebin/delete.py create mode 100644 src/plone/restapi/services/recyclebin/get.py create mode 100644 src/plone/restapi/services/recyclebin/post.py create mode 100644 src/plone/restapi/tests/test_services_recyclebin.py diff --git a/src/plone/restapi/services/configure.zcml b/src/plone/restapi/services/configure.zcml index 0f42573cd..2a37051b6 100644 --- a/src/plone/restapi/services/configure.zcml +++ b/src/plone/restapi/services/configure.zcml @@ -34,6 +34,7 @@ + diff --git a/src/plone/restapi/services/recyclebin/__init__.py b/src/plone/restapi/services/recyclebin/__init__.py new file mode 100644 index 000000000..9fd3d107e --- /dev/null +++ b/src/plone/restapi/services/recyclebin/__init__.py @@ -0,0 +1 @@ +"""Plone REST API services for recyclebin management""" \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/configure.zcml b/src/plone/restapi/services/recyclebin/configure.zcml new file mode 100644 index 000000000..4cd04fbeb --- /dev/null +++ b/src/plone/restapi/services/recyclebin/configure.zcml @@ -0,0 +1,53 @@ + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/delete.py b/src/plone/restapi/services/recyclebin/delete.py new file mode 100644 index 000000000..789dfeef9 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/delete.py @@ -0,0 +1,82 @@ +from plone.restapi.deserializer import json_body +from plone.restapi.services import Service +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from zExceptions import BadRequest +from zope.component import getUtility +from zope.interface import alsoProvides +from zope.interface import implementer +from zope.publisher.interfaces import IPublishTraverse + +import plone.protect.interfaces + + +@implementer(IPublishTraverse) +class RecycleBinItemDelete(Service): + """Permanently delete an item from the recyclebin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.item_id = None + + def publishTraverse(self, request, name): + self.item_id = name + return self + + def reply(self): + if not self.item_id: + raise BadRequest("No item ID provided") + + # Disable CSRF protection + if "IDisableCSRFProtection" in dir(plone.protect.interfaces): + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + recyclebin = getUtility(IRecycleBin) + success = recyclebin.purge_item(self.item_id) + + if not success: + self.request.response.setStatus(404) + return {"error": {"type": "NotFound", "message": "Item not found or cannot be deleted"}} + + return self.reply_no_content() + + +class RecycleBinEmpty(Service): + """Empty recyclebin by deleting all items""" + + def reply(self): + # Disable CSRF protection + if "IDisableCSRFProtection" in dir(plone.protect.interfaces): + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + # Check for optional body parameters + data = {} + try: + data = json_body(self.request) + except Exception: + pass + + # Optional filter criteria could be added here + # filter_criteria = data.get("filter", {}) + + recyclebin = getUtility(IRecycleBin) + items = recyclebin.get_items() + + purged_count = 0 + failed_items = [] + + for item in items: + if recyclebin.purge_item(item["recycle_id"]): + purged_count += 1 + else: + failed_items.append(item["recycle_id"]) + + # If we have failures, return them with a 207 Multi-Status response + if failed_items: + self.request.response.setStatus(207) + return { + "purged": purged_count, + "failed": failed_items, + } + + # If everything succeeded, return 204 No Content + return self.reply_no_content() \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/get.py b/src/plone/restapi/services/recyclebin/get.py new file mode 100644 index 000000000..c5d7fec98 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/get.py @@ -0,0 +1,76 @@ +from plone.restapi.interfaces import IExpandableElement +from plone.restapi.services import Service +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from zope.component import adapter +from zope.component import getUtility +from zope.interface import implementer +from zope.interface import Interface +from zope.publisher.interfaces import IPublishTraverse + + +@implementer(IExpandableElement) +@adapter(Interface, Interface) +class RecycleBin: + """Get recyclebin information.""" + + def __init__(self, context, request): + self.context = context + self.request = request + + def __call__(self, expand=False): + result = {"recyclebin": {"@id": f"{self.context.absolute_url()}/@recyclebin"}} + if not expand: + return result + + recyclebin = getUtility(IRecycleBin) + items = recyclebin.get_items() + + results = [] + for item in items: + # Exclude the actual object from the response + data = {k: v for k, v in item.items() if k != "object"} + # Convert datetime to ISO format + if "deletion_date" in data: + data["deletion_date"] = data["deletion_date"].isoformat() + results.append(data) + + return {"recyclebin": {"items": results, "items_total": len(results)}} + + +class RecycleBinGet(Service): + """Get list of all items in recyclebin""" + + def reply(self): + recyclebin = RecycleBin(self.context, self.request) + return recyclebin(expand=True)["recyclebin"] + + +@implementer(IPublishTraverse) +class RecycleBinItemGet(Service): + """Get a specific item from the recyclebin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.item_id = None + + def publishTraverse(self, request, name): + self.item_id = name + return self + + def reply(self): + if not self.item_id: + return self.reply_no_content(status=404) + + recyclebin = getUtility(IRecycleBin) + item = recyclebin.get_item(self.item_id) + + if not item: + return self.reply_no_content(status=404) + + # Exclude the actual object from the response + data = {k: v for k, v in item.items() if k != "object"} + # Convert datetime to ISO format + if "deletion_date" in data: + data["deletion_date"] = data["deletion_date"].isoformat() + + return data \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/post.py b/src/plone/restapi/services/recyclebin/post.py new file mode 100644 index 000000000..b37eefc03 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/post.py @@ -0,0 +1,47 @@ +from plone.restapi.deserializer import json_body +from plone.restapi.services import Service +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from zExceptions import BadRequest +from zope.component import getUtility +from zope.interface import alsoProvides +from zope.interface import implementer +from zope.publisher.interfaces import IPublishTraverse + +import plone.protect.interfaces + + +@implementer(IPublishTraverse) +class RecycleBinItemRestore(Service): + """Restore an item from the recyclebin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.item_id = None + + def publishTraverse(self, request, name): + self.item_id = name + return self + + def reply(self): + if not self.item_id: + raise BadRequest("No item ID provided") + + # Disable CSRF protection + if "IDisableCSRFProtection" in dir(plone.protect.interfaces): + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + recyclebin = getUtility(IRecycleBin) + restored = recyclebin.restore_item(self.item_id) + + if not restored: + self.request.response.setStatus(404) + return {"error": {"type": "NotFound", "message": "Item not found or cannot be restored"}} + + self.request.response.setStatus(200) + return { + "status": "success", + "message": "Item restored successfully", + "@id": restored.absolute_url(), + "title": getattr(restored, "title", ""), + "path": "/".join(restored.getPhysicalPath()) + } \ No newline at end of file diff --git a/src/plone/restapi/tests/test_services_recyclebin.py b/src/plone/restapi/tests/test_services_recyclebin.py new file mode 100644 index 000000000..245a24ff4 --- /dev/null +++ b/src/plone/restapi/tests/test_services_recyclebin.py @@ -0,0 +1,282 @@ +import unittest +from datetime import datetime +import json +from plone.app.testing import PLONE_INTEGRATION_TESTING +from plone.app.testing import TEST_USER_ID +from plone.app.testing import setRoles +from zope.component import getUtility +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from plone.restapi.testing import RelativeSession + + +class TestRecycleBinRESTAPI(unittest.TestCase): + """Test the RecycleBin REST API endpoints.""" + + layer = PLONE_INTEGRATION_TESTING + + def setUp(self): + self.app = self.layer["app"] + self.portal = self.layer["portal"] + self.portal_url = self.portal.absolute_url() + self.api_session = RelativeSession(self.portal_url) + self.api_session.headers.update({"Accept": "application/json"}) + self.recyclebin = getUtility(IRecycleBin) + + # Set up test content and permissions + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + + # Create and delete an item to have something in the recyclebin + self.portal.invokeFactory("Document", id="test-doc", title="Test Document") + test_doc = self.portal["test-doc"] + + # Store original path for verification in tests + self.original_path = "/".join(test_doc.getPhysicalPath()) + + # Remove the item to put it in the recyclebin + self.portal.manage_delObjects(["test-doc"]) + + # Get the recycle_id for the deleted item + items = self.recyclebin.get_items() + self.recycle_id = items[0]["recycle_id"] if items else None + + def test_get_recyclebin_items(self): + """Test getting all items from recyclebin.""" + response = self.api_session.get("/@recyclebin") + + self.assertEqual(response.status_code, 200) + data = response.json() + + self.assertIn("items", data) + self.assertIn("items_total", data) + self.assertGreaterEqual(data["items_total"], 1) + + # Verify the deleted document is in the items list + items = data["items"] + self.assertTrue(any(item["title"] == "Test Document" for item in items)) + + # Verify datetime is properly serialized as ISO format + for item in items: + self.assertIn("deletion_date", item) + # Check if it's a valid ISO date format (this is a simple check) + self.assertIn("T", item["deletion_date"]) + + def test_get_recyclebin_item(self): + """Test getting a specific item from recyclebin.""" + if not self.recycle_id: + self.skipTest("No items in recyclebin") + + response = self.api_session.get(f"/@recyclebin/{self.recycle_id}") + + self.assertEqual(response.status_code, 200) + data = response.json() + + self.assertEqual(data["title"], "Test Document") + self.assertEqual(data["recycle_id"], self.recycle_id) + self.assertIn("deletion_date", data) + + # Verify the 'object' is not in the response + self.assertNotIn("object", data) + + def test_get_nonexistent_recyclebin_item(self): + """Test getting a nonexistent item from recyclebin.""" + response = self.api_session.get("/@recyclebin/non-existent-id") + + self.assertEqual(response.status_code, 404) + + def test_restore_recyclebin_item(self): + """Test restoring an item from recyclebin.""" + if not self.recycle_id: + self.skipTest("No items in recyclebin") + + response = self.api_session.post(f"/@recyclebin/{self.recycle_id}/restore") + + self.assertEqual(response.status_code, 200) + data = response.json() + + self.assertEqual(data["status"], "success") + self.assertIn("message", data) + self.assertIn("@id", data) + self.assertEqual(data["title"], "Test Document") + + # Verify the item exists in the portal + self.assertIn("test-doc", self.portal) + + # Try getting the item from recyclebin (should fail) + response = self.api_session.get(f"/@recyclebin/{self.recycle_id}") + self.assertEqual(response.status_code, 404) + + def test_restore_nonexistent_recyclebin_item(self): + """Test restoring a nonexistent item from recyclebin.""" + response = self.api_session.post("/@recyclebin/non-existent-id/restore") + + self.assertEqual(response.status_code, 404) + data = response.json() + self.assertIn("error", data) + + def test_delete_recyclebin_item(self): + """Test permanently deleting an item from recyclebin.""" + # First, create and delete another item + self.portal.invokeFactory("Document", id="doc-to-purge", title="Document to Purge") + self.portal.manage_delObjects(["doc-to-purge"]) + + # Get the recycle_id for the newly deleted item + items = self.recyclebin.get_items() + purge_id = next((item["recycle_id"] for item in items + if item.get("title") == "Document to Purge"), None) + + if not purge_id: + self.skipTest("Could not find Document to Purge in recyclebin") + + response = self.api_session.delete(f"/@recyclebin/{purge_id}") + + self.assertEqual(response.status_code, 204) + + # Verify item is no longer in recyclebin + response = self.api_session.get(f"/@recyclebin/{purge_id}") + self.assertEqual(response.status_code, 404) + + def test_delete_nonexistent_recyclebin_item(self): + """Test deleting a nonexistent item from recyclebin.""" + response = self.api_session.delete("/@recyclebin/non-existent-id") + + self.assertEqual(response.status_code, 404) + data = response.json() + self.assertIn("error", data) + + def test_empty_recyclebin(self): + """Test emptying the entire recyclebin.""" + # First ensure there's something in the recyclebin + if not self.recyclebin.get_items(): + # Create and delete a new item if recyclebin is empty + self.portal.invokeFactory("Document", id="doc-for-empty", title="Document for Empty") + self.portal.manage_delObjects(["doc-for-empty"]) + + # Get initial count + initial_count = len(self.recyclebin.get_items()) + self.assertGreater(initial_count, 0, "Recyclebin is empty, can't test emptying it") + + # Empty the recyclebin + response = self.api_session.delete("/@recyclebin") + + self.assertEqual(response.status_code, 204) + + # Verify recyclebin is now empty + self.assertEqual(len(self.recyclebin.get_items()), 0) + + def test_empty_recyclebin_with_failures(self): + """Test emptying recyclebin with simulated failures.""" + # This test requires mocking the recyclebin's purge_item method + # to simulate failures. Since mocking is complex in this context, + # we'll provide a stub implementation that you can expand with the + # appropriate mocking library in your actual tests. + + # Skip this test for now - implement with proper mocking in real tests + self.skipTest("This test requires mocking and will be implemented separately") + + # Example of how this would work with mocking: + # with mock.patch.object(self.recyclebin, 'purge_item', side_effect=[True, False, True]): + # response = self.api_session.delete("/@recyclebin") + # self.assertEqual(response.status_code, 207) + # data = response.json() + # self.assertEqual(data["purged"], 2) + # self.assertEqual(len(data["failed"]), 1) + + +class TestRecycleBinRESTAPIIntegration(unittest.TestCase): + """Integration tests for RecycleBin REST API.""" + + layer = PLONE_INTEGRATION_TESTING + + def setUp(self): + self.app = self.layer["app"] + self.portal = self.layer["portal"] + self.portal_url = self.portal.absolute_url() + self.api_session = RelativeSession(self.portal_url) + self.api_session.headers.update({"Accept": "application/json"}) + + # Set up test content and permissions + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + + # Create a folder with content + self.portal.invokeFactory("Folder", id="test-folder", title="Test Folder") + folder = self.portal["test-folder"] + folder.invokeFactory("Document", id="doc1", title="Document 1") + folder.invokeFactory("Document", id="doc2", title="Document 2") + + # Delete the folder to put it and its contents in recyclebin + self.portal.manage_delObjects(["test-folder"]) + + # Get the recycle IDs + self.recyclebin = getUtility(IRecycleBin) + self.items = self.recyclebin.get_items() + self.folder_id = next((item["recycle_id"] for item in self.items + if item.get("title") == "Test Folder"), None) + + def test_restore_folder_hierarchy(self): + """Test restoring a folder with its contents.""" + if not self.folder_id: + self.skipTest("Could not find Test Folder in recyclebin") + + # Restore the folder + response = self.api_session.post(f"/@recyclebin/{self.folder_id}/restore") + + self.assertEqual(response.status_code, 200) + + # Verify the folder exists + self.assertIn("test-folder", self.portal) + folder = self.portal["test-folder"] + + # Verify the folder's contents were restored + self.assertIn("doc1", folder) + self.assertIn("doc2", folder) + + # Verify items are no longer in recyclebin + items = self.recyclebin.get_items() + folder_items = [item for item in items if item.get("title") in ["Test Folder", "Document 1", "Document 2"]] + self.assertEqual(len(folder_items), 0) + + def test_recyclebin_workflow_integration(self): + """Test recyclebin integration with workflow state.""" + # Create a document with a specific workflow state + self.portal.invokeFactory("Document", id="workflow-doc", title="Workflow Document") + doc = self.portal["workflow-doc"] + + # Change workflow state (if your test environment supports this) + try: + workflow_tool = self.portal.portal_workflow + if 'simple_publication_workflow' in workflow_tool.getWorkflowsFor(doc): + workflow_tool.doActionFor(doc, 'publish') + original_state = workflow_tool.getInfoFor(doc, 'review_state') + self.assertEqual(original_state, 'published') + else: + self.skipTest("Required workflow not available") + return + except Exception: + self.skipTest("Workflow transition failed") + return + + # Delete the document + self.portal.manage_delObjects(["workflow-doc"]) + + # Find the document in recyclebin + items = self.recyclebin.get_items() + doc_id = next((item["recycle_id"] for item in items + if item.get("title") == "Workflow Document"), None) + + if not doc_id: + self.skipTest("Could not find Workflow Document in recyclebin") + return + + # Restore the document + self.api_session.post(f"/@recyclebin/{doc_id}/restore") + + # Verify workflow state is preserved (if your implementation preserves it) + doc = self.portal["workflow-doc"] + restored_state = workflow_tool.getInfoFor(doc, 'review_state') + + # This assertion might fail if your recyclebin implementation + # doesn't preserve workflow state - adjust accordingly + self.assertEqual(restored_state, original_state, + "Workflow state should be preserved after restore") + + From 659bdc39c8403250c11b58899969be27f3d0badb Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Sun, 27 Apr 2025 22:07:35 +0530 Subject: [PATCH 2/7] feat(recyclebin): refactor and enhance REST API for recyclebin management --- .../restapi/services/recyclebin/__init__.py | 2 +- .../services/recyclebin/configure.zcml | 73 ++--- .../restapi/services/recyclebin/delete.py | 82 ----- .../services/recyclebin/documentation.md | 147 +++++++++ src/plone/restapi/services/recyclebin/get.py | 104 +++---- src/plone/restapi/services/recyclebin/post.py | 47 --- .../restapi/services/recyclebin/purge.py | 93 ++++++ .../restapi/services/recyclebin/restore.py | 96 ++++++ .../restapi/tests/test_services_recyclebin.py | 282 ------------------ 9 files changed, 405 insertions(+), 521 deletions(-) delete mode 100644 src/plone/restapi/services/recyclebin/delete.py create mode 100644 src/plone/restapi/services/recyclebin/documentation.md delete mode 100644 src/plone/restapi/services/recyclebin/post.py create mode 100644 src/plone/restapi/services/recyclebin/purge.py create mode 100644 src/plone/restapi/services/recyclebin/restore.py delete mode 100644 src/plone/restapi/tests/test_services_recyclebin.py diff --git a/src/plone/restapi/services/recyclebin/__init__.py b/src/plone/restapi/services/recyclebin/__init__.py index 9fd3d107e..a000b69d0 100644 --- a/src/plone/restapi/services/recyclebin/__init__.py +++ b/src/plone/restapi/services/recyclebin/__init__.py @@ -1 +1 @@ -"""Plone REST API services for recyclebin management""" \ No newline at end of file +# Empty init file to make the directory a Python package diff --git a/src/plone/restapi/services/recyclebin/configure.zcml b/src/plone/restapi/services/recyclebin/configure.zcml index 4cd04fbeb..0c13176c2 100644 --- a/src/plone/restapi/services/recyclebin/configure.zcml +++ b/src/plone/restapi/services/recyclebin/configure.zcml @@ -3,51 +3,38 @@ xmlns:plone="http://namespaces.plone.org/plone" xmlns:zcml="http://namespaces.zope.org/zcml"> - + - - + - - + - - + - - + - - - - \ No newline at end of file + diff --git a/src/plone/restapi/services/recyclebin/delete.py b/src/plone/restapi/services/recyclebin/delete.py deleted file mode 100644 index 789dfeef9..000000000 --- a/src/plone/restapi/services/recyclebin/delete.py +++ /dev/null @@ -1,82 +0,0 @@ -from plone.restapi.deserializer import json_body -from plone.restapi.services import Service -from Products.CMFPlone.interfaces.recyclebin import IRecycleBin -from zExceptions import BadRequest -from zope.component import getUtility -from zope.interface import alsoProvides -from zope.interface import implementer -from zope.publisher.interfaces import IPublishTraverse - -import plone.protect.interfaces - - -@implementer(IPublishTraverse) -class RecycleBinItemDelete(Service): - """Permanently delete an item from the recyclebin""" - - def __init__(self, context, request): - super().__init__(context, request) - self.item_id = None - - def publishTraverse(self, request, name): - self.item_id = name - return self - - def reply(self): - if not self.item_id: - raise BadRequest("No item ID provided") - - # Disable CSRF protection - if "IDisableCSRFProtection" in dir(plone.protect.interfaces): - alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) - - recyclebin = getUtility(IRecycleBin) - success = recyclebin.purge_item(self.item_id) - - if not success: - self.request.response.setStatus(404) - return {"error": {"type": "NotFound", "message": "Item not found or cannot be deleted"}} - - return self.reply_no_content() - - -class RecycleBinEmpty(Service): - """Empty recyclebin by deleting all items""" - - def reply(self): - # Disable CSRF protection - if "IDisableCSRFProtection" in dir(plone.protect.interfaces): - alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) - - # Check for optional body parameters - data = {} - try: - data = json_body(self.request) - except Exception: - pass - - # Optional filter criteria could be added here - # filter_criteria = data.get("filter", {}) - - recyclebin = getUtility(IRecycleBin) - items = recyclebin.get_items() - - purged_count = 0 - failed_items = [] - - for item in items: - if recyclebin.purge_item(item["recycle_id"]): - purged_count += 1 - else: - failed_items.append(item["recycle_id"]) - - # If we have failures, return them with a 207 Multi-Status response - if failed_items: - self.request.response.setStatus(207) - return { - "purged": purged_count, - "failed": failed_items, - } - - # If everything succeeded, return 204 No Content - return self.reply_no_content() \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/documentation.md b/src/plone/restapi/services/recyclebin/documentation.md new file mode 100644 index 000000000..8b9ff9890 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/documentation.md @@ -0,0 +1,147 @@ +# Recycle Bin REST API + +The Recycle Bin REST API provides endpoints to interact with the Plone Recycle Bin functionality. + +## List recycle bin contents + +To list all items in the recycle bin, send a GET request to the `@recyclebin` endpoint: + +```http +GET /@recyclebin HTTP/1.1 +Accept: application/json +``` + +Response: + +```json +{ + "@id": "http://localhost:8080/Plone/@recyclebin", + "items": [ + { + "@id": "http://localhost:8080/Plone/@recyclebin/6d6d626f-8c85-4f22-8747-adb979bbe3b1", + "id": "document-1", + "title": "My Document", + "type": "Document", + "path": "/Plone/folder/document-1", + "parent_path": "/Plone/folder", + "deletion_date": "2025-04-27T10:30:45.123456", + "size": 1024, + "recycle_id": "6d6d626f-8c85-4f22-8747-adb979bbe3b1", + "actions": { + "restore": "http://localhost:8080/Plone/@recyclebin-restore", + "purge": "http://localhost:8080/Plone/@recyclebin-purge" + } + } + ], + "items_total": 1 +} +``` + +## Restore an item from the recycle bin + +To restore an item from the recycle bin, send a POST request to the `@recyclebin-restore` endpoint: + +```http +POST /@recyclebin-restore HTTP/1.1 +Accept: application/json +Content-Type: application/json + +{ + "item_id": "6d6d626f-8c85-4f22-8747-adb979bbe3b1" +} +``` + +You can optionally specify a target path to restore to: + +```json +{ + "item_id": "6d6d626f-8c85-4f22-8747-adb979bbe3b1", + "target_path": "/Plone/another-folder" +} +``` + +Response: + +```json +{ + "status": "success", + "message": "Item document-1 restored successfully", + "restored_item": { + "@id": "http://localhost:8080/Plone/document-1", + "id": "document-1", + "title": "My Document", + "type": "Document" + } +} +``` + +## Purge an item from the recycle bin + +To permanently delete an item from the recycle bin, send a POST request to the `@recyclebin-purge` endpoint: + +```http +POST /@recyclebin-purge HTTP/1.1 +Accept: application/json +Content-Type: application/json + +{ + "item_id": "6d6d626f-8c85-4f22-8747-adb979bbe3b1" +} +``` + +Response: + +```json +{ + "status": "success", + "message": "Item document-1 purged successfully" +} +``` + +## Purge all items from the recycle bin + +To purge all items from the recycle bin: + +```http +POST /@recyclebin-purge HTTP/1.1 +Accept: application/json +Content-Type: application/json + +{ + "purge_all": true +} +``` + +Response: + +```json +{ + "status": "success", + "purged_count": 5, + "message": "Purged 5 items from recycle bin" +} +``` + +## Purge expired items from the recycle bin + +To purge only expired items (based on the retention period): + +```http +POST /@recyclebin-purge HTTP/1.1 +Accept: application/json +Content-Type: application/json + +{ + "purge_expired": true +} +``` + +Response: + +```json +{ + "status": "success", + "purged_count": 2, + "message": "Purged 2 expired items from recycle bin" +} +``` diff --git a/src/plone/restapi/services/recyclebin/get.py b/src/plone/restapi/services/recyclebin/get.py index c5d7fec98..f0237ee70 100644 --- a/src/plone/restapi/services/recyclebin/get.py +++ b/src/plone/restapi/services/recyclebin/get.py @@ -1,76 +1,48 @@ -from plone.restapi.interfaces import IExpandableElement from plone.restapi.services import Service from Products.CMFPlone.interfaces.recyclebin import IRecycleBin -from zope.component import adapter from zope.component import getUtility -from zope.interface import implementer -from zope.interface import Interface -from zope.publisher.interfaces import IPublishTraverse -@implementer(IExpandableElement) -@adapter(Interface, Interface) -class RecycleBin: - """Get recyclebin information.""" - - def __init__(self, context, request): - self.context = context - self.request = request - - def __call__(self, expand=False): - result = {"recyclebin": {"@id": f"{self.context.absolute_url()}/@recyclebin"}} - if not expand: - return result +class RecycleBinGet(Service): + """GET /@recyclebin - List items in the recycle bin""" - recyclebin = getUtility(IRecycleBin) - items = recyclebin.get_items() + def reply(self): + recycle_bin = getUtility(IRecycleBin) + + # Check if recycle bin is enabled + if not recycle_bin.is_enabled(): + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": "Recycle Bin is disabled", + } + } + # Get all items from recycle bin + items = recycle_bin.get_items() + + # Format items for response results = [] for item in items: - # Exclude the actual object from the response - data = {k: v for k, v in item.items() if k != "object"} - # Convert datetime to ISO format - if "deletion_date" in data: - data["deletion_date"] = data["deletion_date"].isoformat() - results.append(data) - - return {"recyclebin": {"items": results, "items_total": len(results)}} - - -class RecycleBinGet(Service): - """Get list of all items in recyclebin""" - - def reply(self): - recyclebin = RecycleBin(self.context, self.request) - return recyclebin(expand=True)["recyclebin"] - - -@implementer(IPublishTraverse) -class RecycleBinItemGet(Service): - """Get a specific item from the recyclebin""" - - def __init__(self, context, request): - super().__init__(context, request) - self.item_id = None - - def publishTraverse(self, request, name): - self.item_id = name - return self - - def reply(self): - if not self.item_id: - return self.reply_no_content(status=404) - - recyclebin = getUtility(IRecycleBin) - item = recyclebin.get_item(self.item_id) + results.append({ + "@id": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}", + "id": item["id"], + "title": item["title"], + "type": item["type"], + "path": item["path"], + "parent_path": item["parent_path"], + "deletion_date": item["deletion_date"].isoformat(), + "size": item["size"], + "recycle_id": item["recycle_id"], + "actions": { + "restore": f"{self.context.absolute_url()}/@recyclebin-restore", + "purge": f"{self.context.absolute_url()}/@recyclebin-purge" + } + }) - if not item: - return self.reply_no_content(status=404) - - # Exclude the actual object from the response - data = {k: v for k, v in item.items() if k != "object"} - # Convert datetime to ISO format - if "deletion_date" in data: - data["deletion_date"] = data["deletion_date"].isoformat() - - return data \ No newline at end of file + return { + "@id": f"{self.context.absolute_url()}/@recyclebin", + "items": results, + "items_total": len(results) + } diff --git a/src/plone/restapi/services/recyclebin/post.py b/src/plone/restapi/services/recyclebin/post.py deleted file mode 100644 index b37eefc03..000000000 --- a/src/plone/restapi/services/recyclebin/post.py +++ /dev/null @@ -1,47 +0,0 @@ -from plone.restapi.deserializer import json_body -from plone.restapi.services import Service -from Products.CMFPlone.interfaces.recyclebin import IRecycleBin -from zExceptions import BadRequest -from zope.component import getUtility -from zope.interface import alsoProvides -from zope.interface import implementer -from zope.publisher.interfaces import IPublishTraverse - -import plone.protect.interfaces - - -@implementer(IPublishTraverse) -class RecycleBinItemRestore(Service): - """Restore an item from the recyclebin""" - - def __init__(self, context, request): - super().__init__(context, request) - self.item_id = None - - def publishTraverse(self, request, name): - self.item_id = name - return self - - def reply(self): - if not self.item_id: - raise BadRequest("No item ID provided") - - # Disable CSRF protection - if "IDisableCSRFProtection" in dir(plone.protect.interfaces): - alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) - - recyclebin = getUtility(IRecycleBin) - restored = recyclebin.restore_item(self.item_id) - - if not restored: - self.request.response.setStatus(404) - return {"error": {"type": "NotFound", "message": "Item not found or cannot be restored"}} - - self.request.response.setStatus(200) - return { - "status": "success", - "message": "Item restored successfully", - "@id": restored.absolute_url(), - "title": getattr(restored, "title", ""), - "path": "/".join(restored.getPhysicalPath()) - } \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/purge.py b/src/plone/restapi/services/recyclebin/purge.py new file mode 100644 index 000000000..0261aa6c1 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/purge.py @@ -0,0 +1,93 @@ +from plone.restapi.deserializer import json_body +from plone.restapi.services import Service +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from zope.component import getUtility +from zope.interface import alsoProvides + +import plone.protect.interfaces + + +class RecycleBinPurge(Service): + """POST /@recyclebin-purge - Permanently delete an item from the recycle bin""" + + def reply(self): + # Disable CSRF protection for this request + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + data = json_body(self.request) + item_id = data.get("item_id", None) + purge_all = data.get("purge_all", False) + purge_expired = data.get("purge_expired", False) + + recycle_bin = getUtility(IRecycleBin) + + # Check if recycle bin is enabled + if not recycle_bin.is_enabled(): + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": "Recycle Bin is disabled", + } + } + + # Handle purging all items + if purge_all: + purged_count = 0 + for item in recycle_bin.get_items(): + if recycle_bin.purge_item(item["recycle_id"]): + purged_count += 1 + + return { + "status": "success", + "purged_count": purged_count, + "message": f"Purged {purged_count} items from recycle bin" + } + + # Handle purging expired items + if purge_expired: + purged_count = recycle_bin.purge_expired_items() + + return { + "status": "success", + "purged_count": purged_count, + "message": f"Purged {purged_count} expired items from recycle bin" + } + + # Handle purging a specific item + if not item_id: + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": "Missing required parameter: item_id or purge_all or purge_expired", + } + } + + # Get the item to purge + item_data = recycle_bin.get_item(item_id) + if not item_data: + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": f"Item with ID {item_id} not found in recycle bin", + } + } + + # Purge the item + success = recycle_bin.purge_item(item_id) + + if not success: + self.request.response.setStatus(500) + return { + "error": { + "type": "InternalServerError", + "message": "Failed to purge item", + } + } + + return { + "status": "success", + "message": f"Item {item_data['id']} purged successfully" + } diff --git a/src/plone/restapi/services/recyclebin/restore.py b/src/plone/restapi/services/recyclebin/restore.py new file mode 100644 index 000000000..0dd7311b1 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/restore.py @@ -0,0 +1,96 @@ +from plone.restapi.deserializer import json_body +from plone.restapi.services import Service +from Products.CMFPlone.interfaces.recyclebin import IRecycleBin +from zope.component import getUtility +from zope.interface import alsoProvides +from zope.publisher.interfaces import IPublishTraverse + +import plone.protect.interfaces + + +class RecycleBinRestore(Service): + """POST /@recyclebin-restore - Restore an item from the recycle bin""" + + def __init__(self, context, request): + super(RecycleBinRestore, self).__init__(context, request) + self.params = {} + + def reply(self): + # Disable CSRF protection for this request + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + data = json_body(self.request) + item_id = data.get("item_id", None) + + if not item_id: + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": "Missing required parameter: item_id", + } + } + + recycle_bin = getUtility(IRecycleBin) + + # Check if recycle bin is enabled + if not recycle_bin.is_enabled(): + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": "Recycle Bin is disabled", + } + } + + # Get the item to restore + item_data = recycle_bin.get_item(item_id) + if not item_data: + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": f"Item with ID {item_id} not found in recycle bin", + } + } + + # Get optional target container path + target_path = data.get("target_path", None) + target_container = None + + if target_path: + try: + portal = self.context.portal_url.getPortalObject() + target_container = portal.unrestrictedTraverse(target_path) + except (KeyError, AttributeError): + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": f"Target path {target_path} not found", + } + } + + # Restore the item + restored_obj = recycle_bin.restore_item(item_id, target_container) + + if not restored_obj: + self.request.response.setStatus(500) + return { + "error": { + "type": "InternalServerError", + "message": "Failed to restore item", + } + } + + self.request.response.setStatus(200) + return { + "status": "success", + "message": f"Item {item_data['id']} restored successfully", + "restored_item": { + "@id": restored_obj.absolute_url(), + "id": restored_obj.getId(), + "title": restored_obj.Title(), + "type": restored_obj.portal_type + } + } diff --git a/src/plone/restapi/tests/test_services_recyclebin.py b/src/plone/restapi/tests/test_services_recyclebin.py deleted file mode 100644 index 245a24ff4..000000000 --- a/src/plone/restapi/tests/test_services_recyclebin.py +++ /dev/null @@ -1,282 +0,0 @@ -import unittest -from datetime import datetime -import json -from plone.app.testing import PLONE_INTEGRATION_TESTING -from plone.app.testing import TEST_USER_ID -from plone.app.testing import setRoles -from zope.component import getUtility -from Products.CMFPlone.interfaces.recyclebin import IRecycleBin -from plone.restapi.testing import RelativeSession - - -class TestRecycleBinRESTAPI(unittest.TestCase): - """Test the RecycleBin REST API endpoints.""" - - layer = PLONE_INTEGRATION_TESTING - - def setUp(self): - self.app = self.layer["app"] - self.portal = self.layer["portal"] - self.portal_url = self.portal.absolute_url() - self.api_session = RelativeSession(self.portal_url) - self.api_session.headers.update({"Accept": "application/json"}) - self.recyclebin = getUtility(IRecycleBin) - - # Set up test content and permissions - setRoles(self.portal, TEST_USER_ID, ["Manager"]) - - # Create and delete an item to have something in the recyclebin - self.portal.invokeFactory("Document", id="test-doc", title="Test Document") - test_doc = self.portal["test-doc"] - - # Store original path for verification in tests - self.original_path = "/".join(test_doc.getPhysicalPath()) - - # Remove the item to put it in the recyclebin - self.portal.manage_delObjects(["test-doc"]) - - # Get the recycle_id for the deleted item - items = self.recyclebin.get_items() - self.recycle_id = items[0]["recycle_id"] if items else None - - def test_get_recyclebin_items(self): - """Test getting all items from recyclebin.""" - response = self.api_session.get("/@recyclebin") - - self.assertEqual(response.status_code, 200) - data = response.json() - - self.assertIn("items", data) - self.assertIn("items_total", data) - self.assertGreaterEqual(data["items_total"], 1) - - # Verify the deleted document is in the items list - items = data["items"] - self.assertTrue(any(item["title"] == "Test Document" for item in items)) - - # Verify datetime is properly serialized as ISO format - for item in items: - self.assertIn("deletion_date", item) - # Check if it's a valid ISO date format (this is a simple check) - self.assertIn("T", item["deletion_date"]) - - def test_get_recyclebin_item(self): - """Test getting a specific item from recyclebin.""" - if not self.recycle_id: - self.skipTest("No items in recyclebin") - - response = self.api_session.get(f"/@recyclebin/{self.recycle_id}") - - self.assertEqual(response.status_code, 200) - data = response.json() - - self.assertEqual(data["title"], "Test Document") - self.assertEqual(data["recycle_id"], self.recycle_id) - self.assertIn("deletion_date", data) - - # Verify the 'object' is not in the response - self.assertNotIn("object", data) - - def test_get_nonexistent_recyclebin_item(self): - """Test getting a nonexistent item from recyclebin.""" - response = self.api_session.get("/@recyclebin/non-existent-id") - - self.assertEqual(response.status_code, 404) - - def test_restore_recyclebin_item(self): - """Test restoring an item from recyclebin.""" - if not self.recycle_id: - self.skipTest("No items in recyclebin") - - response = self.api_session.post(f"/@recyclebin/{self.recycle_id}/restore") - - self.assertEqual(response.status_code, 200) - data = response.json() - - self.assertEqual(data["status"], "success") - self.assertIn("message", data) - self.assertIn("@id", data) - self.assertEqual(data["title"], "Test Document") - - # Verify the item exists in the portal - self.assertIn("test-doc", self.portal) - - # Try getting the item from recyclebin (should fail) - response = self.api_session.get(f"/@recyclebin/{self.recycle_id}") - self.assertEqual(response.status_code, 404) - - def test_restore_nonexistent_recyclebin_item(self): - """Test restoring a nonexistent item from recyclebin.""" - response = self.api_session.post("/@recyclebin/non-existent-id/restore") - - self.assertEqual(response.status_code, 404) - data = response.json() - self.assertIn("error", data) - - def test_delete_recyclebin_item(self): - """Test permanently deleting an item from recyclebin.""" - # First, create and delete another item - self.portal.invokeFactory("Document", id="doc-to-purge", title="Document to Purge") - self.portal.manage_delObjects(["doc-to-purge"]) - - # Get the recycle_id for the newly deleted item - items = self.recyclebin.get_items() - purge_id = next((item["recycle_id"] for item in items - if item.get("title") == "Document to Purge"), None) - - if not purge_id: - self.skipTest("Could not find Document to Purge in recyclebin") - - response = self.api_session.delete(f"/@recyclebin/{purge_id}") - - self.assertEqual(response.status_code, 204) - - # Verify item is no longer in recyclebin - response = self.api_session.get(f"/@recyclebin/{purge_id}") - self.assertEqual(response.status_code, 404) - - def test_delete_nonexistent_recyclebin_item(self): - """Test deleting a nonexistent item from recyclebin.""" - response = self.api_session.delete("/@recyclebin/non-existent-id") - - self.assertEqual(response.status_code, 404) - data = response.json() - self.assertIn("error", data) - - def test_empty_recyclebin(self): - """Test emptying the entire recyclebin.""" - # First ensure there's something in the recyclebin - if not self.recyclebin.get_items(): - # Create and delete a new item if recyclebin is empty - self.portal.invokeFactory("Document", id="doc-for-empty", title="Document for Empty") - self.portal.manage_delObjects(["doc-for-empty"]) - - # Get initial count - initial_count = len(self.recyclebin.get_items()) - self.assertGreater(initial_count, 0, "Recyclebin is empty, can't test emptying it") - - # Empty the recyclebin - response = self.api_session.delete("/@recyclebin") - - self.assertEqual(response.status_code, 204) - - # Verify recyclebin is now empty - self.assertEqual(len(self.recyclebin.get_items()), 0) - - def test_empty_recyclebin_with_failures(self): - """Test emptying recyclebin with simulated failures.""" - # This test requires mocking the recyclebin's purge_item method - # to simulate failures. Since mocking is complex in this context, - # we'll provide a stub implementation that you can expand with the - # appropriate mocking library in your actual tests. - - # Skip this test for now - implement with proper mocking in real tests - self.skipTest("This test requires mocking and will be implemented separately") - - # Example of how this would work with mocking: - # with mock.patch.object(self.recyclebin, 'purge_item', side_effect=[True, False, True]): - # response = self.api_session.delete("/@recyclebin") - # self.assertEqual(response.status_code, 207) - # data = response.json() - # self.assertEqual(data["purged"], 2) - # self.assertEqual(len(data["failed"]), 1) - - -class TestRecycleBinRESTAPIIntegration(unittest.TestCase): - """Integration tests for RecycleBin REST API.""" - - layer = PLONE_INTEGRATION_TESTING - - def setUp(self): - self.app = self.layer["app"] - self.portal = self.layer["portal"] - self.portal_url = self.portal.absolute_url() - self.api_session = RelativeSession(self.portal_url) - self.api_session.headers.update({"Accept": "application/json"}) - - # Set up test content and permissions - setRoles(self.portal, TEST_USER_ID, ["Manager"]) - - # Create a folder with content - self.portal.invokeFactory("Folder", id="test-folder", title="Test Folder") - folder = self.portal["test-folder"] - folder.invokeFactory("Document", id="doc1", title="Document 1") - folder.invokeFactory("Document", id="doc2", title="Document 2") - - # Delete the folder to put it and its contents in recyclebin - self.portal.manage_delObjects(["test-folder"]) - - # Get the recycle IDs - self.recyclebin = getUtility(IRecycleBin) - self.items = self.recyclebin.get_items() - self.folder_id = next((item["recycle_id"] for item in self.items - if item.get("title") == "Test Folder"), None) - - def test_restore_folder_hierarchy(self): - """Test restoring a folder with its contents.""" - if not self.folder_id: - self.skipTest("Could not find Test Folder in recyclebin") - - # Restore the folder - response = self.api_session.post(f"/@recyclebin/{self.folder_id}/restore") - - self.assertEqual(response.status_code, 200) - - # Verify the folder exists - self.assertIn("test-folder", self.portal) - folder = self.portal["test-folder"] - - # Verify the folder's contents were restored - self.assertIn("doc1", folder) - self.assertIn("doc2", folder) - - # Verify items are no longer in recyclebin - items = self.recyclebin.get_items() - folder_items = [item for item in items if item.get("title") in ["Test Folder", "Document 1", "Document 2"]] - self.assertEqual(len(folder_items), 0) - - def test_recyclebin_workflow_integration(self): - """Test recyclebin integration with workflow state.""" - # Create a document with a specific workflow state - self.portal.invokeFactory("Document", id="workflow-doc", title="Workflow Document") - doc = self.portal["workflow-doc"] - - # Change workflow state (if your test environment supports this) - try: - workflow_tool = self.portal.portal_workflow - if 'simple_publication_workflow' in workflow_tool.getWorkflowsFor(doc): - workflow_tool.doActionFor(doc, 'publish') - original_state = workflow_tool.getInfoFor(doc, 'review_state') - self.assertEqual(original_state, 'published') - else: - self.skipTest("Required workflow not available") - return - except Exception: - self.skipTest("Workflow transition failed") - return - - # Delete the document - self.portal.manage_delObjects(["workflow-doc"]) - - # Find the document in recyclebin - items = self.recyclebin.get_items() - doc_id = next((item["recycle_id"] for item in items - if item.get("title") == "Workflow Document"), None) - - if not doc_id: - self.skipTest("Could not find Workflow Document in recyclebin") - return - - # Restore the document - self.api_session.post(f"/@recyclebin/{doc_id}/restore") - - # Verify workflow state is preserved (if your implementation preserves it) - doc = self.portal["workflow-doc"] - restored_state = workflow_tool.getInfoFor(doc, 'review_state') - - # This assertion might fail if your recyclebin implementation - # doesn't preserve workflow state - adjust accordingly - self.assertEqual(restored_state, original_state, - "Workflow state should be preserved after restore") - - From 8055417cc3fbd093c39af03dad8ce55fae3447cc Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Sun, 27 Apr 2025 22:07:57 +0530 Subject: [PATCH 3/7] test(recyclebin): add tests for recycle bin REST API functionality --- src/plone/restapi/tests/test_recyclebin.py | 418 +++++++++++++++++++++ 1 file changed, 418 insertions(+) create mode 100644 src/plone/restapi/tests/test_recyclebin.py diff --git a/src/plone/restapi/tests/test_recyclebin.py b/src/plone/restapi/tests/test_recyclebin.py new file mode 100644 index 000000000..618bdf659 --- /dev/null +++ b/src/plone/restapi/tests/test_recyclebin.py @@ -0,0 +1,418 @@ +import unittest +from datetime import datetime +from unittest import mock + +import plone.restapi.testing +from plone.app.testing import setRoles +from plone.app.testing import SITE_OWNER_NAME +from plone.app.testing import SITE_OWNER_PASSWORD +from plone.app.testing import TEST_USER_ID +from plone.restapi.testing import RelativeSession + +import transaction + + +class TestRecycleBin(unittest.TestCase): + layer = plone.restapi.testing.PLONE_RESTAPI_DX_FUNCTIONAL_TESTING + + def setUp(self): + self.portal = self.layer["portal"] + self.request = self.layer["request"] + self.api_session = RelativeSession(self.portal.absolute_url()) + self.api_session.headers.update({"Accept": "application/json"}) + self.api_session.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) + # For POST requests, set Content-Type header + self.api_session.headers.update({"Content-Type": "application/json"}) + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + transaction.commit() + + def tearDown(self): + self.api_session.close() + + # GET tests (from TestRecycleBinGet) + def test_recyclebin_get_disabled(self): + """Test GET /@recyclebin when recycle bin is disabled""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = False + + with mock.patch("plone.restapi.services.recyclebin.get.getUtility", + return_value=recycle_bin): + response = self.api_session.get("/@recyclebin") + + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + + def test_recyclebin_get_enabled_empty(self): + """Test GET /@recyclebin when recycle bin is enabled but empty""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_items.return_value = [] + + with mock.patch("plone.restapi.services.recyclebin.get.getUtility", + return_value=recycle_bin): + response = self.api_session.get("/@recyclebin") + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual(self.portal.absolute_url() + "/@recyclebin", result["@id"]) + self.assertEqual(0, result["items_total"]) + self.assertEqual([], result["items"]) + + def test_recyclebin_get_enabled_with_items(self): + """Test GET /@recyclebin when recycle bin has items""" + sample_date = datetime(2023, 1, 1, 12, 0, 0) + mock_items = [ + { + "id": "document1", + "title": "Test Document", + "type": "Document", + "path": "/plone/document1", + "parent_path": "/plone", + "deletion_date": sample_date, + "size": 1024, + "recycle_id": "123456789", + }, + { + "id": "folder1", + "title": "Test Folder", + "type": "Folder", + "path": "/plone/folder1", + "parent_path": "/plone", + "deletion_date": sample_date, + "size": 512, + "recycle_id": "987654321", + }, + ] + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_items.return_value = mock_items + + with mock.patch("plone.restapi.services.recyclebin.get.getUtility", + return_value=recycle_bin): + response = self.api_session.get("/@recyclebin") + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual(self.portal.absolute_url() + "/@recyclebin", result["@id"]) + self.assertEqual(2, result["items_total"]) + + # Check first item + item1 = result["items"][0] + self.assertEqual("document1", item1["id"]) + self.assertEqual("Test Document", item1["title"]) + self.assertEqual("Document", item1["type"]) + self.assertEqual("/plone/document1", item1["path"]) + self.assertEqual("/plone", item1["parent_path"]) + self.assertEqual(sample_date.isoformat(), item1["deletion_date"]) + self.assertEqual(1024, item1["size"]) + self.assertEqual("123456789", item1["recycle_id"]) + self.assertEqual( + self.portal.absolute_url() + "/@recyclebin/123456789", + item1["@id"] + ) + + # Verify actions + self.assertEqual( + self.portal.absolute_url() + "/@recyclebin-restore", + item1["actions"]["restore"] + ) + self.assertEqual( + self.portal.absolute_url() + "/@recyclebin-purge", + item1["actions"]["purge"] + ) + + # RESTORE tests (from TestRecycleBinRestore) + def test_restore_missing_item_id(self): + """Test restore with missing item_id parameter""" + response = self.api_session.post( + "/@recyclebin-restore", + json={} + ) + + self.assertEqual(400, response.status_code) + self.assertEqual("BadRequest", response.json()["error"]["type"]) + self.assertEqual( + "Missing required parameter: item_id", + response.json()["error"]["message"] + ) + + def test_restore_disabled_recyclebin(self): + """Test restore when recyclebin is disabled""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = False + + with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-restore", + json={"item_id": "123456789"} + ) + + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + + def test_restore_nonexistent_item(self): + """Test restore for a non-existent item""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = None + + with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-restore", + json={"item_id": "nonexistent"} + ) + + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + self.assertEqual( + "Item with ID nonexistent not found in recycle bin", + response.json()["error"]["message"] + ) + + def test_restore_invalid_target_path(self): + """Test restore with an invalid target path""" + item_data = { + "id": "document1", + "title": "Test Document", + "recycle_id": "123456789" + } + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = item_data + + with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-restore", + json={ + "item_id": "123456789", + "target_path": "/non/existent/path" + } + ) + + self.assertEqual(400, response.status_code) + self.assertEqual("BadRequest", response.json()["error"]["type"]) + self.assertEqual( + "Target path /non/existent/path not found", + response.json()["error"]["message"] + ) + + def test_restore_failure(self): + """Test when restore operation fails""" + item_data = { + "id": "document1", + "title": "Test Document", + "recycle_id": "123456789" + } + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = item_data + recycle_bin.restore_item.return_value = None + + with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-restore", + json={"item_id": "123456789"} + ) + + self.assertEqual(500, response.status_code) + self.assertEqual("InternalServerError", response.json()["error"]["type"]) + self.assertEqual("Failed to restore item", response.json()["error"]["message"]) + + def test_restore_success(self): + """Test successful item restoration""" + item_data = { + "id": "document1", + "title": "Test Document", + "recycle_id": "123456789" + } + + # Mock restored object + mock_obj = mock.Mock() + mock_obj.absolute_url.return_value = "http://localhost:8080/plone/document1" + mock_obj.getId.return_value = "document1" + mock_obj.Title.return_value = "Test Document" + mock_obj.portal_type = "Document" + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = item_data + recycle_bin.restore_item.return_value = mock_obj + + with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-restore", + json={"item_id": "123456789"} + ) + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual("success", result["status"]) + self.assertEqual("Item document1 restored successfully", result["message"]) + + # Verify restored item data + restored = result["restored_item"] + self.assertEqual("http://localhost:8080/plone/document1", restored["@id"]) + self.assertEqual("document1", restored["id"]) + self.assertEqual("Test Document", restored["title"]) + self.assertEqual("Document", restored["type"]) + + # PURGE tests (from TestRecycleBinPurge) + def test_purge_missing_parameters(self): + """Test purge with missing parameters""" + response = self.api_session.post( + "/@recyclebin-purge", + json={} + ) + + self.assertEqual(400, response.status_code) + self.assertEqual("BadRequest", response.json()["error"]["type"]) + self.assertEqual( + "Missing required parameter: item_id or purge_all or purge_expired", + response.json()["error"]["message"] + ) + + def test_purge_disabled_recyclebin(self): + """Test purge when recyclebin is disabled""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = False + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"item_id": "123456789"} + ) + + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + + def test_purge_nonexistent_item(self): + """Test purge for a non-existent item""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = None + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"item_id": "nonexistent"} + ) + + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + self.assertEqual( + "Item with ID nonexistent not found in recycle bin", + response.json()["error"]["message"] + ) + + def test_purge_failure(self): + """Test when purge operation fails""" + item_data = { + "id": "document1", + "title": "Test Document", + "recycle_id": "123456789" + } + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = item_data + recycle_bin.purge_item.return_value = False + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"item_id": "123456789"} + ) + + self.assertEqual(500, response.status_code) + self.assertEqual("InternalServerError", response.json()["error"]["type"]) + self.assertEqual("Failed to purge item", response.json()["error"]["message"]) + + def test_purge_success(self): + """Test successful item purge""" + item_data = { + "id": "document1", + "title": "Test Document", + "recycle_id": "123456789" + } + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_item.return_value = item_data + recycle_bin.purge_item.return_value = True + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"item_id": "123456789"} + ) + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual("success", result["status"]) + self.assertEqual("Item document1 purged successfully", result["message"]) + + def test_purge_all(self): + """Test purging all items""" + mock_items = [ + {"id": "document1", "recycle_id": "123"}, + {"id": "document2", "recycle_id": "456"}, + {"id": "document3", "recycle_id": "789"} + ] + + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + recycle_bin.get_items.return_value = mock_items + # Configure purge_item to return True for successful purge + recycle_bin.purge_item.return_value = True + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"purge_all": True} + ) + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual("success", result["status"]) + self.assertEqual(3, result["purged_count"]) + self.assertEqual("Purged 3 items from recycle bin", result["message"]) + # Verify that purge_item was called for each item + recycle_bin.purge_item.assert_any_call("123") + recycle_bin.purge_item.assert_any_call("456") + recycle_bin.purge_item.assert_any_call("789") + + def test_purge_expired(self): + """Test purging expired items""" + recycle_bin = mock.Mock() + recycle_bin.is_enabled.return_value = True + # Configure purge_expired_items to return the number of purged items + recycle_bin.purge_expired_items.return_value = 2 + + with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin): + response = self.api_session.post( + "/@recyclebin-purge", + json={"purge_expired": True} + ) + + self.assertEqual(200, response.status_code) + result = response.json() + self.assertEqual("success", result["status"]) + self.assertEqual(2, result["purged_count"]) + self.assertEqual("Purged 2 expired items from recycle bin", result["message"]) From 76293ec005d2494c58f126354dd6fef1b168adf8 Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Sun, 27 Apr 2025 22:11:03 +0530 Subject: [PATCH 4/7] lint --- .../services/recyclebin/configure.zcml | 61 ++-- src/plone/restapi/services/recyclebin/get.py | 40 +-- .../restapi/services/recyclebin/purge.py | 32 +-- .../restapi/services/recyclebin/restore.py | 31 +- src/plone/restapi/tests/test_recyclebin.py | 266 +++++++++--------- 5 files changed, 217 insertions(+), 213 deletions(-) diff --git a/src/plone/restapi/services/recyclebin/configure.zcml b/src/plone/restapi/services/recyclebin/configure.zcml index 0c13176c2..13b918171 100644 --- a/src/plone/restapi/services/recyclebin/configure.zcml +++ b/src/plone/restapi/services/recyclebin/configure.zcml @@ -1,40 +1,41 @@ + xmlns:zcml="http://namespaces.zope.org/zcml" + > - + - + - + - + - + diff --git a/src/plone/restapi/services/recyclebin/get.py b/src/plone/restapi/services/recyclebin/get.py index f0237ee70..c178e5b15 100644 --- a/src/plone/restapi/services/recyclebin/get.py +++ b/src/plone/restapi/services/recyclebin/get.py @@ -8,7 +8,7 @@ class RecycleBinGet(Service): def reply(self): recycle_bin = getUtility(IRecycleBin) - + # Check if recycle bin is enabled if not recycle_bin.is_enabled(): self.request.response.setStatus(404) @@ -18,31 +18,33 @@ def reply(self): "message": "Recycle Bin is disabled", } } - + # Get all items from recycle bin items = recycle_bin.get_items() - + # Format items for response results = [] for item in items: - results.append({ - "@id": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}", - "id": item["id"], - "title": item["title"], - "type": item["type"], - "path": item["path"], - "parent_path": item["parent_path"], - "deletion_date": item["deletion_date"].isoformat(), - "size": item["size"], - "recycle_id": item["recycle_id"], - "actions": { - "restore": f"{self.context.absolute_url()}/@recyclebin-restore", - "purge": f"{self.context.absolute_url()}/@recyclebin-purge" + results.append( + { + "@id": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}", + "id": item["id"], + "title": item["title"], + "type": item["type"], + "path": item["path"], + "parent_path": item["parent_path"], + "deletion_date": item["deletion_date"].isoformat(), + "size": item["size"], + "recycle_id": item["recycle_id"], + "actions": { + "restore": f"{self.context.absolute_url()}/@recyclebin-restore", + "purge": f"{self.context.absolute_url()}/@recyclebin-purge", + }, } - }) - + ) + return { "@id": f"{self.context.absolute_url()}/@recyclebin", "items": results, - "items_total": len(results) + "items_total": len(results), } diff --git a/src/plone/restapi/services/recyclebin/purge.py b/src/plone/restapi/services/recyclebin/purge.py index 0261aa6c1..2fb1e9367 100644 --- a/src/plone/restapi/services/recyclebin/purge.py +++ b/src/plone/restapi/services/recyclebin/purge.py @@ -9,18 +9,18 @@ class RecycleBinPurge(Service): """POST /@recyclebin-purge - Permanently delete an item from the recycle bin""" - + def reply(self): # Disable CSRF protection for this request alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) - + data = json_body(self.request) item_id = data.get("item_id", None) purge_all = data.get("purge_all", False) purge_expired = data.get("purge_expired", False) - + recycle_bin = getUtility(IRecycleBin) - + # Check if recycle bin is enabled if not recycle_bin.is_enabled(): self.request.response.setStatus(404) @@ -30,30 +30,30 @@ def reply(self): "message": "Recycle Bin is disabled", } } - + # Handle purging all items if purge_all: purged_count = 0 for item in recycle_bin.get_items(): if recycle_bin.purge_item(item["recycle_id"]): purged_count += 1 - + return { "status": "success", "purged_count": purged_count, - "message": f"Purged {purged_count} items from recycle bin" + "message": f"Purged {purged_count} items from recycle bin", } - + # Handle purging expired items if purge_expired: purged_count = recycle_bin.purge_expired_items() - + return { "status": "success", "purged_count": purged_count, - "message": f"Purged {purged_count} expired items from recycle bin" + "message": f"Purged {purged_count} expired items from recycle bin", } - + # Handle purging a specific item if not item_id: self.request.response.setStatus(400) @@ -63,7 +63,7 @@ def reply(self): "message": "Missing required parameter: item_id or purge_all or purge_expired", } } - + # Get the item to purge item_data = recycle_bin.get_item(item_id) if not item_data: @@ -74,10 +74,10 @@ def reply(self): "message": f"Item with ID {item_id} not found in recycle bin", } } - + # Purge the item success = recycle_bin.purge_item(item_id) - + if not success: self.request.response.setStatus(500) return { @@ -86,8 +86,8 @@ def reply(self): "message": "Failed to purge item", } } - + return { "status": "success", - "message": f"Item {item_data['id']} purged successfully" + "message": f"Item {item_data['id']} purged successfully", } diff --git a/src/plone/restapi/services/recyclebin/restore.py b/src/plone/restapi/services/recyclebin/restore.py index 0dd7311b1..e4218fa44 100644 --- a/src/plone/restapi/services/recyclebin/restore.py +++ b/src/plone/restapi/services/recyclebin/restore.py @@ -3,25 +3,24 @@ from Products.CMFPlone.interfaces.recyclebin import IRecycleBin from zope.component import getUtility from zope.interface import alsoProvides -from zope.publisher.interfaces import IPublishTraverse import plone.protect.interfaces class RecycleBinRestore(Service): """POST /@recyclebin-restore - Restore an item from the recycle bin""" - + def __init__(self, context, request): - super(RecycleBinRestore, self).__init__(context, request) + super().__init__(context, request) self.params = {} - + def reply(self): # Disable CSRF protection for this request alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) - + data = json_body(self.request) item_id = data.get("item_id", None) - + if not item_id: self.request.response.setStatus(400) return { @@ -30,9 +29,9 @@ def reply(self): "message": "Missing required parameter: item_id", } } - + recycle_bin = getUtility(IRecycleBin) - + # Check if recycle bin is enabled if not recycle_bin.is_enabled(): self.request.response.setStatus(404) @@ -42,7 +41,7 @@ def reply(self): "message": "Recycle Bin is disabled", } } - + # Get the item to restore item_data = recycle_bin.get_item(item_id) if not item_data: @@ -53,11 +52,11 @@ def reply(self): "message": f"Item with ID {item_id} not found in recycle bin", } } - + # Get optional target container path target_path = data.get("target_path", None) target_container = None - + if target_path: try: portal = self.context.portal_url.getPortalObject() @@ -70,10 +69,10 @@ def reply(self): "message": f"Target path {target_path} not found", } } - + # Restore the item restored_obj = recycle_bin.restore_item(item_id, target_container) - + if not restored_obj: self.request.response.setStatus(500) return { @@ -82,7 +81,7 @@ def reply(self): "message": "Failed to restore item", } } - + self.request.response.setStatus(200) return { "status": "success", @@ -91,6 +90,6 @@ def reply(self): "@id": restored_obj.absolute_url(), "id": restored_obj.getId(), "title": restored_obj.Title(), - "type": restored_obj.portal_type - } + "type": restored_obj.portal_type, + }, } diff --git a/src/plone/restapi/tests/test_recyclebin.py b/src/plone/restapi/tests/test_recyclebin.py index 618bdf659..5553a603b 100644 --- a/src/plone/restapi/tests/test_recyclebin.py +++ b/src/plone/restapi/tests/test_recyclebin.py @@ -1,15 +1,14 @@ -import unittest from datetime import datetime -from unittest import mock - -import plone.restapi.testing from plone.app.testing import setRoles from plone.app.testing import SITE_OWNER_NAME from plone.app.testing import SITE_OWNER_PASSWORD from plone.app.testing import TEST_USER_ID from plone.restapi.testing import RelativeSession +from unittest import mock +import plone.restapi.testing import transaction +import unittest class TestRecycleBin(unittest.TestCase): @@ -25,7 +24,7 @@ def setUp(self): self.api_session.headers.update({"Content-Type": "application/json"}) setRoles(self.portal, TEST_USER_ID, ["Manager"]) transaction.commit() - + def tearDown(self): self.api_session.close() @@ -34,11 +33,12 @@ def test_recyclebin_get_disabled(self): """Test GET /@recyclebin when recycle bin is disabled""" recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = False - - with mock.patch("plone.restapi.services.recyclebin.get.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.get.getUtility", return_value=recycle_bin + ): response = self.api_session.get("/@recyclebin") - + self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) @@ -48,11 +48,12 @@ def test_recyclebin_get_enabled_empty(self): recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_items.return_value = [] - - with mock.patch("plone.restapi.services.recyclebin.get.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.get.getUtility", return_value=recycle_bin + ): response = self.api_session.get("/@recyclebin") - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual(self.portal.absolute_url() + "/@recyclebin", result["@id"]) @@ -84,20 +85,21 @@ def test_recyclebin_get_enabled_with_items(self): "recycle_id": "987654321", }, ] - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_items.return_value = mock_items - - with mock.patch("plone.restapi.services.recyclebin.get.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.get.getUtility", return_value=recycle_bin + ): response = self.api_session.get("/@recyclebin") - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual(self.portal.absolute_url() + "/@recyclebin", result["@id"]) self.assertEqual(2, result["items_total"]) - + # Check first item item1 = result["items"][0] self.assertEqual("document1", item1["id"]) @@ -109,47 +111,42 @@ def test_recyclebin_get_enabled_with_items(self): self.assertEqual(1024, item1["size"]) self.assertEqual("123456789", item1["recycle_id"]) self.assertEqual( - self.portal.absolute_url() + "/@recyclebin/123456789", - item1["@id"] + self.portal.absolute_url() + "/@recyclebin/123456789", item1["@id"] ) - + # Verify actions self.assertEqual( - self.portal.absolute_url() + "/@recyclebin-restore", - item1["actions"]["restore"] + self.portal.absolute_url() + "/@recyclebin-restore", + item1["actions"]["restore"], ) self.assertEqual( - self.portal.absolute_url() + "/@recyclebin-purge", - item1["actions"]["purge"] + self.portal.absolute_url() + "/@recyclebin-purge", item1["actions"]["purge"] ) # RESTORE tests (from TestRecycleBinRestore) def test_restore_missing_item_id(self): """Test restore with missing item_id parameter""" - response = self.api_session.post( - "/@recyclebin-restore", - json={} - ) - + response = self.api_session.post("/@recyclebin-restore", json={}) + self.assertEqual(400, response.status_code) self.assertEqual("BadRequest", response.json()["error"]["type"]) self.assertEqual( - "Missing required parameter: item_id", - response.json()["error"]["message"] + "Missing required parameter: item_id", response.json()["error"]["message"] ) def test_restore_disabled_recyclebin(self): """Test restore when recyclebin is disabled""" recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = False - - with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-restore", - json={"item_id": "123456789"} + "/@recyclebin-restore", json={"item_id": "123456789"} ) - + self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) @@ -159,70 +156,71 @@ def test_restore_nonexistent_item(self): recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = None - - with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-restore", - json={"item_id": "nonexistent"} + "/@recyclebin-restore", json={"item_id": "nonexistent"} ) - + self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) self.assertEqual( - "Item with ID nonexistent not found in recycle bin", - response.json()["error"]["message"] + "Item with ID nonexistent not found in recycle bin", + response.json()["error"]["message"], ) - + def test_restore_invalid_target_path(self): """Test restore with an invalid target path""" item_data = { "id": "document1", "title": "Test Document", - "recycle_id": "123456789" + "recycle_id": "123456789", } - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = item_data - - with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-restore", - json={ - "item_id": "123456789", - "target_path": "/non/existent/path" - } + "/@recyclebin-restore", + json={"item_id": "123456789", "target_path": "/non/existent/path"}, ) - + self.assertEqual(400, response.status_code) self.assertEqual("BadRequest", response.json()["error"]["type"]) self.assertEqual( - "Target path /non/existent/path not found", - response.json()["error"]["message"] + "Target path /non/existent/path not found", + response.json()["error"]["message"], ) - + def test_restore_failure(self): """Test when restore operation fails""" item_data = { "id": "document1", "title": "Test Document", - "recycle_id": "123456789" + "recycle_id": "123456789", } - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = item_data recycle_bin.restore_item.return_value = None - - with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-restore", - json={"item_id": "123456789"} + "/@recyclebin-restore", json={"item_id": "123456789"} ) - + self.assertEqual(500, response.status_code) self.assertEqual("InternalServerError", response.json()["error"]["type"]) self.assertEqual("Failed to restore item", response.json()["error"]["message"]) @@ -232,33 +230,34 @@ def test_restore_success(self): item_data = { "id": "document1", "title": "Test Document", - "recycle_id": "123456789" + "recycle_id": "123456789", } - + # Mock restored object mock_obj = mock.Mock() mock_obj.absolute_url.return_value = "http://localhost:8080/plone/document1" mock_obj.getId.return_value = "document1" mock_obj.Title.return_value = "Test Document" mock_obj.portal_type = "Document" - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = item_data recycle_bin.restore_item.return_value = mock_obj - - with mock.patch("plone.restapi.services.recyclebin.restore.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.restore.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-restore", - json={"item_id": "123456789"} + "/@recyclebin-restore", json={"item_id": "123456789"} ) - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual("success", result["status"]) self.assertEqual("Item document1 restored successfully", result["message"]) - + # Verify restored item data restored = result["restored_item"] self.assertEqual("http://localhost:8080/plone/document1", restored["@id"]) @@ -269,30 +268,28 @@ def test_restore_success(self): # PURGE tests (from TestRecycleBinPurge) def test_purge_missing_parameters(self): """Test purge with missing parameters""" - response = self.api_session.post( - "/@recyclebin-purge", - json={} - ) - + response = self.api_session.post("/@recyclebin-purge", json={}) + self.assertEqual(400, response.status_code) self.assertEqual("BadRequest", response.json()["error"]["type"]) self.assertEqual( - "Missing required parameter: item_id or purge_all or purge_expired", - response.json()["error"]["message"] + "Missing required parameter: item_id or purge_all or purge_expired", + response.json()["error"]["message"], ) def test_purge_disabled_recyclebin(self): """Test purge when recyclebin is disabled""" recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = False - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"item_id": "123456789"} + "/@recyclebin-purge", json={"item_id": "123456789"} ) - + self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) @@ -302,41 +299,43 @@ def test_purge_nonexistent_item(self): recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = None - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"item_id": "nonexistent"} + "/@recyclebin-purge", json={"item_id": "nonexistent"} ) - + self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) self.assertEqual( - "Item with ID nonexistent not found in recycle bin", - response.json()["error"]["message"] + "Item with ID nonexistent not found in recycle bin", + response.json()["error"]["message"], ) - + def test_purge_failure(self): """Test when purge operation fails""" item_data = { "id": "document1", "title": "Test Document", - "recycle_id": "123456789" + "recycle_id": "123456789", } - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = item_data recycle_bin.purge_item.return_value = False - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"item_id": "123456789"} + "/@recyclebin-purge", json={"item_id": "123456789"} ) - + self.assertEqual(500, response.status_code) self.assertEqual("InternalServerError", response.json()["error"]["type"]) self.assertEqual("Failed to purge item", response.json()["error"]["message"]) @@ -346,21 +345,22 @@ def test_purge_success(self): item_data = { "id": "document1", "title": "Test Document", - "recycle_id": "123456789" + "recycle_id": "123456789", } - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_item.return_value = item_data recycle_bin.purge_item.return_value = True - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"item_id": "123456789"} + "/@recyclebin-purge", json={"item_id": "123456789"} ) - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual("success", result["status"]) @@ -371,22 +371,23 @@ def test_purge_all(self): mock_items = [ {"id": "document1", "recycle_id": "123"}, {"id": "document2", "recycle_id": "456"}, - {"id": "document3", "recycle_id": "789"} + {"id": "document3", "recycle_id": "789"}, ] - + recycle_bin = mock.Mock() recycle_bin.is_enabled.return_value = True recycle_bin.get_items.return_value = mock_items # Configure purge_item to return True for successful purge recycle_bin.purge_item.return_value = True - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"purge_all": True} + "/@recyclebin-purge", json={"purge_all": True} ) - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual("success", result["status"]) @@ -403,14 +404,15 @@ def test_purge_expired(self): recycle_bin.is_enabled.return_value = True # Configure purge_expired_items to return the number of purged items recycle_bin.purge_expired_items.return_value = 2 - - with mock.patch("plone.restapi.services.recyclebin.purge.getUtility", - return_value=recycle_bin): + + with mock.patch( + "plone.restapi.services.recyclebin.purge.getUtility", + return_value=recycle_bin, + ): response = self.api_session.post( - "/@recyclebin-purge", - json={"purge_expired": True} + "/@recyclebin-purge", json={"purge_expired": True} ) - + self.assertEqual(200, response.status_code) result = response.json() self.assertEqual("success", result["status"]) From 4f1ad73390dd04803c6355ef2798850bb4a56b90 Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Sun, 27 Apr 2025 22:15:14 +0530 Subject: [PATCH 5/7] changelog --- news/1919.feature | 1 + 1 file changed, 1 insertion(+) create mode 100644 news/1919.feature diff --git a/news/1919.feature b/news/1919.feature new file mode 100644 index 000000000..5fa92b167 --- /dev/null +++ b/news/1919.feature @@ -0,0 +1 @@ +Implement RESTAPI endpoint for managing recycle bin @rohnsha0 \ No newline at end of file From 97aa41fb31fa008062bdac44d7c5a374806b4a13 Mon Sep 17 00:00:00 2001 From: Rohan Shaw <86848116+rohnsha0@users.noreply.github.com> Date: Mon, 28 Apr 2025 06:21:24 +0530 Subject: [PATCH 6/7] Apply suggestions from code review Co-authored-by: Steve Piercy --- news/1919.feature | 2 +- src/plone/restapi/services/recyclebin/get.py | 2 +- src/plone/restapi/services/recyclebin/purge.py | 4 ++-- src/plone/restapi/services/recyclebin/restore.py | 2 +- src/plone/restapi/tests/test_recyclebin.py | 8 ++++---- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/news/1919.feature b/news/1919.feature index 5fa92b167..4f1a01205 100644 --- a/news/1919.feature +++ b/news/1919.feature @@ -1 +1 @@ -Implement RESTAPI endpoint for managing recycle bin @rohnsha0 \ No newline at end of file +Add endpoint for managing recycle bin. @rohnsha0 \ No newline at end of file diff --git a/src/plone/restapi/services/recyclebin/get.py b/src/plone/restapi/services/recyclebin/get.py index c178e5b15..a7a30685f 100644 --- a/src/plone/restapi/services/recyclebin/get.py +++ b/src/plone/restapi/services/recyclebin/get.py @@ -15,7 +15,7 @@ def reply(self): return { "error": { "type": "NotFound", - "message": "Recycle Bin is disabled", + "message": "Recycle bin is disabled", } } diff --git a/src/plone/restapi/services/recyclebin/purge.py b/src/plone/restapi/services/recyclebin/purge.py index 2fb1e9367..a62bc74cc 100644 --- a/src/plone/restapi/services/recyclebin/purge.py +++ b/src/plone/restapi/services/recyclebin/purge.py @@ -27,7 +27,7 @@ def reply(self): return { "error": { "type": "NotFound", - "message": "Recycle Bin is disabled", + "message": "Recycle bin is disabled", } } @@ -60,7 +60,7 @@ def reply(self): return { "error": { "type": "BadRequest", - "message": "Missing required parameter: item_id or purge_all or purge_expired", + "message": "Missing required parameter: item_id, purge_all, or purge_expired", } } diff --git a/src/plone/restapi/services/recyclebin/restore.py b/src/plone/restapi/services/recyclebin/restore.py index e4218fa44..0a38314bd 100644 --- a/src/plone/restapi/services/recyclebin/restore.py +++ b/src/plone/restapi/services/recyclebin/restore.py @@ -38,7 +38,7 @@ def reply(self): return { "error": { "type": "NotFound", - "message": "Recycle Bin is disabled", + "message": "Recycle bin is disabled", } } diff --git a/src/plone/restapi/tests/test_recyclebin.py b/src/plone/restapi/tests/test_recyclebin.py index 5553a603b..056d8933f 100644 --- a/src/plone/restapi/tests/test_recyclebin.py +++ b/src/plone/restapi/tests/test_recyclebin.py @@ -41,7 +41,7 @@ def test_recyclebin_get_disabled(self): self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) - self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + self.assertEqual("Recycle bin is disabled", response.json()["error"]["message"]) def test_recyclebin_get_enabled_empty(self): """Test GET /@recyclebin when recycle bin is enabled but empty""" @@ -149,7 +149,7 @@ def test_restore_disabled_recyclebin(self): self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) - self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + self.assertEqual("Recycle bin is disabled", response.json()["error"]["message"]) def test_restore_nonexistent_item(self): """Test restore for a non-existent item""" @@ -273,7 +273,7 @@ def test_purge_missing_parameters(self): self.assertEqual(400, response.status_code) self.assertEqual("BadRequest", response.json()["error"]["type"]) self.assertEqual( - "Missing required parameter: item_id or purge_all or purge_expired", + "Missing required parameter: item_id, purge_all, or purge_expired", response.json()["error"]["message"], ) @@ -292,7 +292,7 @@ def test_purge_disabled_recyclebin(self): self.assertEqual(404, response.status_code) self.assertEqual("NotFound", response.json()["error"]["type"]) - self.assertEqual("Recycle Bin is disabled", response.json()["error"]["message"]) + self.assertEqual("Recycle bin is disabled", response.json()["error"]["message"]) def test_purge_nonexistent_item(self): """Test purge for a non-existent item""" From 04016ce0ea66c8dd3acf711ecb1f599bfe80a16d Mon Sep 17 00:00:00 2001 From: Rohan Shaw Date: Mon, 28 Apr 2025 06:22:48 +0530 Subject: [PATCH 7/7] add docs --- docs/source/endpoints/index.md | 1 + .../source/endpoints/recycle-bin.md | 12 ++++++------ 2 files changed, 7 insertions(+), 6 deletions(-) rename src/plone/restapi/services/recyclebin/documentation.md => docs/source/endpoints/recycle-bin.md (96%) diff --git a/docs/source/endpoints/index.md b/docs/source/endpoints/index.md index 3d33783b4..8aaefc0b1 100644 --- a/docs/source/endpoints/index.md +++ b/docs/source/endpoints/index.md @@ -42,6 +42,7 @@ portrait principals querystring querystringsearch +recycle-bin registry relations roles diff --git a/src/plone/restapi/services/recyclebin/documentation.md b/docs/source/endpoints/recycle-bin.md similarity index 96% rename from src/plone/restapi/services/recyclebin/documentation.md rename to docs/source/endpoints/recycle-bin.md index 8b9ff9890..930200208 100644 --- a/src/plone/restapi/services/recyclebin/documentation.md +++ b/docs/source/endpoints/recycle-bin.md @@ -1,4 +1,4 @@ -# Recycle Bin REST API +# Recycle Bin The Recycle Bin REST API provides endpoints to interact with the Plone Recycle Bin functionality. @@ -6,7 +6,7 @@ The Recycle Bin REST API provides endpoints to interact with the Plone Recycle B To list all items in the recycle bin, send a GET request to the `@recyclebin` endpoint: -```http +```http-example GET /@recyclebin HTTP/1.1 Accept: application/json ``` @@ -41,7 +41,7 @@ Response: To restore an item from the recycle bin, send a POST request to the `@recyclebin-restore` endpoint: -```http +```http-example POST /@recyclebin-restore HTTP/1.1 Accept: application/json Content-Type: application/json @@ -79,7 +79,7 @@ Response: To permanently delete an item from the recycle bin, send a POST request to the `@recyclebin-purge` endpoint: -```http +```http-example POST /@recyclebin-purge HTTP/1.1 Accept: application/json Content-Type: application/json @@ -102,7 +102,7 @@ Response: To purge all items from the recycle bin: -```http +```http-example POST /@recyclebin-purge HTTP/1.1 Accept: application/json Content-Type: application/json @@ -126,7 +126,7 @@ Response: To purge only expired items (based on the retention period): -```http +```http-example POST /@recyclebin-purge HTTP/1.1 Accept: application/json Content-Type: application/json