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/docs/source/endpoints/recycle-bin.md b/docs/source/endpoints/recycle-bin.md new file mode 100644 index 000000000..a5eb18539 --- /dev/null +++ b/docs/source/endpoints/recycle-bin.md @@ -0,0 +1,126 @@ +--- +myst: + html_meta: + "description": "The Recycle Bin REST API provides endpoints to interact with the Plone Recycle Bin functionality." + "property=og:description": "The Recycle Bin REST API provides endpoints to interact with the Plone Recycle Bin functionality." + "property=og:title": "Recycle Bin" + "keywords": "Plone, plone.restapi, REST, API, Recycle Bin" +--- + +# Recycle bin + +Plone's recycle bin functionality is managed through the `@recyclebin` endpoint. + +Reading or writing recycle bin data requires the `cmf.ManagePortal` permission. + +## List recycle bin contents + +A list of all items in the recycle bin can be retrieved by sending a `GET` request to the `@recyclebin` endpoint: + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_get.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_get.resp +:language: http +``` + +### Filtering and Sorting Parameters + +The listing supports various query parameters for filtering and sorting: + +| Parameter | Description | Example | +|-----------|-------------|---------| +| `title` | Filter by title (case-insensitive substring match) | `title=my doc` | +| `path` | Filter by path (case-insensitive substring match) | `path=/plone/news` | +| `portal_type` | Filter by content type | `portal_type=Document` | +| `date_from` | Filter by deletion date from (YYYY-MM-DD) | `date_from=2024-01-01` | +| `date_to` | Filter by deletion date to (YYYY-MM-DD) | `date_to=2024-12-31` | +| `deleted_by` | Filter by the user ID who deleted the item | `deleted_by=admin` | +| `has_subitems` | Filter items with (`true`) or without (`false`) children | `has_subitems=true` | +| `language` | Filter by language code | `language=it` | +| `review_state` | Filter by workflow state | `review_state=published` | +| `sort_on` | Sort field: `title`, `portal_type`, `path`, `deletion_date`, `review_state` | `sort_on=title` | +| `sort_order` | Sort direction: `ascending` or `descending` (default) | `sort_order=ascending` | + +### Batching + +The API supports standard Plone REST API batching parameters (`b_start`, `b_size`). + +#### Example with filtering and sorting + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_get_filtered.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_get_filtered.resp +:language: http +``` + +## Get individual item from recycle bin + +To retrieve detailed information about a specific item in the recycle bin, including its sub-items, send a `GET` request to `@recyclebin/{item_id}`. +The response includes a paginated `items` list with all flattened descendants. Standard batching parameters (`b_start`, `b_size`) are supported. + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_get_item.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_get_item.resp +:language: http +``` + +## Restore an item from the recycle bin + +An item can be restored to its original location by issuing a `POST` to `@recyclebin/{item_id}/restore`: + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_restore.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_restore.resp +:language: http +``` + +### Restore to a specific location + +Pass a `target_path` in the request body to restore the item to a different folder: + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_restore_target.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_restore_target.resp +:language: http +``` + +## Purge a specific item from the recycle bin + +To permanently delete a specific item, send a `DELETE` request to `@recyclebin/{item_id}`: + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_purge_item.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_purge_item.resp +:language: http +``` + +## Empty the entire recycle bin + +To permanently delete all items, send a `DELETE` request to `@recyclebin`: + +```{eval-rst} +.. http:example:: curl httpie python-requests + :request: ../../../src/plone/restapi/tests/http-examples/recyclebin_purge_all.req +``` + +```{literalinclude} ../../../src/plone/restapi/tests/http-examples/recyclebin_purge_all.resp +:language: http +``` diff --git a/news/1919.feature b/news/1919.feature new file mode 100644 index 000000000..4f1a01205 --- /dev/null +++ b/news/1919.feature @@ -0,0 +1 @@ +Add endpoint for managing recycle bin. @rohnsha0 \ No newline at end of file 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..a000b69d0 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/__init__.py @@ -0,0 +1 @@ +# 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 new file mode 100644 index 000000000..552729529 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/configure.zcml @@ -0,0 +1,33 @@ + + + + + + + + + + + diff --git a/src/plone/restapi/services/recyclebin/get.py b/src/plone/restapi/services/recyclebin/get.py new file mode 100644 index 000000000..197aeb4c4 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/get.py @@ -0,0 +1,206 @@ +from datetime import datetime +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.restapi.batching import HypermediaBatch +from plone.restapi.deserializer import boolean_value +from plone.restapi.services import Service +from zExceptions import BadRequest +from zope.component import getUtility +from zope.interface import implementer +from zope.publisher.interfaces import IPublishTraverse + + +@implementer(IPublishTraverse) +class RecycleBinGet(Service): + """GET /@recyclebin - List items in the recycle bin + GET /@recyclebin/{item_id} - Get a specific item from the recycle bin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.params = [] + + def publishTraverse(self, request, name): + # Consume any path segments after /@recyclebin as parameters + self.params.append(name) + return self + + 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", + } + } + + # If we have a parameter, handle individual item request + if self.params: + return self._reply_individual_item(recycle_bin) + + # Otherwise, handle listing request + return self._reply_listing(recycle_bin) + + def _reply_individual_item(self, recycle_bin): + """Handle GET /@recyclebin/{item_id} - Get a specific item""" + if len(self.params) != 1: + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": "Invalid URL pattern. Expected: /@recyclebin/{item_id}", + } + } + + item_id = self.params[0] + + # Get the specific item from recycle bin + item = recycle_bin.get_item(item_id) + + if item is None: + self.request.response.setStatus(404) + return { + "error": { + "type": "NotFound", + "message": f"Item with ID '{item_id}' not found in recycle bin", + } + } + + # Convert to serializable format with detailed information + children_dict = item.get("children", {}) + + serialized_item = { + "@id": f"{self.context.absolute_url()}/@recyclebin/{item_id}", + "id": item["id"], + "title": item["title"], + "@type": item["portal_type"], + "path": item["path"], + "parent_path": item["parent_path"], + "deletion_date": item["deletion_date"].isoformat(), + "recycle_id": item_id, + "deleted_by": item.get("deleted_by", ""), + "language": item.get("language", ""), + "review_state": item.get("review_state", ""), + "has_children": bool(children_dict), + "actions": { + "restore": f"{self.context.absolute_url()}/@recyclebin/{item_id}/restore", + "purge": f"{self.context.absolute_url()}/@recyclebin/{item_id}", + }, + } + + # Flatten all descendants and apply batching + flattened = list(self._flatten_children(children_dict)) + batch = HypermediaBatch(self.request, flattened) + + serialized_item["items_total"] = batch.items_total + links = batch.links + if links: + serialized_item["batching"] = links + serialized_item["items"] = list(batch) + + return serialized_item + + def _flatten_children(self, children_dict): + """Recursively yield all descendants as a flat sequence.""" + for child_data in children_dict.values(): + entry = { + "id": child_data["id"], + "title": child_data["title"], + "@type": child_data.get("portal_type", "Unknown"), + "path": child_data.get("path", ""), + "language": child_data.get("language", ""), + "review_state": child_data.get("review_state", ""), + "restore_id": child_data.get("restore_id", ""), + } + nested = child_data.get("children", {}) + if isinstance(nested, dict) and nested: + entry["children_count"] = self._count_descendants(nested) + yield entry + if isinstance(nested, dict) and nested: + yield from self._flatten_children(nested) + + def _count_descendants(self, children_dict): + """Recursively count all descendants in a children dict.""" + count = 0 + for child_data in children_dict.values(): + count += 1 + nested = child_data.get("children", {}) + if isinstance(nested, dict) and nested: + count += self._count_descendants(nested) + return count + + def _reply_listing(self, recycle_bin): + """Handle GET /@recyclebin - List items in the recycle bin""" + form = self.request.form + + # Parse date parameters + date_from = date_to = None + if form.get("date_from"): + try: + date_from = datetime.strptime(form["date_from"], "%Y-%m-%d").date() + except ValueError: + raise BadRequest("Invalid date_from format. Expected: YYYY-MM-DD") + if form.get("date_to"): + try: + date_to = datetime.strptime(form["date_to"], "%Y-%m-%d").date() + except ValueError: + raise BadRequest("Invalid date_to format. Expected: YYYY-MM-DD") + + # Parse has_subitems boolean + has_subitems_raw = form.get("has_subitems") + has_subitems = ( + boolean_value(has_subitems_raw) if has_subitems_raw is not None else None + ) + + items = recycle_bin.search( + title=form.get("title") or None, + path=form.get("path") or None, + portal_type=form.get("portal_type") or None, + date_from=date_from, + date_to=date_to, + deleted_by=form.get("deleted_by") or None, + has_subitems=has_subitems, + language=form.get("language") or None, + review_state=form.get("review_state") or None, + sort_on=form.get("sort_on", "deletion_date"), + sort_order=form.get("sort_order", "descending"), + ) + + # Convert to serializable format + serialized_items = [ + { + "@id": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}", + "@type": item["portal_type"], + "id": item["id"], + "title": item["title"], + "path": item["path"], + "parent_path": item["parent_path"], + "deletion_date": item["deletion_date"].isoformat(), + "recycle_id": item["recycle_id"], + "deleted_by": item.get("deleted_by", ""), + "language": item.get("language", ""), + "review_state": item.get("review_state", ""), + "has_children": bool(item.get("children")), + "actions": { + "restore": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}/restore", + "purge": f"{self.context.absolute_url()}/@recyclebin/{item['recycle_id']}", + }, + } + for item in items + ] + + # Apply batching + batch = HypermediaBatch(self.request, serialized_items) + + results = {} + results["@id"] = batch.canonical_url + results["items_total"] = batch.items_total + links = batch.links + if links: + results["batching"] = links + + results["items"] = list(batch) + + return results diff --git a/src/plone/restapi/services/recyclebin/purge.py b/src/plone/restapi/services/recyclebin/purge.py new file mode 100644 index 000000000..2a6114446 --- /dev/null +++ b/src/plone/restapi/services/recyclebin/purge.py @@ -0,0 +1,81 @@ +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.restapi.services import Service +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 RecycleBinPurge(Service): + """DELETE /@recyclebin/{item_id} - Permanently delete an item from the recycle bin + DELETE /@recyclebin - Empty the entire recycle bin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.params = [] + + def publishTraverse(self, request, name): + # Consume any path segments after /@recyclebin as parameters + self.params.append(name) + return self + + def reply(self): + # Disable CSRF protection for this request + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + 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 different cases based on path parameters + if len(self.params) == 0: + # DELETE /@recyclebin - Empty the entire recycle bin + recycle_bin.clear() + + # Return 204 No Content for successful DELETE as per REST conventions + return self.reply_no_content() + elif len(self.params) == 1: + # DELETE /@recyclebin/{item_id} - Delete specific item + return self._purge_single_item(recycle_bin, self.params[0]) + else: + raise BadRequest("Invalid path parameters") + + def _purge_single_item(self, recycle_bin, item_id): + """Purge a single item from the recycle bin""" + # 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 204 No Content for successful DELETE as per REST conventions + return self.reply_no_content() diff --git a/src/plone/restapi/services/recyclebin/restore.py b/src/plone/restapi/services/recyclebin/restore.py new file mode 100644 index 000000000..0b992d40f --- /dev/null +++ b/src/plone/restapi/services/recyclebin/restore.py @@ -0,0 +1,123 @@ +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.restapi.deserializer import json_body +from plone.restapi.services import Service +from zExceptions import BadRequest +from zope.component import getMultiAdapter +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 RecycleBinRestore(Service): + """POST /@recyclebin/{item_id}/restore - Restore an item from the recycle bin""" + + def __init__(self, context, request): + super().__init__(context, request) + self.params = [] + + def publishTraverse(self, request, name): + # Consume any path segments after /@recyclebin as parameters + self.params.append(name) + return self + + def reply(self): + # Disable CSRF protection for this request + alsoProvides(self.request, plone.protect.interfaces.IDisableCSRFProtection) + + # Validate URL pattern: /@recyclebin/{item_id}/restore + if len(self.params) != 2: + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": "Invalid URL pattern. Expected: /@recyclebin/{item_id}/restore", + } + } + + item_id = self.params[0] + action = self.params[1] + + if action != "restore": + self.request.response.setStatus(400) + return { + "error": { + "type": "BadRequest", + "message": "Invalid action. Expected: restore", + } + } + + 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 fields from request body + data = json_body(self.request) + target_path = data.get("target_path", None) if data else None + restore_id = data.get("restore_id", None) if data else None + + if restore_id and not target_path: + raise BadRequest("target_path is required when restoring a child item") + + target_container = None + if target_path: + try: + portal = getMultiAdapter( + (self.context, self.request), name="plone_portal_state" + ).portal() + target_container = portal.unrestrictedTraverse(target_path) + except (KeyError, AttributeError): + raise BadRequest(f"Target path {target_path} not found") + + if restore_id: + restored_obj = recycle_bin.restore_child_item( + item_id, restore_id, target_container + ) + else: + restored_obj = recycle_bin.restore_item(item_id, target_container) + + if isinstance(restored_obj, dict) and not restored_obj.get("success", True): + raise BadRequest(restored_obj.get("error", "Failed to restore item")) + + 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/testing.py b/src/plone/restapi/testing.py index 93e4691e5..27fd9036b 100644 --- a/src/plone/restapi/testing.py +++ b/src/plone/restapi/testing.py @@ -31,9 +31,11 @@ import collective.MockMailHost import os +import random import re import requests import time +import uuid try: from plone.app.caching.testing import PloneAppCachingBase @@ -145,6 +147,10 @@ def setUpPloneSite(self, portal): states = portal.portal_workflow["simple_publication_workflow"].states states["published"].title = "Published with accent é" # noqa: E501 + # Seed random number generator for consistent uuids in tests + rng = random.Random(12345) + uuid.uuid4 = lambda: uuid.UUID(int=rng.getrandbits(128)) + PLONE_RESTAPI_DX_FIXTURE = PloneRestApiDXLayer() PLONE_RESTAPI_DX_INTEGRATION_TESTING = IntegrationTesting( diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get.req b/src/plone/restapi/tests/http-examples/recyclebin_get.req new file mode 100644 index 000000000..297e01159 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get.req @@ -0,0 +1,3 @@ +GET /plone/@recyclebin HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get.resp b/src/plone/restapi/tests/http-examples/recyclebin_get.resp new file mode 100644 index 000000000..c877bb9d9 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get.resp @@ -0,0 +1,45 @@ +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "@id": "http://localhost:55001/plone/@recyclebin", + "items": [ + { + "@id": "http://localhost:55001/plone/@recyclebin/64120e33-5de0-5e85-69d3-a24678a065dc", + "@type": "Folder", + "actions": { + "purge": "http://localhost:55001/plone/@recyclebin/64120e33-5de0-5e85-69d3-a24678a065dc", + "restore": "http://localhost:55001/plone/@recyclebin/64120e33-5de0-5e85-69d3-a24678a065dc/restore" + }, + "deleted_by": "test_user_1_", + "deletion_date": "1995-07-31T17:30:00+00:00", + "has_children": true, + "id": "folder1", + "language": "", + "parent_path": "/plone", + "path": "/plone/folder1", + "recycle_id": "64120e33-5de0-5e85-69d3-a24678a065dc", + "review_state": "private", + "title": "My Folder" + }, + { + "@id": "http://localhost:55001/plone/@recyclebin/639aeb4c-1991-7494-a9d1-f9b2657fe36d", + "@type": "Document", + "actions": { + "purge": "http://localhost:55001/plone/@recyclebin/639aeb4c-1991-7494-a9d1-f9b2657fe36d", + "restore": "http://localhost:55001/plone/@recyclebin/639aeb4c-1991-7494-a9d1-f9b2657fe36d/restore" + }, + "deleted_by": "test_user_1_", + "deletion_date": "1995-07-31T17:30:00+00:00", + "has_children": false, + "id": "document1", + "language": "", + "parent_path": "/plone", + "path": "/plone/document1", + "recycle_id": "639aeb4c-1991-7494-a9d1-f9b2657fe36d", + "review_state": "private", + "title": "My Document" + } + ], + "items_total": 2 +} diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.req b/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.req new file mode 100644 index 000000000..220ffb297 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.req @@ -0,0 +1,3 @@ +GET /plone/@recyclebin?portal_type=Document&sort_on=title&sort_order=ascending HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.resp b/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.resp new file mode 100644 index 000000000..a444681bf --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get_filtered.resp @@ -0,0 +1,45 @@ +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "@id": "http://localhost:55001/plone/@recyclebin?portal_type=Document", + "items": [ + { + "@id": "http://localhost:55001/plone/@recyclebin/8df93d6e-de82-2c67-2919-6e6fca9b0666", + "@type": "Document", + "actions": { + "purge": "http://localhost:55001/plone/@recyclebin/8df93d6e-de82-2c67-2919-6e6fca9b0666", + "restore": "http://localhost:55001/plone/@recyclebin/8df93d6e-de82-2c67-2919-6e6fca9b0666/restore" + }, + "deleted_by": "test_user_1_", + "deletion_date": "1995-07-31T17:30:00+00:00", + "has_children": false, + "id": "document1", + "language": "", + "parent_path": "/plone", + "path": "/plone/document1", + "recycle_id": "8df93d6e-de82-2c67-2919-6e6fca9b0666", + "review_state": "private", + "title": "My Document" + }, + { + "@id": "http://localhost:55001/plone/@recyclebin/d46174f6-045e-f661-eada-a2ce0d342a98", + "@type": "Folder", + "actions": { + "purge": "http://localhost:55001/plone/@recyclebin/d46174f6-045e-f661-eada-a2ce0d342a98", + "restore": "http://localhost:55001/plone/@recyclebin/d46174f6-045e-f661-eada-a2ce0d342a98/restore" + }, + "deleted_by": "test_user_1_", + "deletion_date": "1995-07-31T17:30:00+00:00", + "has_children": true, + "id": "folder1", + "language": "", + "parent_path": "/plone", + "path": "/plone/folder1", + "recycle_id": "d46174f6-045e-f661-eada-a2ce0d342a98", + "review_state": "private", + "title": "My Folder" + } + ], + "items_total": 2 +} diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get_item.req b/src/plone/restapi/tests/http-examples/recyclebin_get_item.req new file mode 100644 index 000000000..334eb4f56 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get_item.req @@ -0,0 +1,3 @@ +GET /plone/@recyclebin/ecf00e14-441d-4264-94de-6bbb9d0c66e7 HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_get_item.resp b/src/plone/restapi/tests/http-examples/recyclebin_get_item.resp new file mode 100644 index 000000000..0655fd011 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_get_item.resp @@ -0,0 +1,23 @@ +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "@id": "http://localhost:55001/plone/@recyclebin/ecf00e14-441d-4264-94de-6bbb9d0c66e7", + "@type": "Document", + "actions": { + "purge": "http://localhost:55001/plone/@recyclebin/ecf00e14-441d-4264-94de-6bbb9d0c66e7", + "restore": "http://localhost:55001/plone/@recyclebin/ecf00e14-441d-4264-94de-6bbb9d0c66e7/restore" + }, + "deleted_by": "test_user_1_", + "deletion_date": "1995-07-31T17:30:00+00:00", + "has_children": false, + "id": "document1", + "items": [], + "items_total": 0, + "language": "", + "parent_path": "/plone", + "path": "/plone/document1", + "recycle_id": "ecf00e14-441d-4264-94de-6bbb9d0c66e7", + "review_state": "private", + "title": "My Document" +} diff --git a/src/plone/restapi/tests/http-examples/recyclebin_purge_all.req b/src/plone/restapi/tests/http-examples/recyclebin_purge_all.req new file mode 100644 index 000000000..7ba6f4d17 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_purge_all.req @@ -0,0 +1,3 @@ +DELETE /plone/@recyclebin HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_purge_all.resp b/src/plone/restapi/tests/http-examples/recyclebin_purge_all.resp new file mode 100644 index 000000000..0074ded3b --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_purge_all.resp @@ -0,0 +1,2 @@ +HTTP/1.1 204 No Content + diff --git a/src/plone/restapi/tests/http-examples/recyclebin_purge_item.req b/src/plone/restapi/tests/http-examples/recyclebin_purge_item.req new file mode 100644 index 000000000..2bbf8a22a --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_purge_item.req @@ -0,0 +1,3 @@ +DELETE /plone/@recyclebin/7b2a4872-a94e-2372-34c2-7fb8f1e38ed1 HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_purge_item.resp b/src/plone/restapi/tests/http-examples/recyclebin_purge_item.resp new file mode 100644 index 000000000..0074ded3b --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_purge_item.resp @@ -0,0 +1,2 @@ +HTTP/1.1 204 No Content + diff --git a/src/plone/restapi/tests/http-examples/recyclebin_restore.req b/src/plone/restapi/tests/http-examples/recyclebin_restore.req new file mode 100644 index 000000000..c474ddef6 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_restore.req @@ -0,0 +1,3 @@ +POST /plone/@recyclebin/b7323098-47c0-4e2f-1ef5-097bc92b8682/restore HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 diff --git a/src/plone/restapi/tests/http-examples/recyclebin_restore.resp b/src/plone/restapi/tests/http-examples/recyclebin_restore.resp new file mode 100644 index 000000000..ad8919791 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_restore.resp @@ -0,0 +1,13 @@ +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "message": "Item folder1 restored successfully", + "restored_item": { + "@id": "http://localhost:55001/plone/folder1", + "@type": "Folder", + "id": "folder1", + "title": "My Folder" + }, + "status": "success" +} diff --git a/src/plone/restapi/tests/http-examples/recyclebin_restore_target.req b/src/plone/restapi/tests/http-examples/recyclebin_restore_target.req new file mode 100644 index 000000000..cad6a740b --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_restore_target.req @@ -0,0 +1,8 @@ +POST /plone/@recyclebin/7cda273b-9e6a-3b17-3b2f-00ac6fd2e766/restore HTTP/1.1 +Accept: application/json +Authorization: Basic YWRtaW46c2VjcmV0 +Content-Type: application/json + +{ + "target_path": "/plone/target_folder" +} diff --git a/src/plone/restapi/tests/http-examples/recyclebin_restore_target.resp b/src/plone/restapi/tests/http-examples/recyclebin_restore_target.resp new file mode 100644 index 000000000..f01f9d8d7 --- /dev/null +++ b/src/plone/restapi/tests/http-examples/recyclebin_restore_target.resp @@ -0,0 +1,13 @@ +HTTP/1.1 200 OK +Content-Type: application/json + +{ + "message": "Item document1 restored successfully", + "restored_item": { + "@id": "http://localhost:55001/plone/target_folder/document1", + "@type": "Document", + "id": "document1", + "title": "My Document" + }, + "status": "success" +} diff --git a/src/plone/restapi/tests/http-examples/registry_get_list.resp b/src/plone/restapi/tests/http-examples/registry_get_list.resp index 5af98dc5d..95402e6f2 100644 --- a/src/plone/restapi/tests/http-examples/registry_get_list.resp +++ b/src/plone/restapi/tests/http-examples/registry_get_list.resp @@ -6,7 +6,7 @@ Content-Type: application/json "batching": { "@id": "http://localhost:55001/plone/@registry", "first": "http://localhost:55001/plone/@registry?b_start=0", - "last": "http://localhost:55001/plone/@registry?b_start=2975", + "last": "http://localhost:55001/plone/@registry?b_start=3000", "next": "http://localhost:55001/plone/@registry?b_start=25" }, "items": [ @@ -423,5 +423,5 @@ Content-Type: application/json "value": "The person that created an item" } ], - "items_total": 3000 + "items_total": 3003 } diff --git a/src/plone/restapi/tests/statictime.py b/src/plone/restapi/tests/statictime.py index c5f76f20a..59feeff23 100644 --- a/src/plone/restapi/tests/statictime.py +++ b/src/plone/restapi/tests/statictime.py @@ -13,6 +13,11 @@ except ImportError: Comment = None +try: + from Products.CMFPlone.recyclebin import RecycleBin +except ImportError: + RecycleBin = None + _originals = { "WorkflowTool.getInfoFor": WorkflowTool.getInfoFor, @@ -20,6 +25,8 @@ "TTWLockable.lock_info": TTWLockable.lock_info, "WorkingCopyInfo.created": WorkingCopyInfo.created, } +if RecycleBin is not None: + _originals["RecycleBin._get_deletion_date"] = RecycleBin._get_deletion_date class StaticTime: @@ -136,6 +143,9 @@ def start(self): WorkingCopyInfo.created = static_wc_info_factory(self.static_created) + if RecycleBin is not None: + RecycleBin._get_deletion_date = lambda x: self.static_modified + def stop(self): """Undo all the patches.""" TTWLockable.lock_info = _originals["TTWLockable.lock_info"] @@ -153,6 +163,9 @@ def stop(self): del DexterityContent.modification_date del DexterityContent.creation_date + if RecycleBin is not None: + RecycleBin._get_deletion_date = _originals["RecycleBin._get_deletion_date"] + def static_get_info_for_factory(dt_value): """Returns a static time replacement for WorkflowTool.getInfoFor diff --git a/src/plone/restapi/tests/test_documentation.py b/src/plone/restapi/tests/test_documentation.py index 379501c7e..a7935c5ca 100644 --- a/src/plone/restapi/tests/test_documentation.py +++ b/src/plone/restapi/tests/test_documentation.py @@ -1901,6 +1901,115 @@ def test_inherit_expansion(self): ) save_request_and_response_for_docs("inherit_expansion", response) + def setUp_recyclebin(self): + """Helper to set up recycle bin for tests""" + # Enable recycle bin + from plone.base.interfaces.recyclebin import IRecycleBinControlPanelSettings + from plone.registry.interfaces import IRegistry + from zope.component import getUtility + + registry = getUtility(IRegistry) + settings = registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + settings.recycling_enabled = True + + # Create some content to delete + self.portal.invokeFactory("Document", id="document1", title="My Document") + self.portal.invokeFactory("Folder", id="folder1", title="My Folder") + self.portal["folder1"].invokeFactory( + "Document", id="subdoc", title="Sub Document" + ) + transaction.commit() + + # Delete the items to put them in recycle bin + self.portal.manage_delObjects(["document1", "folder1"]) + transaction.commit() + + def test_documentation_recyclebin_get(self): + self.setUp_recyclebin() + + # Test getting recycle bin list + response = self.api_session.get("/@recyclebin") + save_request_and_response_for_docs("recyclebin_get", response) + + def test_documentation_recyclebin_get_filtered(self): + self.setUp_recyclebin() + + # This test documents filtering options + response = self.api_session.get( + "/@recyclebin?portal_type=Document&sort_on=title&sort_order=ascending" + ) + save_request_and_response_for_docs("recyclebin_get_filtered", response) + + def test_documentation_recyclebin_get_item(self): + self.setUp_recyclebin() + + # Get the recycle bin to find an item ID + response = self.api_session.get("/@recyclebin") + data = response.json() + if "items" in data and data["items"]: + item_id = data["items"][0]["recycle_id"] + # Test getting specific item + response = self.api_session.get(f"/@recyclebin/{item_id}") + save_request_and_response_for_docs("recyclebin_get_item", response) + + def test_documentation_recyclebin_restore(self): + self.setUp_recyclebin() + + # Get an item to restore + response = self.api_session.get("/@recyclebin") + data = response.json() + if "items" in data and data["items"]: + item_id = data["items"][0]["recycle_id"] + # Test restoring item + response = self.api_session.post(f"/@recyclebin/{item_id}/restore") + save_request_and_response_for_docs("recyclebin_restore", response) + + def test_documentation_recyclebin_restore_target(self): + self.setUp_recyclebin() + + # Create a target folder first + self.portal.invokeFactory("Folder", id="target_folder", title="Target Folder") + transaction.commit() + + # Create and delete another document for this test + self.portal.invokeFactory("Document", id="document2", title="Another Document") + transaction.commit() + self.portal.manage_delObjects(["document2"]) + transaction.commit() + + # Get an item to restore + response = self.api_session.get("/@recyclebin") + data = response.json() + if "items" in data and data["items"]: + item_id = data["items"][0]["recycle_id"] + # Test restoring to specific target + response = self.api_session.post( + f"/@recyclebin/{item_id}/restore", + json={"target_path": "/plone/target_folder"}, + ) + save_request_and_response_for_docs("recyclebin_restore_target", response) + + def test_documentation_recyclebin_purge_item(self): + self.setUp_recyclebin() + + # Get an item to purge + response = self.api_session.get("/@recyclebin") + data = response.json() + if "items" in data and data["items"]: + item_id = data["items"][0]["recycle_id"] + # Test purging specific item + response = self.api_session.delete(f"/@recyclebin/{item_id}") + save_request_and_response_for_docs("recyclebin_purge_item", response) + + def test_documentation_recyclebin_purge_all(self): + self.setUp_recyclebin() + + # Test purging all items + response = self.api_session.delete("/@recyclebin") + save_request_and_response_for_docs("recyclebin_purge_all", response) + class TestDocumentationMessageTranslations(TestDocumentationBase): layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING 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..03c760cf3 --- /dev/null +++ b/src/plone/restapi/tests/test_services_recyclebin.py @@ -0,0 +1,715 @@ +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.app.testing import TEST_USER_PASSWORD +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.base.interfaces.recyclebin import IRecycleBinControlPanelSettings +from plone.registry.interfaces import IRegistry +from plone.restapi.testing import PLONE_RESTAPI_DX_FUNCTIONAL_TESTING +from plone.restapi.testing import RelativeSession +from zope.component import getUtility + +import plone.api as api +import transaction +import unittest + + +class RecycleBinTestBase(unittest.TestCase): + """Base class for recyclebin service tests that handles common setup.""" + + layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING + + def setUp(self): + self.app = self.layer["app"] + self.portal = self.layer["portal"] + self.portal_url = self.portal.absolute_url() + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + + self.api_session = RelativeSession(self.portal_url, test=self) + self.api_session.headers.update({"Accept": "application/json"}) + self.api_session.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) + + # Enable recycle bin + registry = getUtility(IRegistry) + settings = registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + settings.recycling_enabled = True + settings.retention_period = 30 + settings.restore_to_initial_state = False + + # Clear the recycle bin before each test + recyclebin = getUtility(IRecycleBin) + recyclebin.clear() + + transaction.commit() + + def tearDown(self): + self.api_session.close() + + def _delete_to_recyclebin(self, obj): + """Delete obj via manage_delObjects so the event handler populates the bin. + Returns the recycle_id assigned by the event handler.""" + recyclebin = getUtility(IRecycleBin) + items_before = {item["recycle_id"] for item in recyclebin.get_items()} + obj.aq_parent.manage_delObjects([obj.getId()]) + transaction.commit() + items_after = {item["recycle_id"] for item in recyclebin.get_items()} + new_ids = items_after - items_before + return new_ids.pop() if new_ids else None + + def _add_document_to_recyclebin(self, doc_id="test-doc", doc_title="Test Document"): + """Helper: create a document, delete it so the event handler puts it in the + recyclebin, and return (recycle_id, original_title).""" + self.portal.invokeFactory("Document", doc_id, title=doc_title) + doc = self.portal[doc_id] + recycle_id = self._delete_to_recyclebin(doc) + return recycle_id, doc_title + + def _add_folder_to_recyclebin( + self, folder_id="test-folder", folder_title="Test Folder" + ): + """Helper: create a folder with a child document, delete it so the event + handler puts it in the recyclebin, and return (recycle_id, original_title).""" + self.portal.invokeFactory("Folder", folder_id, title=folder_title) + folder = self.portal[folder_id] + folder.invokeFactory("Document", "child-doc", title="Child Document") + recycle_id = self._delete_to_recyclebin(folder) + return recycle_id, folder_title + + def _add_nested_folder_to_recyclebin(self): + """Helper: create a 3-level deep folder tree and delete it. + + Structure: + level-0/ (Folder) + doc-0 (Document) + level-1/ (Folder) + doc-1 (Document) + level-2/ (Folder) + doc-2 (Document) + + Returns (recycle_id, "Level 0"). + """ + self.portal.invokeFactory("Folder", "level-0", title="Level 0") + f0 = self.portal["level-0"] + f0.invokeFactory("Document", "doc-0", title="Doc 0") + f0.invokeFactory("Folder", "level-1", title="Level 1") + f1 = f0["level-1"] + f1.invokeFactory("Document", "doc-1", title="Doc 1") + f1.invokeFactory("Folder", "level-2", title="Level 2") + f2 = f1["level-2"] + f2.invokeFactory("Document", "doc-2", title="Doc 2") + recycle_id = self._delete_to_recyclebin(f0) + return recycle_id, "Level 0" + + def _disable_recyclebin(self): + """Helper: disable the recycle bin and commit.""" + registry = getUtility(IRegistry) + settings = registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + settings.recycling_enabled = False + transaction.commit() + + +class TestRecycleBinGET(RecycleBinTestBase): + """Tests for GET /@recyclebin (list) and GET /@recyclebin/{item_id} (single item).""" + + # ------------------------------------------------------------------ + # Listing tests + # ------------------------------------------------------------------ + + def test_get_empty_recyclebin_returns_200(self): + """GET /@recyclebin on empty bin returns 200 with empty items list.""" + response = self.api_session.get("/@recyclebin") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertEqual(0, data["items_total"]) + self.assertEqual([], data["items"]) + + def test_get_listing_returns_expected_keys(self): + """GET /@recyclebin listing contains required top-level keys.""" + response = self.api_session.get("/@recyclebin") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertIn("@id", data) + self.assertIn("items_total", data) + self.assertIn("items", data) + + def test_get_listing_with_items(self): + """GET /@recyclebin lists deleted items.""" + self._add_document_to_recyclebin() + response = self.api_session.get("/@recyclebin") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertEqual(1, len(data["items"])) + + def test_get_listing_item_structure(self): + """Each item in GET /@recyclebin listing contains the expected keys.""" + self._add_document_to_recyclebin(doc_title="My Page") + response = self.api_session.get("/@recyclebin") + item = response.json()["items"][0] + expected_keys = { + "@id", + "@type", + "id", + "title", + "path", + "parent_path", + "deletion_date", + "recycle_id", + "deleted_by", + "language", + "review_state", + "has_children", + "actions", + } + self.assertEqual(expected_keys, set(item.keys())) + + def test_get_listing_item_title_matches(self): + """Item title in listing matches the original document title.""" + self._add_document_to_recyclebin(doc_title="My Special Page") + response = self.api_session.get("/@recyclebin") + item = response.json()["items"][0] + self.assertEqual("My Special Page", item["title"]) + + def test_get_listing_item_has_actions(self): + """Items in listing expose restore and purge action URLs.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.get("/@recyclebin") + item = response.json()["items"][0] + self.assertIn("restore", item["actions"]) + self.assertIn("purge", item["actions"]) + self.assertIn(recycle_id, item["actions"]["restore"]) + self.assertIn(recycle_id, item["actions"]["purge"]) + + def test_get_listing_folder_has_children_flag(self): + """Folder with children has has_children=True in listing.""" + self._add_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin") + item = response.json()["items"][0] + self.assertTrue(item["has_children"]) + + def test_get_listing_document_has_no_children(self): + """Document entry has has_children=False in listing.""" + self._add_document_to_recyclebin() + response = self.api_session.get("/@recyclebin") + item = response.json()["items"][0] + self.assertFalse(item["has_children"]) + + # ------------------------------------------------------------------ + # Filtering tests + # ------------------------------------------------------------------ + + def test_filter_by_type(self): + """portal_type parameter returns items of matching portal_type, including + parents whose children contain the specified type.""" + self._add_document_to_recyclebin(doc_id="doc1", doc_title="Doc 1") + self._add_folder_to_recyclebin(folder_id="folder1", folder_title="Folder 1") + response = self.api_session.get("/@recyclebin?portal_type=Document") + data = response.json() + # Doc1 matches directly; Folder1 matches because its child is a Document. + self.assertEqual(2, data["items_total"]) + types_returned = {item["@type"] for item in data["items"]} + self.assertIn("Document", types_returned) + + def test_filter_by_search_query_title(self): + """title parameter filters by title.""" + self._add_document_to_recyclebin(doc_id="doc-alpha", doc_title="Alpha Document") + self._add_document_to_recyclebin(doc_id="doc-beta", doc_title="Beta Document") + response = self.api_session.get("/@recyclebin?title=alpha") + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertIn("Alpha", data["items"][0]["title"]) + + def test_filter_by_has_subitems_with_subitems(self): + """has_subitems=true returns only items with children.""" + self._add_document_to_recyclebin() + self._add_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin?has_subitems=true") + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertTrue(data["items"][0]["has_children"]) + + def test_filter_by_has_subitems_without_subitems(self): + """has_subitems=false returns only items without children.""" + self._add_document_to_recyclebin() + self._add_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin?has_subitems=false") + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertFalse(data["items"][0]["has_children"]) + + def test_filter_no_match_returns_empty(self): + """Filtering with a portal_type that doesn't exist returns empty items list.""" + self._add_document_to_recyclebin() + response = self.api_session.get("/@recyclebin?portal_type=NonExistentType") + data = response.json() + self.assertEqual(0, data["items_total"]) + self.assertEqual([], data["items"]) + + def test_filter_title_matches_nested_child(self): + """title filter matches a deeply nested child and surfaces the parent.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin?title=Doc+2") + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertEqual(recycle_id, data["items"][0]["recycle_id"]) + + def test_filter_title_no_match_in_children(self): + """title filter that doesn't match root or any child returns nothing.""" + self._add_nested_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin?title=Nonexistent+XYZ") + data = response.json() + self.assertEqual(0, data["items_total"]) + + def test_filter_portal_type_matches_nested_child(self): + """portal_type filter matches a child type and surfaces the parent.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + # The nested structure contains Documents; root is a Folder. + response = self.api_session.get("/@recyclebin?portal_type=Document") + data = response.json() + self.assertEqual(1, data["items_total"]) + self.assertEqual(recycle_id, data["items"][0]["recycle_id"]) + + def test_filter_portal_type_no_match_in_children(self): + """portal_type filter that matches neither root nor any child returns nothing.""" + self._add_nested_folder_to_recyclebin() + response = self.api_session.get("/@recyclebin?portal_type=Event") + data = response.json() + self.assertEqual(0, data["items_total"]) + + def test_filter_invalid_date_from_returns_bad_request(self): + """A malformed date_from returns 400 BadRequest.""" + response = self.api_session.get("/@recyclebin?date_from=not-a-date") + self.assertEqual(400, response.status_code) + self.assertIn("date_from", response.json()["message"]) + + def test_filter_invalid_date_to_returns_bad_request(self): + """A malformed date_to returns 400 BadRequest.""" + response = self.api_session.get("/@recyclebin?date_to=31/12/2024") + self.assertEqual(400, response.status_code) + self.assertIn("date_to", response.json()["message"]) + + # ------------------------------------------------------------------ + # Sorting tests + # ------------------------------------------------------------------ + + def test_sort_by_title_asc(self): + """sort_on=title&sort_order=ascending returns items alphabetically ascending by title.""" + self._add_document_to_recyclebin(doc_id="doc-z", doc_title="Zebra") + self._add_document_to_recyclebin(doc_id="doc-a", doc_title="Apple") + response = self.api_session.get( + "/@recyclebin?sort_on=title&sort_order=ascending" + ) + titles = [item["title"] for item in response.json()["items"]] + self.assertEqual(["Apple", "Zebra"], titles) + + def test_sort_by_title_desc(self): + """sort_on=title&sort_order=descending returns items alphabetically descending by title.""" + self._add_document_to_recyclebin(doc_id="doc-z2", doc_title="Zebra") + self._add_document_to_recyclebin(doc_id="doc-a2", doc_title="Apple") + response = self.api_session.get( + "/@recyclebin?sort_on=title&sort_order=descending" + ) + titles = [item["title"] for item in response.json()["items"]] + self.assertEqual(["Zebra", "Apple"], titles) + + # ------------------------------------------------------------------ + # Individual item tests + # ------------------------------------------------------------------ + + def test_get_individual_item_by_id(self): + """GET /@recyclebin/{id} returns the specific deleted item.""" + recycle_id, _title = self._add_document_to_recyclebin(doc_title="Single Item") + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertEqual(recycle_id, data["recycle_id"]) + self.assertEqual("Single Item", data["title"]) + + def test_get_individual_item_structure(self): + """GET /@recyclebin/{id} contains the expected keys.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + data = response.json() + expected_keys = { + "@id", + "id", + "title", + "@type", + "path", + "parent_path", + "deletion_date", + "recycle_id", + "deleted_by", + "language", + "review_state", + "has_children", + "items_total", + "items", + "actions", + } + self.assertEqual(expected_keys, set(data.keys())) + + def test_get_individual_item_not_found_returns_404(self): + """GET /@recyclebin/{non-existent-id} returns 404.""" + response = self.api_session.get("/@recyclebin/this-id-does-not-exist") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + def test_get_individual_item_folder_with_children(self): + """GET /@recyclebin/{id} for a folder shows children details.""" + recycle_id, _title = self._add_folder_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertTrue(data["has_children"]) + self.assertGreater(data["items_total"], 0) + self.assertGreater(len(data["items"]), 0) + + def test_get_individual_item_children_structure(self): + """Children entries in GET /@recyclebin/{id} have expected keys.""" + recycle_id, _title = self._add_folder_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + child = response.json()["items"][0] + for key in ( + "id", + "title", + "@type", + "path", + "restore_id", + "language", + "review_state", + ): + self.assertIn(key, child) + + def test_get_individual_item_nested_children_flattened(self): + """GET /@recyclebin/{id} flattens all descendants.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + self.assertEqual(200, response.status_code) + data = response.json() + + items = data["items"] + ids = [c["id"] for c in items] + + # All 5 descendants must be present (doc-0, level-1, doc-1, level-2, doc-2) + self.assertEqual(5, len(items)) + for expected_id in ("doc-0", "level-1", "doc-1", "level-2", "doc-2"): + self.assertIn(expected_id, ids) + + # items_total on root should be total descendants = 5 + self.assertEqual(5, data["items_total"]) + + def test_get_individual_item_nested_children_have_restore_id(self): + """All descendants in flattened items have a non-empty restore_id.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + for child in response.json()["items"]: + self.assertTrue( + child["restore_id"], f"Missing restore_id for {child['id']}" + ) + + def test_get_individual_item_folders_have_children_count(self): + """Folder descendants include children_count for their sub-tree.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + response = self.api_session.get(f"/@recyclebin/{recycle_id}") + by_id = {c["id"]: c for c in response.json()["items"]} + + # level-1 has 3 descendants: doc-1, level-2, doc-2 + self.assertEqual(3, by_id["level-1"]["children_count"]) + # level-2 has 1 descendant: doc-2 + self.assertEqual(1, by_id["level-2"]["children_count"]) + # Documents should not have children_count + self.assertNotIn("children_count", by_id["doc-0"]) + + def test_get_individual_item_items_batching(self): + """GET /@recyclebin/{id}?b_size=2 paginates flattened descendants.""" + recycle_id, _title = self._add_nested_folder_to_recyclebin() + # 5 descendants total; request first page of 2 + response = self.api_session.get(f"/@recyclebin/{recycle_id}?b_size=2") + self.assertEqual(200, response.status_code) + data = response.json() + self.assertEqual(5, data["items_total"]) + self.assertEqual(2, len(data["items"])) + self.assertIn("batching", data) + + # ------------------------------------------------------------------ + # Disabled recycle bin + # ------------------------------------------------------------------ + + def test_get_listing_disabled_recyclebin_returns_404(self): + """GET /@recyclebin returns 404 when recycle bin is disabled.""" + self._disable_recyclebin() + response = self.api_session.get("/@recyclebin") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + def test_get_item_disabled_recyclebin_returns_404(self): + """GET /@recyclebin/{id} returns 404 when recycle bin is disabled.""" + self._disable_recyclebin() + response = self.api_session.get("/@recyclebin/some-id") + self.assertEqual(404, response.status_code) + + +class TestRecycleBinPurge(RecycleBinTestBase): + """Tests for DELETE /@recyclebin/{item_id} and DELETE /@recyclebin.""" + + def test_purge_single_item_returns_204(self): + """DELETE /@recyclebin/{id} returns 204 No Content.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.delete(f"/@recyclebin/{recycle_id}") + self.assertEqual(204, response.status_code) + + def test_purge_single_item_removes_item(self): + """After DELETE /@recyclebin/{id} the item is no longer in the bin.""" + recycle_id, _title = self._add_document_to_recyclebin() + self.api_session.delete(f"/@recyclebin/{recycle_id}") + transaction.abort() # sync with what the WSGI request committed + recyclebin = getUtility(IRecycleBin) + self.assertIsNone(recyclebin.get_item(recycle_id)) + + def test_purge_nonexistent_item_returns_404(self): + """DELETE /@recyclebin/{non-existent-id} returns 404.""" + response = self.api_session.delete("/@recyclebin/this-id-does-not-exist") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + def test_purge_all_returns_204(self): + """DELETE /@recyclebin (no item_id) empties the bin and returns 204.""" + self._add_document_to_recyclebin(doc_id="doc-purge-1", doc_title="D1") + self._add_document_to_recyclebin(doc_id="doc-purge-2", doc_title="D2") + response = self.api_session.delete("/@recyclebin") + self.assertEqual(204, response.status_code) + + def test_purge_all_empties_bin(self): + """After DELETE /@recyclebin the recycle bin is empty.""" + self._add_document_to_recyclebin(doc_id="doc-empty-1", doc_title="E1") + self._add_document_to_recyclebin(doc_id="doc-empty-2", doc_title="E2") + self.api_session.delete("/@recyclebin") + transaction.abort() # sync with what the WSGI request committed + recyclebin = getUtility(IRecycleBin) + self.assertEqual(0, len(recyclebin.get_items())) + + def test_purge_disabled_recyclebin_returns_404(self): + """DELETE /@recyclebin returns 404 when recycle bin is disabled.""" + self._disable_recyclebin() + response = self.api_session.delete("/@recyclebin") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + def test_purge_item_disabled_recyclebin_returns_404(self): + """DELETE /@recyclebin/{id} returns 404 when recycle bin is disabled.""" + self._disable_recyclebin() + response = self.api_session.delete("/@recyclebin/some-id") + self.assertEqual(404, response.status_code) + + +class TestRecycleBinRestore(RecycleBinTestBase): + """Tests for POST /@recyclebin/{item_id}/restore.""" + + def test_restore_item_returns_200(self): + """POST /@recyclebin/{id}/restore returns 200 on success.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + self.assertEqual(200, response.status_code) + + def test_restore_item_response_structure(self): + """POST /@recyclebin/{id}/restore response contains status and restored_item.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + data = response.json() + self.assertEqual("success", data["status"]) + self.assertIn("message", data) + self.assertIn("restored_item", data) + + def test_restore_item_restored_item_keys(self): + """Restored item in response contains @id, id, title, @type.""" + recycle_id, _title = self._add_document_to_recyclebin(doc_title="Restored Doc") + response = self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + restored = response.json()["restored_item"] + self.assertIn("@id", restored) + self.assertIn("id", restored) + self.assertIn("title", restored) + self.assertIn("@type", restored) + + def test_restore_item_title_matches(self): + """Restored item title matches the original document title.""" + recycle_id, _title = self._add_document_to_recyclebin( + doc_title="My Restored Doc" + ) + response = self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + self.assertEqual("My Restored Doc", response.json()["restored_item"]["title"]) + + def test_restore_item_appears_in_portal(self): + """After restore, the document exists back in the portal.""" + recycle_id, _title = self._add_document_to_recyclebin(doc_id="restore-me") + response = self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + self.assertEqual(200, response.status_code) + restored_id = response.json()["restored_item"]["id"] + transaction.abort() # sync with what the WSGI request committed + self.assertIn(restored_id, self.portal) + + def test_restore_removes_item_from_recyclebin(self): + """After restore, the item is no longer in the recycle bin.""" + recycle_id, _title = self._add_document_to_recyclebin() + self.api_session.post(f"/@recyclebin/{recycle_id}/restore") + transaction.abort() # sync with what the WSGI request committed + recyclebin = getUtility(IRecycleBin) + self.assertIsNone(recyclebin.get_item(recycle_id)) + + def test_restore_nonexistent_item_returns_404(self): + """POST /@recyclebin/{non-existent-id}/restore returns 404.""" + response = self.api_session.post("/@recyclebin/does-not-exist/restore") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + def test_restore_with_invalid_action_returns_400(self): + """POST /@recyclebin/{id}/not-restore returns 400 bad request.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.post(f"/@recyclebin/{recycle_id}/somethingelse") + self.assertEqual(400, response.status_code) + + def test_restore_missing_action_segment_returns_400(self): + """POST /@recyclebin/{id} (missing /restore) returns 400.""" + recycle_id, _title = self._add_document_to_recyclebin() + response = self.api_session.post(f"/@recyclebin/{recycle_id}") + self.assertEqual(400, response.status_code) + + def test_restore_to_different_target_path(self): + """POST /@recyclebin/{id}/restore with target_path restores to given folder.""" + # Create a target folder (not deleted, so it stays in the portal) + self.portal.invokeFactory("Folder", "target-folder", title="Target Folder") + transaction.commit() + + recycle_id, _title = self._add_document_to_recyclebin(doc_id="doc-to-move") + + target_path = "/".join(self.portal["target-folder"].getPhysicalPath()) + # Strip the site root prefix and leading slash so it's a portal-relative path + site_path = "/".join(self.portal.getPhysicalPath()) + relative_target = target_path[len(site_path) :].lstrip("/") + + response = self.api_session.post( + f"/@recyclebin/{recycle_id}/restore", + json={"target_path": relative_target}, + ) + self.assertEqual(200, response.status_code) + restored_id = response.json()["restored_item"]["id"] + transaction.abort() # sync with what the WSGI request committed + self.assertIn(restored_id, self.portal["target-folder"]) + + def test_restore_disabled_recyclebin_returns_404(self): + """POST /@recyclebin/{id}/restore returns 404 when recycle bin is disabled.""" + self._disable_recyclebin() + response = self.api_session.post("/@recyclebin/some-id/restore") + self.assertEqual(404, response.status_code) + self.assertEqual("NotFound", response.json()["error"]["type"]) + + +class TestRecycleBinPermissions(unittest.TestCase): + """Tests that @recyclebin endpoints require cmf.ManagePortal permission. + + Anonymous users should get 401; authenticated users without ManagePortal + should get 403. + """ + + layer = PLONE_RESTAPI_DX_FUNCTIONAL_TESTING + + def setUp(self): + self.app = self.layer["app"] + self.portal = self.layer["portal"] + self.portal_url = self.portal.absolute_url() + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + + # Enable recycle bin + registry = getUtility(IRegistry) + settings = registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + settings.recycling_enabled = True + + # Create a regular user without ManagePortal + api.user.create( + email="editor@example.com", + username="editor-user", + password=TEST_USER_PASSWORD, + ) + + self.api_session = RelativeSession(self.portal_url, test=self) + self.api_session.headers.update({"Accept": "application/json"}) + self.api_session.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) + + transaction.commit() + + def tearDown(self): + self.api_session.close() + + # ------------------------------------------------------------------ + # Anonymous (no credentials) → 401 + # ------------------------------------------------------------------ + + def test_anonymous_get_listing_returns_401(self): + """Anonymous cannot list the recycle bin.""" + anon_session = RelativeSession(self.portal_url, test=self) + anon_session.headers.update({"Accept": "application/json"}) + response = anon_session.get("/@recyclebin") + self.assertEqual(401, response.status_code) + anon_session.close() + + def test_anonymous_get_item_returns_401(self): + """Anonymous cannot retrieve a single recycle bin item.""" + anon_session = RelativeSession(self.portal_url, test=self) + anon_session.headers.update({"Accept": "application/json"}) + response = anon_session.get("/@recyclebin/some-id") + self.assertEqual(401, response.status_code) + anon_session.close() + + def test_anonymous_purge_returns_401(self): + """Anonymous cannot purge the recycle bin.""" + anon_session = RelativeSession(self.portal_url, test=self) + anon_session.headers.update({"Accept": "application/json"}) + response = anon_session.delete("/@recyclebin") + self.assertEqual(401, response.status_code) + anon_session.close() + + def test_anonymous_restore_returns_401(self): + """Anonymous cannot restore an item from the recycle bin.""" + anon_session = RelativeSession(self.portal_url, test=self) + anon_session.headers.update({"Accept": "application/json"}) + response = anon_session.post("/@recyclebin/some-id/restore") + self.assertEqual(401, response.status_code) + anon_session.close() + + # ------------------------------------------------------------------ + # Regular editor (Member + Editor, no ManagePortal) → 401 + # Zope challenges with 401 even for authenticated users when the + # permission check is handled purely by ZCML (no explicit code-level + # Forbidden raise). + # ------------------------------------------------------------------ + + def test_editor_get_listing_returns_401(self): + """Editor without ManagePortal gets 401 (Zope permission challenge).""" + self.api_session.auth = ("editor-user", TEST_USER_PASSWORD) + response = self.api_session.get("/@recyclebin") + self.assertEqual(401, response.status_code) + + def test_editor_get_item_returns_401(self): + """Editor without ManagePortal gets 401 on single item request.""" + self.api_session.auth = ("editor-user", TEST_USER_PASSWORD) + response = self.api_session.get("/@recyclebin/some-id") + self.assertEqual(401, response.status_code) + + def test_editor_purge_returns_401(self): + """Editor without ManagePortal gets 401 on purge.""" + self.api_session.auth = ("editor-user", TEST_USER_PASSWORD) + response = self.api_session.delete("/@recyclebin") + self.assertEqual(401, response.status_code) + + def test_editor_restore_returns_401(self): + """Editor without ManagePortal gets 401 on restore.""" + self.api_session.auth = ("editor-user", TEST_USER_PASSWORD) + response = self.api_session.post("/@recyclebin/some-id/restore") + self.assertEqual(401, response.status_code)