diff --git a/news/2966.feature b/news/2966.feature new file mode 100644 index 0000000000..30d4db4c5b --- /dev/null +++ b/news/2966.feature @@ -0,0 +1 @@ +Added recycle bin feature. @rohnsha0 \ No newline at end of file diff --git a/src/Products/CMFPlone/browser/configure.zcml b/src/Products/CMFPlone/browser/configure.zcml index 23ecdcfdd3..1d5c614f79 100644 --- a/src/Products/CMFPlone/browser/configure.zcml +++ b/src/Products/CMFPlone/browser/configure.zcml @@ -305,4 +305,26 @@ permission="zope2.View" /> + + + + + + + diff --git a/src/Products/CMFPlone/browser/recyclebin.py b/src/Products/CMFPlone/browser/recyclebin.py new file mode 100644 index 0000000000..49764d4b08 --- /dev/null +++ b/src/Products/CMFPlone/browser/recyclebin.py @@ -0,0 +1,1016 @@ +from datetime import datetime +from plone.base import PloneMessageFactory as _ +from plone.base.batch import Batch +from plone.base.interfaces.recyclebin import IRecycleBin +from Products.CMFCore.utils import getToolByName +from Products.Five.browser import BrowserView +from Products.Five.browser.pagetemplatefile import ViewPageTemplateFile +from Products.statusmessages.interfaces import IStatusMessage +from urllib.parse import urlencode +from z3c.form import button +from z3c.form import field +from z3c.form import form +from zExceptions import NotFound +from zope import schema +from zope.component import getMultiAdapter +from zope.component import getUtility +from zope.i18n import translate +from zope.interface import implementer +from zope.interface import Interface +from zope.publisher.interfaces import IPublishTraverse + +import logging + +logger = logging.getLogger(__name__) + + +class IRecycleBinForm(Interface): + """Schema for the recycle bin form""" + + selected_items = schema.List( + title=_("Selected Items"), + description=_("Selected items for operations"), + value_type=schema.TextLine(), + required=False, + ) + + +class IRecycleBinItemForm(Interface): + """Schema for the recycle bin item form""" + + target_container = schema.TextLine( + title=_("Target container"), + description=_( + "Enter the path to the container where the item should be restored (e.g., /folder1/subfolder)" + ), + required=False, + ) + + +def _is_error_result(result): + """Helper method to check if a result is an error dictionary + + Args: + result: The result to check + + Returns: + Boolean indicating if the result is an error dictionary + """ + return isinstance(result, dict) and not result.get("success", True) + + +class RecycleBinWorkflowMixin: + """Mixin class providing common workflow state methods for recycle bin views""" + + def _get_workflow_state(self, item): + """Get the workflow state that the item had when it was deleted + + Args: + item: The recycled item data dictionary + + Returns: + String representing the workflow state or None + """ + # Try to get the object from the item + obj = item.get("object") + if not obj: + # For RecycleBinView, we need to get the full item data first + if hasattr(self, "recycle_bin") and "recycle_id" in item: + full_item_data = self.recycle_bin.get_item(item.get("recycle_id")) + if full_item_data: + obj = full_item_data.get("object") + + if not obj: + return None + + # Try to get the workflow state from the object + try: + # Get workflow tool + workflow_tool = getToolByName(self.context, "portal_workflow") + + # Get the workflow state + return workflow_tool.getInfoFor(obj, "review_state", None) + + except Exception as e: + logger.warning( + f"Could not determine workflow state for item {item.get('id')}: {e}" + ) + return None + + def get_workflow_state_title(self, state, portal_type=None): + """Get user-friendly title for workflow state + + Args: + state: The workflow state ID + portal_type: The portal type of the object (optional) + + Returns: + Human-readable title for the state + """ + if not state: + return translate(_("Unknown"), context=self.request) + + workflow_tool = getToolByName(self.context, "portal_workflow") + title = workflow_tool.getTitleForStateOnType(state, portal_type) + return translate(title, context=self.request) + + def get_workflow_state_class(self, state): + """Get CSS class for workflow state badge + + Args: + state: The workflow state ID + + Returns: + CSS class string for styling the state badge + """ + if not state: + return "bg-secondary text-white" + + # Color coding for different states + state_classes = { + "private": "bg-danger text-white", + "published": "bg-success text-white", + "pending": "bg-warning text-dark", + "visible": "bg-info text-white", + "internal": "bg-primary text-white", + "draft": "bg-secondary text-white", + "review": "bg-warning text-dark", + "rejected": "bg-danger text-white", + "external": "bg-dark text-white", + "retracted": "bg-secondary text-white", + } + + return state_classes.get(state, "bg-light text-dark") + + def _flatten_children(self, children_dict, depth=0): + """Recursively yield all descendants as a flat sequence. + + Each entry is a copy of the child data dict (without the nested + 'children' key) augmented with a 'depth' field for indentation. + Nodes that have sub-children also get a 'children_count' field. + """ + for child_data in children_dict.values(): + entry = {k: v for k, v in child_data.items() if k != "children"} + entry["restore_id"] = child_data.get("restore_id", "") + entry["depth"] = depth + 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, depth + 1) + + 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 + + +class RecycleBinView(RecycleBinWorkflowMixin, form.Form): + """Form view for recycle bin management""" + + ignoreContext = True + template = ViewPageTemplateFile("templates/recyclebin.pt") + + # Add an ID for the form + id = "recyclebin-form" + + def __init__(self, context, request): + super().__init__(context, request) + self.recycle_bin = getUtility(IRecycleBin) + self._batch = None + + @button.buttonAndHandler(_("Restore Selected"), name="restore") + def handle_restore(self, action): + """Restore selected items handler""" + data, errors = self.extractData() + + # Get the selected items from the request directly + selected_items = self.request.form.get("selected_items", []) + if not isinstance(selected_items, list): + selected_items = [selected_items] + + if not selected_items: + message = translate( + _("No items selected for restoration."), context=self.request + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + return + + restored_count = 0 + missing_parents_count = 0 + missing_parent_items = [] + + for item_id in selected_items: + result = self.recycle_bin.restore_item(item_id) + + # Handle different types of return values + if _is_error_result(result): + # This is a failed restoration with an error message + missing_parents_count += 1 + # Get the item title for better user feedback + item_data = self.recycle_bin.get_item(item_id) + if item_data: + missing_parent_items.append( + { + "id": item_id, + "title": item_data.get("title", "Unknown"), + "parent_path": item_data.get("parent_path", "Unknown"), + "error": result.get("error", "Unknown error"), + } + ) + elif result: + # Successful restoration + restored_count += 1 + + # Success message for restored items + if restored_count > 0: + message = translate( + _( + "${count} item(s) restored successfully.", + mapping={"count": restored_count}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + + # Error message for items with missing parents + if missing_parents_count > 0: + if len(missing_parent_items) == 1: + # Single item message + item = missing_parent_items[0] + message = translate( + _( + "The item '${title}' could not be restored because its original location no longer exists. " + "Please choose a different location.", + mapping={"title": item["title"]}, + ), + context=self.request, + ) + # Redirect to the item's detail page if only one item had this issue + self.request.response.redirect( + f"{self.context.absolute_url()}/@@recyclebin-item/{item['id']}" + ) + else: + # Multiple items message + message = translate( + _( + "${count} items could not be restored because their original locations no longer exist. " + "Please visit each item's detail page to specify a new location.", + mapping={"count": missing_parents_count}, + ), + context=self.request, + ) + + IStatusMessage(self.request).addStatusMessage(message, type="error") + + @button.buttonAndHandler(_("Delete selected"), name="delete") + def handle_delete(self, action): + """Delete selected items handler""" + data, errors = self.extractData() + + # Get the selected items from the request directly + selected_items = self.request.form.get("selected_items", []) + if not isinstance(selected_items, list): + selected_items = [selected_items] + + if not selected_items: + message = translate( + _("No items selected for deletion."), context=self.request + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + return + + deleted_count = 0 + for item_id in selected_items: + if self.recycle_bin.purge_item(item_id): + deleted_count += 1 + + message = translate( + _( + "${count} item(s) permanently deleted.", + mapping={"count": deleted_count}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + + @button.buttonAndHandler(_("Empty Recycle Bin"), name="empty") + def handle_empty(self, action): + """Empty recycle bin handler""" + data, errors = self.extractData() + + # Get count before clearing for the status message + items = self.recycle_bin.get_items() + deleted_count = len(items) + + self.recycle_bin.clear() + + message = translate( + _( + "Recycle bin emptied. ${count} item(s) permanently deleted.", + mapping={"count": deleted_count}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + + def get_search_query(self): + """Get the search query from the request""" + return self.request.form.get("search_query", "") + + def get_sort_option(self): + """Get the current sort option from the request""" + return self.request.form.get("sort_by", "date_desc") + + def get_filter_type(self): + """Get the content type filter from the request""" + return self.request.form.get("filter_type", "") + + def get_date_from(self): + """Get the start date filter from the request""" + date_str = self.request.form.get("date_from", "") + if date_str: + return datetime.strptime(date_str, "%Y-%m-%d").date() + + return None + + def get_date_to(self): + """Get the end date filter from the request""" + date_str = self.request.form.get("date_to", "") + if date_str: + return datetime.strptime(date_str, "%Y-%m-%d").date() + + return None + + def get_date_from_str(self): + """Get the start date filter as string from the request""" + return self.request.form.get("date_from", "") + + def get_date_to_str(self): + """Get the end date filter as string from the request""" + return self.request.form.get("date_to", "") + + def get_filter_deleted_by(self): + """Get the deleted by user filter from the request""" + return self.request.form.get("filter_deleted_by", "") + + def get_filter_has_subitems(self): + """Get the has sub-items filter from the request""" + return self.request.form.get("filter_has_subitems", "") + + def get_filter_language(self): + """Get the language filter from the request""" + return self.request.form.get("filter_language", "") + + def get_filter_workflow_state(self): + """Get the workflow state filter from the request""" + return self.request.form.get("filter_workflow_state", "") + + def get_b_start(self): + """Get the batch start index from the request""" + return int(self.request.form.get("b_start", 0)) + + def get_b_size(self): + """Get the batch size from the request (default 20)""" + return int(self.request.form.get("b_size", 20)) + + def get_batch(self): + """Get a batch of items for pagination""" + if self._batch is None: + # Get all items first (this applies filters and sorting) + all_items = self.get_items() + + # Create batch with pagination + b_start = self.get_b_start() + b_size = self.get_b_size() + + self._batch = Batch(all_items, size=b_size, start=b_start, orphan=1) + + return self._batch + + def get_page_size_options(self): + """Get available page size options""" + return [10, 20, 50, 100] + + def get_sort_labels(self): + """Get a dictionary of human-readable sort option labels""" + return { + "date_desc": _("Newest first (default)"), + "date_asc": _("Oldest first"), + "title_asc": _("Title (A-Z)"), + "title_desc": _("Title (Z-A)"), + "type_asc": _("Type (A-Z)"), + "type_desc": _("Type (Z-A)"), + "path_asc": _("Path (A-Z)"), + "path_desc": _("Path (Z-A)"), + "workflow_asc": _("Workflow state (A-Z)"), + "workflow_desc": _("Workflow state (Z-A)"), + } + + def get_clear_url(self, param_to_remove): + """Generate a URL that clears a specific filter parameter while preserving others + + Args: + param_to_remove: The parameter name to remove from the URL + + Returns: + URL string with the specified parameter removed + """ + base_url = f"{self.context.absolute_url()}/@@recyclebin" + params = {} + + # Add search query if it exists and is not being removed + if param_to_remove != "search_query" and self.get_search_query(): + params["search_query"] = self.get_search_query() + + # Add filter type if it exists and is not being removed + if param_to_remove != "filter_type" and self.get_filter_type(): + params["filter_type"] = self.get_filter_type() + + # Add date from filter if it exists and is not being removed + date_from = self.get_date_from() + if param_to_remove != "date_from" and date_from: + params["date_from"] = date_from.strftime("%Y-%m-%d") + + # Add date to filter if it exists and is not being removed + date_to = self.get_date_to() + if param_to_remove != "date_to" and date_to: + params["date_to"] = date_to.strftime("%Y-%m-%d") + + # Add deleted by filter if it exists and is not being removed + if param_to_remove != "filter_deleted_by" and self.get_filter_deleted_by(): + params["filter_deleted_by"] = self.get_filter_deleted_by() + + # Add has sub-items filter if it exists and is not being removed + if param_to_remove != "filter_has_subitems" and self.get_filter_has_subitems(): + params["filter_has_subitems"] = self.get_filter_has_subitems() + + # Add language filter if it exists and is not being removed + if param_to_remove != "filter_language" and self.get_filter_language(): + params["filter_language"] = self.get_filter_language() + + # Add workflow state filter if it exists and is not being removed + if ( + param_to_remove != "filter_workflow_state" + and self.get_filter_workflow_state() + ): + params["filter_workflow_state"] = self.get_filter_workflow_state() + + # Add sort option if it exists, is not default, and is not being removed + sort_option = self.get_sort_option() + if param_to_remove != "sort_by" and sort_option != "date_desc": + params["sort_by"] = sort_option + + # Construct final URL using urlencode for proper URL encoding + if params: + query_string = urlencode(params) + return f"{base_url}?{query_string}" + return base_url + + def get_available_types(self, items): + """Get a list of all content types present in the recycle bin""" + types = set() + for item in items: + item_type = item.get("portal_type") + if item_type: + types.add(item_type) + return sorted(list(types)) + + def get_available_deleted_by_users(self, items): + """Get a list of all users who have deleted items in the recycle bin""" + users = set() + for item in items: + deleted_by = item.get("deleted_by") + if deleted_by: + users.add(deleted_by) + return sorted(list(users)) + + def get_available_languages(self, items): + """Get a list of all languages present in the recycle bin""" + languages = set() + for item in items: + language = item.get("language") + if language: + languages.add(language) + return sorted(list(languages)) + + def get_available_workflow_states(self, items): + """Get a list of all workflow states present in the recycle bin""" + states = set() + for item in items: + workflow_state = item.get("review_state") + if workflow_state: + states.add(workflow_state) + return sorted(list(states)) + + def _check_item_matches_search(self, item, search_query): + """Check if an item matches the search query. + + Args: + item: The item to check + search_query: The search query string (lowercase) + + Returns: + Boolean indicating if the item matches + """ + # Search in title + if search_query in item.get("title", "").lower(): + return True + + # Search in path + if search_query in item.get("path", "").lower(): + return True + + # Search in parent path + if search_query in item.get("parent_path", "").lower(): + return True + + # Search in ID + if search_query in item.get("id", "").lower(): + return True + + # Search in type + if search_query in item.get("portal_type", "").lower(): + return True + + return False + + def _check_item_matches_date_range(self, item, date_from, date_to): + """Check if an item's deletion date falls within the specified date range. + + Args: + item: The item to check + date_from: Start date as date object or None + date_to: End date as date object or None + + Returns: + Boolean indicating if the item matches the date range + """ + if not date_from and not date_to: + return True # No date filter applied + + deletion_date = item.get("deletion_date") + if not deletion_date: + return False # Can't filter items without deletion date + + # Convert deletion_date to date object for comparison + if hasattr(deletion_date, "date"): + item_date = deletion_date.date() + else: + # If it's already a date object + item_date = deletion_date + + # Check date range (dates are already parsed) + if date_from and item_date < date_from: + return False + + if date_to and item_date > date_to: + return False + + return True + + def _find_matching_children(self, item, search_query): + """Find children of an item that match the search query. + + Args: + item: The parent item to check children of + search_query: The search query string (lowercase) + + Returns: + List of matching children or None if no matches + """ + if "children" in item and isinstance(item["children"], dict): + child_matches = [] + + for child_id, child_data in item["children"].items(): + # Check each child for matches + if ( + search_query in child_data.get("title", "").lower() + or search_query in child_data.get("path", "").lower() + or search_query in child_data.get("id", "").lower() + or search_query in child_data.get("portal_type", "").lower() + ): + child_matches.append(child_data) + + if child_matches: + return child_matches + + return None + + def _apply_sorting(self, items, sort_option): + """Apply sorting to the items list. + + Args: + items: List of items to sort + sort_option: The sort option to apply + + Returns: + Sorted list of items + """ + if sort_option == "title_asc": + items.sort(key=lambda x: x.get("title", "").lower()) + elif sort_option == "title_desc": + items.sort(key=lambda x: x.get("title", "").lower(), reverse=True) + elif sort_option == "type_asc": + items.sort(key=lambda x: x.get("type", "").lower()) + elif sort_option == "type_desc": + items.sort(key=lambda x: x.get("type", "").lower(), reverse=True) + elif sort_option == "path_asc": + items.sort(key=lambda x: x.get("path", "").lower()) + elif sort_option == "path_desc": + items.sort(key=lambda x: x.get("path", "").lower(), reverse=True) + elif sort_option == "size_asc": + items.sort(key=lambda x: x.get("size", 0)) + elif sort_option == "size_desc": + items.sort(key=lambda x: x.get("size", 0), reverse=True) + elif sort_option == "date_asc": + items.sort(key=lambda x: x.get("deletion_date", datetime.now())) + elif sort_option == "workflow_asc": + items.sort(key=lambda x: (x.get("workflow_state") or "").lower()) + elif sort_option == "workflow_desc": + items.sort( + key=lambda x: (x.get("workflow_state") or "").lower(), reverse=True + ) + else: + # Default: date_desc + items.sort( + key=lambda x: x.get("deletion_date", datetime.now()), reverse=True + ) + return items + + def get_items(self): + """Get all items in the recycle bin""" + items = self.recycle_bin.get_items() + + # Get filters early to avoid multiple lookups during the loop + filter_type = self.get_filter_type() + search_query = self.get_search_query().lower() + date_from = self.get_date_from() + date_to = self.get_date_to() + filter_deleted_by = self.get_filter_deleted_by() + filter_has_subitems = self.get_filter_has_subitems() + filter_language = self.get_filter_language() + filter_workflow_state = self.get_filter_workflow_state() + + # Create a list of all items that are children of a parent in the recycle bin + child_items_to_exclude = [] + for item in items: + # If this item is a parent with children, add its children to exclusion list + if "children" in item: + for child_id in item.get("children", {}): + child_items_to_exclude.append(child_id) + + logger.debug(f"Child items to exclude: {child_items_to_exclude}") + + # Process items with direct matches + filtered_items = [] + items_with_matching_children = [] + + for item in items: + if item.get("id") not in child_items_to_exclude: + # Apply type filtering + if filter_type and item.get("portal_type") != filter_type: + continue + + # Apply date range filtering + if not self._check_item_matches_date_range(item, date_from, date_to): + continue + + # Apply deleted by filtering + if filter_deleted_by and item.get("deleted_by") != filter_deleted_by: + continue + + # Apply has sub-items filtering + if filter_has_subitems: + has_children = bool(item.get("children")) + if filter_has_subitems == "with_subitems" and not has_children: + continue + elif filter_has_subitems == "without_subitems" and has_children: + continue + + # Apply language filtering + if filter_language and item.get("language") != filter_language: + continue + + # Apply workflow state filtering + if ( + filter_workflow_state + and item.get("review_state") != filter_workflow_state + ): + continue + + # Add children count information + if "children" in item: + item["children_count"] = self._count_descendants(item["children"]) + + # Apply search query filtering + if search_query: + # Check for direct matches + if self._check_item_matches_search(item, search_query): + filtered_items.append(item) + continue + + # Check for matches in children + matching_children = self._find_matching_children(item, search_query) + if matching_children: + # Make a copy of the item so we don't modify the original + parent_item = item.copy() + parent_item["matching_children"] = matching_children + parent_item["matching_children_count"] = len(matching_children) + items_with_matching_children.append(parent_item) + else: + # No search query, include all items + filtered_items.append(item) + + # Combine results based on whether we're searching or not + if search_query: + items = filtered_items + items_with_matching_children + else: + items = filtered_items + + # Apply sorting + return self._apply_sorting(items, self.get_sort_option()) + + +@implementer(IPublishTraverse) +class RecycleBinItemView(RecycleBinWorkflowMixin, form.Form): + """View for managing individual recycled items""" + + ignoreContext = True + template = ViewPageTemplateFile("templates/recyclebin_item.pt") + item_id = None + fields = field.Fields(IRecycleBinItemForm) + + def __init__(self, context, request): + super().__init__(context, request) + self.recycle_bin = getUtility(IRecycleBin) + + def publishTraverse(self, request, name): + """Handle URLs like /recyclebin-item/[item_id]""" + logger.debug(f"RecycleBinItemView.publishTraverse called with name: {name}") + if self.item_id is None: # First traversal + self.item_id = name + logger.debug(f"Set item_id to: {self.item_id}") + return self + logger.debug(f"Additional traversal attempted with name: {name}") + raise NotFound(self, name, request) + + def update(self): + super().update() + + # Check if we have a valid item before proceeding + if self.item_id is None: + logger.debug("No item_id set, redirecting to main recyclebin view") + self.request.response.redirect( + f"{self.context.absolute_url()}/@@recyclebin" + ) + return + + # Handle restoration of children + if "restore.child" in self.request.form: + self._handle_child_restoration() + + @button.buttonAndHandler(_("Restore item"), name="restore") + def handle_restore(self, action): + """Restore this item""" + data, errors = self.extractData() + if errors: + return + + # Get target container if specified + target_path = data.get("target_container", "") + target_container = None + + if target_path: + try: + target_container = self.context.unrestrictedTraverse(target_path) + except (KeyError, AttributeError): + message = translate( + _( + "The folder '${path}' where you are trying to restore this item cannot be found. It may have been moved or deleted. Please choose a different location.", + mapping={"path": target_path}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + return + + # Restore the item + item = self.get_item() + if not item: + message = translate( + _("Item not found. It may have been already restored or deleted."), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + self.request.response.redirect( + f"{self.context.absolute_url()}/@@recyclebin" + ) + return + + result = self.recycle_bin.restore_item(self.item_id, target_container) + + # Check if we got an error dictionary using helper method + if _is_error_result(result): + # Show the error message + error_message = result.get( + "error", "Unknown error occurred during restoration" + ) + IStatusMessage(self.request).addStatusMessage(error_message, type="error") + + # Redirect back to the item view to allow selecting a different target + self.request.response.redirect( + f"{self.context.absolute_url()}/@@recyclebin-item/{self.item_id}" + ) + return + + restored_obj = result + + if restored_obj: + message = translate( + _( + "Item '${title}' successfully restored.", + mapping={"title": restored_obj.Title()}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + + # Determine the appropriate URL to redirect to after restoration + context_state = getMultiAdapter( + (restored_obj, self.request), name="plone_context_state" + ) + redirect_url = context_state.view_url() + + self.request.response.redirect(redirect_url) + else: + message = translate( + _( + "Failed to restore item. It may have been already restored or deleted." + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + self.request.response.redirect( + f"{self.context.absolute_url()}/@@recyclebin" + ) + + @button.buttonAndHandler(_("Permanently delete"), name="delete") + def handle_delete(self, action): + """Permanently delete this item""" + data, errors = self.extractData() + + # Get item info before deletion + item = self.get_item() + if item: + item_title = item.get("title", "Unknown") + + if self.recycle_bin.purge_item(self.item_id): + message = translate( + _( + "Item '${title}' permanently deleted.", + mapping={"title": item_title}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + else: + message = translate( + _( + "Failed to delete item '${title}'.", + mapping={"title": item_title}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + else: + message = translate( + _("Item not found. It may have been already deleted."), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + + self.request.response.redirect(f"{self.context.absolute_url()}/@@recyclebin") + + def _handle_child_restoration(self): + """Restore a child item using the recycle bin tool.""" + restore_id = self.request.form.get("restore_id") + target_path = (self.request.form.get("target_path") or "").strip() + + if not restore_id: + message = translate( + _("Missing child identifier. Please refresh and try again."), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + return + + if not target_path: + item_data = self.recycle_bin.get_item(self.item_id) or {} + children = item_data.get("children", {}) + child_match = self.recycle_bin._find_child_by_restore_id( + children, restore_id + ) + child_data = child_match[0] + target_path = (child_data or {}).get("parent_path", "") + + if not target_path: + message = translate( + _( + "Could not determine the original location for this child item. " + "Please enter a target path." + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + return + + try: + target_container = self.context.unrestrictedTraverse(target_path) + except (KeyError, AttributeError): + message = translate( + _("Target location not found: ${path}", mapping={"path": target_path}), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + return + + try: + result = self.recycle_bin.restore_child_item( + self.item_id, + restore_id=restore_id, + target_container=target_container, + ) + except Exception as e: + logger.error(f"Error restoring child item: {e}") + message = translate( + _("Failed to restore child item."), context=self.request + ) + IStatusMessage(self.request).addStatusMessage(message, type="error") + return + + if _is_error_result(result): + IStatusMessage(self.request).addStatusMessage( + result.get("error", "Unknown error during child restoration"), + type="error", + ) + return + + restored_obj = result + if restored_obj: + # Derive the child id from the restored object so the redirect is accurate + obj_id = restored_obj.getId() + child_data = self.recycle_bin.get_item(self.item_id) or {} + child_title = child_data.get("title", obj_id) + message = translate( + _( + "Child item '${title}' successfully restored.", + mapping={"title": child_title}, + ), + context=self.request, + ) + IStatusMessage(self.request).addStatusMessage(message, type="info") + self.request.response.redirect( + f"{target_container.absolute_url()}/{obj_id}" + ) + + def get_item(self): + """Get the specific recycled item""" + if not self.item_id: + return None + + item = self.recycle_bin.get_item(self.item_id) + if item is None: + logger.debug(f"No item found in recycle bin with ID: {self.item_id}") + else: + logger.debug( + f"Found item: {item.get('title', 'Unknown')} of type {item.get('type', 'Unknown')}" + ) + # Add children count information (total descendants, not just direct children) + if "children" in item: + item["children_count"] = self._count_descendants(item["children"]) + + return item + + def get_children(self): + """Get all descendants of this item as a flat list with depth metadata.""" + item = self.get_item() + + if item and "children" in item: + return list(self._flatten_children(item["children"], depth=0)) + return [] + + +class RecycleBinEnabled(BrowserView): + """View to check if the recycle bin is enabled""" + + def __call__(self): + """Return True if the recycle bin is enabled, False otherwise""" + recycle_bin = getUtility(IRecycleBin) + return recycle_bin.is_enabled() diff --git a/src/Products/CMFPlone/browser/templates/recyclebin.pt b/src/Products/CMFPlone/browser/templates/recyclebin.pt new file mode 100644 index 0000000000..d362307dd0 --- /dev/null +++ b/src/Products/CMFPlone/browser/templates/recyclebin.pt @@ -0,0 +1,1124 @@ + + + + + + + + + + + +
+ +
+
+

Recycle bin

+
+ Items deleted from this site are stored here, and can be restored or permanently deleted. +
+
+
+ + +
+
+ Search and Filter +
+
+
+
+ +
+ +
+ + +
+ + +
+ +
+ + +
+ + +
+ + +
+
+ + +
+
+ + +
+
+
+ + +
+
+ +
+ + +
+ + +
+ + +
+
+ + +
+
+ + +
+ +
+ + +
+
+ +
+
+ + + + +
+
+
+ + +
+ +
+ Active filters: + + + query + + × + + + + + Document + + × + + + + + From: + 2024-01-01 + + × + + + + + To: + 2024-12-31 + + × + + + + + Deleted by: + username + + × + + + + + Sub-items filter + + × + + + + + Language: + en + + × + + + + + Workflow: + Published + + × + + + + + Sort option + + × + + +
+ + + Clear all + +
+ + +
+
+
+
+ Deleted items + + (1-20 + of + 50) + +
+ + +
+ + + + + + + + + + + + + + +
+ + +
+ + + + + + + + + + + + + +
+
+
+ + +
+ + + +
+
+ + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+ + +
+ 		TitleTypeWorkflow stateOriginal pathDeletion dateDeleted by
+
+

+ No items match your search criteria +

+

+ No items in recycle bin +

+

Items that are deleted will appear here

+
+
+
+ + +
+ 		+ Title + +
+ + Contains + 0 + item + items +
+ + +
+ + Found matches in contents: + 3 + items match your search + +
    +
  • + Child title + + (Type) + +
    • +
  • + and + + more... +
    • +
+
+ 		+ Content type + + Published + Unknown + + Original path + + + Deletion date + + Deleted by +
+
+
+ + +
+
+
+ + +
+
+
+ + + +
+
+
+
+ + +
+
+

+ No items match your search criteria +

+

+ No items in recycle bin +

+

Items that are deleted will appear here

+
+
+
+
+ + + + + + + + + + diff --git a/src/Products/CMFPlone/browser/templates/recyclebin_item.pt b/src/Products/CMFPlone/browser/templates/recyclebin_item.pt new file mode 100644 index 0000000000..1e04b42f0e --- /dev/null +++ b/src/Products/CMFPlone/browser/templates/recyclebin_item.pt @@ -0,0 +1,431 @@ + + + + + + + + +
+ + +
+
+ Item not found +
+
+
+

The requested item was not found

+

+ The requested item was not found in the recycle bin. It may have been already restored or deleted. +

+
+ + Return to recycle bin + +
+
+ + +
+ +
+
+ + Back to recycle bin + +

+ Title + Content type +

+
+
+ + + +
+ +
+
+
+ Item details +
+
+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
Original IDID
Workflow state + Published + Unknown +
Original path + Path +
Parent path + Parent +
Deletion date + Date +
Deleted by + User +
Number of items + Count + contained items +
+
+
+
+
+ + +
+
+
+ Restore item +
+
+
+ + + + +
+ +
+ The original location no longer exists. Please choose a new location to restore this item. +
+
+ +
+ + +
+
+
+
+
+ Deleted on: + date +
+
+
+
+
+ + + +
+
+
+
+ Folder contents +
+ 0 +
+
+
+
+ + These items were contained in this folder when it was deleted. + You can restore them individually to any location. + +
+ +
+ + + + + + + + + + + + + + + + + + + +
TitleTypeWorkflow stateOriginal pathActions
+ + + Title + +
+ + Contains + 0 + item + items +
+ 		+ Type + + Published + Unknown + + Path + +
+ +
+ + +
+ +
+
+ +
+
+
+
+
+
+
+ + + +
+ +
+ + + + + + + + diff --git a/src/Products/CMFPlone/configure.zcml b/src/Products/CMFPlone/configure.zcml index e730a7bc46..806c96d8b0 100644 --- a/src/Products/CMFPlone/configure.zcml +++ b/src/Products/CMFPlone/configure.zcml @@ -165,4 +165,17 @@ for="zope.pagetemplate.engine.ZopeBaseEngine" /> + + + + + + diff --git a/src/Products/CMFPlone/controlpanel/browser/configure.zcml b/src/Products/CMFPlone/controlpanel/browser/configure.zcml index 7cca9f0b3a..bf8889e33a 100644 --- a/src/Products/CMFPlone/controlpanel/browser/configure.zcml +++ b/src/Products/CMFPlone/controlpanel/browser/configure.zcml @@ -348,4 +348,13 @@ permission="cmf.ManagePortal" /> + + + + diff --git a/src/Products/CMFPlone/controlpanel/browser/recyclebin.py b/src/Products/CMFPlone/controlpanel/browser/recyclebin.py new file mode 100644 index 0000000000..f2c61c4eac --- /dev/null +++ b/src/Products/CMFPlone/controlpanel/browser/recyclebin.py @@ -0,0 +1,17 @@ +from plone.app.registry.browser.controlpanel import ControlPanelFormWrapper +from plone.app.registry.browser.controlpanel import RegistryEditForm +from plone.base import PloneMessageFactory as _ +from plone.base.interfaces.recyclebin import IRecycleBinControlPanelSettings +from plone.z3cform import layout + + +class RecyclebinControlPanelForm(RegistryEditForm): + schema = IRecycleBinControlPanelSettings + schema_prefix = "recyclebin-controlpanel" + label = _("Recycle bin settings") + description = _("Settings for the Plone recycle bin") + + +RecyclebinControlPanelView = layout.wrap_form( + RecyclebinControlPanelForm, ControlPanelFormWrapper +) diff --git a/src/Products/CMFPlone/events.py b/src/Products/CMFPlone/events.py index 77baf900f6..3428fbaeea 100644 --- a/src/Products/CMFPlone/events.py +++ b/src/Products/CMFPlone/events.py @@ -1,8 +1,13 @@ from plone.base.interfaces import IReorderedEvent from plone.base.interfaces import ISiteManagerCreatedEvent +from plone.base.interfaces.recyclebin import IRecycleBin from plone.base.utils import get_installer +from Products.CMFCore.interfaces import IContentish +from zope.component import adapter +from zope.component import queryUtility from zope.interface import implementer from zope.interface.interfaces import ObjectEvent +from zope.lifecycleevent.interfaces import IObjectRemovedEvent @implementer(ISiteManagerCreatedEvent) @@ -37,3 +42,39 @@ def removeBase(event): https://dev.plone.org/ticket/13705 """ event.request.response.base = None + + +@adapter(IContentish, IObjectRemovedEvent) +def handle_content_removal(obj, event): + """Event handler for content removal + + This intercepts standard content removal and puts the item in the recycle bin + instead of letting it be deleted if the recycle bin is enabled. + """ + # Ignore if the object is being moved + if getattr(obj, "_v_is_being_moved", False): + return + + # Ignore if this event was dispatched from a parent container deletion. + # OFS dispatches IObjectRemovedEvent to all sub-objects via dispatchToSublocations, + # keeping event.object pointing to the original deleted container. When obj != event.object, + # it means obj is a child being notified indirectly — it will be captured as + # nested data when the parent container is added to the recycle bin. + if event.object is not obj: + return + + # Get the recycle bin + recycle_bin = queryUtility(IRecycleBin) + if recycle_bin is None or not recycle_bin.is_enabled(): + return + + # Only process if this is a direct deletion (not part of container deletion) + if event.newParent is not None: + return + + # Get original information + original_container = event.oldParent + original_path = "/".join(obj.getPhysicalPath()) + + # Add to recycle bin - let any exceptions propagate to make problems visible + recycle_bin.add_item(obj, original_container, original_path) diff --git a/src/Products/CMFPlone/profiles/default/actions.xml b/src/Products/CMFPlone/profiles/default/actions.xml index a0343972b3..6094382ab0 100644 --- a/src/Products/CMFPlone/profiles/default/actions.xml +++ b/src/Products/CMFPlone/profiles/default/actions.xml @@ -494,6 +494,24 @@ True + + Recycle Bin + + string:${portal_url}/@@recyclebin + string:plone-delete + portal/@@recyclebin-enabled|nothing + + + + True + Inspect Relations + + + Manage portal + diff --git a/src/Products/CMFPlone/profiles/dependencies/registry.xml b/src/Products/CMFPlone/profiles/dependencies/registry.xml index aa1597aa09..1804a75f2c 100644 --- a/src/Products/CMFPlone/profiles/dependencies/registry.xml +++ b/src/Products/CMFPlone/profiles/dependencies/registry.xml @@ -129,5 +129,8 @@ {"actionOptions": {"displayInModal": false}} + diff --git a/src/Products/CMFPlone/recyclebin.py b/src/Products/CMFPlone/recyclebin.py new file mode 100644 index 0000000000..1d754f08d0 --- /dev/null +++ b/src/Products/CMFPlone/recyclebin.py @@ -0,0 +1,841 @@ +from AccessControl import getSecurityManager +from Acquisition import aq_base +from BTrees.OOBTree import OOBTree +from BTrees.OOBTree import OOTreeSet +from datetime import datetime +from datetime import timedelta +from DateTime import DateTime +from persistent import Persistent +from plone.base import PloneMessageFactory as _ +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.base.interfaces.recyclebin import IRecycleBinControlPanelSettings +from plone.registry.interfaces import IRegistry +from Products.CMFCore.interfaces import IContentish +from Products.CMFCore.utils import getToolByName +from zope.annotation.interfaces import IAnnotations +from zope.component import getUtility +from zope.component.hooks import getSite +from zope.interface import implementer + +import logging +import uuid + +logger = logging.getLogger(__name__) + +ANNOTATION_KEY = "Products.CMFPlone.RecycleBin" + + +class RecycleBinStorage(Persistent): + """Storage class for RecycleBin using BTrees for better performance""" + + def __init__(self): + self.items = OOBTree() + # Add a sorted index that stores (deletion_date, item_id) tuples + # This will automatically maintain items sorted by date + self._sorted_index = OOTreeSet() + + def __getitem__(self, key): + return self.items[key] + + def __setitem__(self, key, value): + # When adding or updating an item, update the sorted index + if key in self.items: + # If updating an existing item, remove old index entry first + self._remove_from_index(key) + + # Add the item to main storage + self.items[key] = value + + # Add to sorted index if it has a deletion_date + self._add_to_index(key, value) + + def __delitem__(self, key): + # When deleting an item, also remove it from the sorted index + self._remove_from_index(key) + + # Remove from main storage + del self.items[key] + + def pop(self, key, default=None): + item = self.items.get(key, default) + if item is not default: + del self[key] + return item + + def _add_to_index(self, key, value): + """Add an item to the sorted index""" + # Store as (date, id) for automatic sorting + self._sorted_index.add((value["deletion_date"], key)) + + def _remove_from_index(self, key): + """Remove an item from the sorted index""" + if key not in self.items: + return + + value = self.items[key] + sort_key = (value["deletion_date"], key) + self._sorted_index.remove(sort_key) + + def __contains__(self, key): + return key in self.items + + def __len__(self): + return len(self.items) + + def get(self, key, default=None): + return self.items.get(key, default) + + def keys(self): + return self.items.keys() + + def values(self): + return self.items.values() + + def get_items(self): + """Return all items as key-value pairs""" + return self.items.items() + + def get_items_sorted_by_date(self, reverse=True): + """Return items sorted by deletion date + + Args: + reverse: If True, return newest items first (default), + if False, return oldest items first + + Returns: + Generator yielding (item_id, item_data) tuples + """ + sorted_keys = list(self._sorted_index) + + # If we want newest first (reverse=True), reverse the list + if reverse: + sorted_keys.reverse() + + # Yield items in the requested order + for date, item_id in sorted_keys: + if item_id in self.items: # Double check item still exists + yield (item_id, self.items[item_id]) + + def clear(self): + """Clear all items from the storage""" + self.items.clear() + self._sorted_index.clear() + + +@implementer(IRecycleBin) +class RecycleBin: + """Stores deleted content items""" + + def __init__(self): + """Initialize the recycle bin utility + + It will get the context (Plone site) on demand using getSite() + """ + pass + + def _get_context(self): + """Get the context (Plone site)""" + return getSite() + + def _get_storage(self): + """Get the storage for recycled items""" + context = self._get_context() + annotations = IAnnotations(context) + + if ANNOTATION_KEY not in annotations: + annotations[ANNOTATION_KEY] = RecycleBinStorage() + + return annotations[ANNOTATION_KEY] + + # Update property for storage to use _get_storage + @property + def storage(self): + return self._get_storage() + + def _get_settings(self): + """Get recycle bin settings from registry""" + registry = getUtility(IRegistry) + return registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + + def is_enabled(self): + """Check if recycle bin is enabled""" + try: + settings = self._get_settings() + return settings.recycling_enabled + except Exception as e: + logger.error( + f"Error checking recycle bin settings: {str(e)}. Recycling is disabled." + ) + return False + + def _process_folder_children(self, folder_obj, folder_path): + """Helper method to process folder children recursively + + Only processes children that provide IContentish interface. + Non-content objects (tools, utilities) are skipped. + """ + folder_children = {} + for child_id in folder_obj.objectIds(): + child = folder_obj[child_id] + + # Skip non-content objects (e.g., tools, utilities) + if not IContentish.providedBy(child): + logger.debug( + f"Skipping non-content object {child_id} in folder {folder_path}" + ) + continue + + child_path = f"{folder_path}/{child_id}" + # Get workflow state for this child + child_workflow_state = None + workflow_tool = getToolByName(self._get_context(), "portal_workflow") + child_workflow_state = workflow_tool.getInfoFor(child, "review_state", None) + + # Store basic data for this child + child_data = { + "id": child_id, + "restore_id": str(uuid.uuid4()), + "title": child.Title(), + "portal_type": getattr(child, "portal_type", "Unknown"), + "path": child_path, + "parent_path": folder_path, + "deletion_date": self._get_deletion_date(), + "language": getattr(child, "language", None) + or getattr(child, "Language", lambda: None)(), + "review_state": child_workflow_state, + "object": child, + } + + # If this child is also a folder, process its children + if hasattr(child, "objectIds") and child.objectIds(): + nested_children = self._process_folder_children(child, child_path) + if nested_children: + child_data["children"] = nested_children + + folder_children[child_id] = child_data + return folder_children + + def add_item( + self, + obj, + original_container, + original_path, + item_type=None, + process_children=True, + ): + """Add deleted item to recycle bin + + Args: + obj: The content object to add. Must provide IContentish interface. + original_container: The original container the object was in + original_path: The original path of the object + item_type: Optional type override + process_children: Whether to recursively process folder children + + Returns: + The recycle ID of the stored item, or None if recycling is disabled + + Raises: + TypeError: If obj does not provide IContentish interface + """ + if not self.is_enabled(): + return None + + # Check if obj provides IContentish interface (i.e., is a content item) + if not IContentish.providedBy(obj): + raise TypeError( + f"Object {repr(obj)} does not provide IContentish interface. " + "Only content items can be added to the recycle bin." + ) + + # Now we know the object has getId() since it provides IContentish + item_id = obj.getId() + + # Add a workflow history entry about the deletion if possible + self._update_workflow_history(obj, "deletion") + + # Generate a meaningful title + item_title = obj.Title() + + children = {} + if process_children and hasattr(obj, "objectIds"): + # Process all children recursively + children = self._process_folder_children(obj, original_path) + + # Store metadata about the deletion + parent_path = ( + "/".join(original_container.getPhysicalPath()) + if original_container + else "/".join(original_path.split("/")[:-1]) + ) + + # Get the current user who is deleting the item + user_id = getSecurityManager().getUser().getId() or "System" + + # Get workflow state at time of deletion + workflow_state = None + workflow_tool = getToolByName(self._get_context(), "portal_workflow") + workflow_state = workflow_tool.getInfoFor(obj, "review_state", None) + + # Generate a unique recycle ID + recycle_id = str(uuid.uuid4()) + + storage_data = { + "id": item_id, + "title": item_title, + "portal_type": item_type or getattr(obj, "portal_type", "Unknown"), + "path": original_path, + "parent_path": parent_path, + "deletion_date": self._get_deletion_date(), + "deleted_by": user_id, + "language": getattr(obj, "language", None) + or getattr(obj, "Language", lambda: None)(), + "review_state": workflow_state, + "object": aq_base(obj), # Store the actual object with no acquisition chain + "recycle_id": recycle_id, + } + + # Add children data if this was a folder/collection + if children: + storage_data["children"] = children + + self._purge_expired_items() + + self.storage[recycle_id] = storage_data + + return recycle_id + + def get_items(self): + """Return all items in recycle bin""" + return [ + {**{k: v for k, v in data.items() if k != "object"}, "recycle_id": item_id} + for item_id, data in self.storage.get_items_sorted_by_date(reverse=True) + ] + + def search( + self, + title=None, + path=None, + portal_type=None, + date_from=None, + date_to=None, + deleted_by=None, + has_subitems=None, + language=None, + review_state=None, + sort_on="deletion_date", + sort_order="descending", + ): + """Return filtered and sorted items from the recycle bin. + + The ``title`` and ``path`` filters also search recursively through + children so that a nested child match surfaces the parent item. + """ + items = self.get_items() + reverse = sort_order != "ascending" + + # --- filtering --- + filtered = [] + for item in items: + if portal_type and item.get("portal_type") != portal_type: + if not self._children_match( + item.get("children", {}), + "portal_type", + portal_type, + exact=True, + ): + continue + + if date_from or date_to: + deletion_date = item.get("deletion_date") + if deletion_date: + item_date = ( + deletion_date.date() + if hasattr(deletion_date, "date") + else deletion_date + ) + if date_from and item_date < date_from: + continue + if date_to and item_date > date_to: + continue + + if deleted_by and item.get("deleted_by") != deleted_by: + continue + + if has_subitems is not None: + has_children = bool(item.get("children")) + if has_subitems and not has_children: + continue + if not has_subitems and has_children: + continue + + if language and item.get("language") != language: + continue + + if review_state and item.get("review_state") != review_state: + continue + + if title: + title_lower = title.lower() + if title_lower not in item.get("title", "").lower(): + if not self._children_match( + item.get("children", {}), "title", title_lower + ): + continue + + if path: + path_lower = path.lower() + if path_lower not in item.get("path", "").lower(): + if not self._children_match( + item.get("children", {}), "path", path_lower + ): + continue + + filtered.append(item) + + # --- sorting --- + sort_keys = { + "title": lambda x: x.get("title", "").lower(), + "portal_type": lambda x: x.get("portal_type", "").lower(), + "path": lambda x: x.get("path", "").lower(), + "deletion_date": lambda x: x.get( + "deletion_date", self._get_deletion_date() + ), + "review_state": lambda x: (x.get("review_state") or "").lower(), + } + key_fn = sort_keys.get(sort_on, sort_keys["deletion_date"]) + filtered.sort(key=key_fn, reverse=reverse) + + return filtered + + def _children_match(self, children_dict, field, value, exact=False): + """Return True if *value* is found in *field* of any descendant. + + When *exact* is False (default) a case-insensitive substring check is + performed. When *exact* is True the field value must match exactly + (case-sensitive). + """ + for child_data in children_dict.values(): + child_value = child_data.get(field, "") + if exact: + if child_value == value: + return True + else: + if value in child_value.lower(): + return True + nested = child_data.get("children", {}) + if isinstance(nested, dict) and nested: + if self._children_match(nested, field, value, exact=exact): + return True + return False + + def get_item(self, item_id): + """Get a specific deleted item by ID""" + return self.storage.get(item_id) + + def _update_workflow_history(self, obj, action_type, item_data=None): + """Add a workflow history entry about deletion or restoration + + Args: + obj: The content object + action_type: Either 'deletion' or 'restoration' + item_data: The recyclebin storage data (needed for restoration to show deletion date) + """ + if not hasattr(obj, "workflow_history"): + return + + workflow_tool = getToolByName(self._get_context(), "portal_workflow") + chains = workflow_tool.getChainFor(obj) + + if not chains: + return + + workflow_id = chains[0] + history = obj.workflow_history.get(workflow_id, ()) + + if not history: + return + + current_state = history[-1].get("review_state", None) if history else None + user_id = getSecurityManager().getUser().getId() or "System" + + entry = { + "action": ( + _("Moved to recycle bin") + if action_type == "deletion" + else _("Restored from recycle bin") + ), + "actor": user_id, + "comments": ( + _("Item was deleted and moved to recycle bin") + if action_type == "deletion" + else _("Restored from recycle bin after deletion") + ), + "time": DateTime(), + "review_state": current_state, + } + + # Add the entry and update the history + obj.workflow_history[workflow_id] = history + (entry,) + + def _reset_workflow_state_if_needed(self, obj): + """Reset object workflow state to initial state if the setting is enabled""" + settings = self._get_settings() + if not settings.restore_to_initial_state: + return + + workflow_tool = getToolByName(self._get_context(), "portal_workflow") + chains = workflow_tool.getChainFor(obj) + + if not chains: + return + + workflow_id = chains[0] + workflow = workflow_tool.getWorkflowById(workflow_id) + + if not workflow: + return + + # Get the initial state of the workflow + initial_state = getattr(workflow, "initial_state", None) + if not initial_state: + logger.warning( + f"Could not determine initial state for workflow {workflow_id}" + ) + return + + # Get current state + current_state = workflow_tool.getInfoFor(obj, "review_state", None) + + # Only reset if current state is different from initial state + if current_state != initial_state: + # Reset the workflow state by updating the workflow history + if hasattr(obj, "workflow_history") and workflow_id in obj.workflow_history: + history = obj.workflow_history[workflow_id] + if history: + # Update the last entry to reflect the state reset + user_id = getSecurityManager().getUser().getId() or "System" + + reset_entry = { + "action": _("Reset to initial state"), + "actor": user_id, + "comments": _( + "Workflow state reset to '${initial_state}' during restoration from recycle bin", + mapping={"initial_state": initial_state}, + ), + "time": DateTime(), + "review_state": initial_state, + } + + obj.workflow_history[workflow_id] = history + (reset_entry,) + + # Force the object's state to be updated + workflow._changeStateOf(obj, workflow.states[initial_state]) + + logger.info( + f"Reset workflow state of {obj.getId()} from '{current_state}' to '{initial_state}'" + ) + + def _reset_folder_children_workflow_if_needed(self, folder_obj): + """Recursively reset workflow states of folder children if the setting is enabled""" + settings = self._get_settings() + if not settings.restore_to_initial_state: + return + + # Check if this is a folder-like object + if not hasattr(folder_obj, "objectIds"): + return + + # Recursively reset workflow states for all children + for child_id in folder_obj.objectIds(): + child = folder_obj[child_id] + self._reset_workflow_state_if_needed(child) + + # If the child is also a folder, recurse + if hasattr(child, "objectIds"): + self._reset_folder_children_workflow_if_needed(child) + + def _find_target_container(self, target_container, parent_path): + """Helper to find the target container for restoration + + Returns a tuple (success, container, error_message) where: + - success: Boolean indicating if the container was found + - container: The container object (None if not found) + - error_message: Error message if success is False + """ + site = self._get_context() + if target_container is None: + # Try to get the original parent + try: + target_container = site.unrestrictedTraverse(parent_path) + return True, target_container, None + except (KeyError, AttributeError): + # We need an explicit target container if original parent is gone + error_message = ( + f"Original parent container at {parent_path} no longer exists. " + "You must specify a target_container to restore this item." + ) + return False, None, error_message + return True, target_container, None + + def _reindex_recursive(self, obj): + """Reindex obj and all its descendants in the portal catalog. + + This is necessary after restoring a folder to a different location so + that every item gets the correct new path in the catalog. + """ + if hasattr(obj, "reindexObject"): + obj.reindexObject() + if hasattr(obj, "objectValues"): + for child in obj.objectValues(): + self._reindex_recursive(child) + + def _find_child_by_restore_id(self, children_dict, restore_id): + """Recursively find a child by restore_id in the children tree. + + Returns (child_data, parent_dict, key) or (None, None, None) if not found. + """ + for key, child_data in children_dict.items(): + if child_data.get("restore_id") == restore_id: + return child_data, children_dict, key + nested = child_data.get("children", {}) + if isinstance(nested, dict) and nested: + result = self._find_child_by_restore_id(nested, restore_id) + if result[0] is not None: + return result + return None, None, None + + def restore_child_item(self, item_id, restore_id, target_container): + """Restore a specific child from a recycled folder item. + + Args: + item_id: The recycle bin ID of the parent item. + restore_id: The unique ID assigned to each child for lookup. + target_container: The container object where the child will be restored. + + Returns: + The restored object on success, or a dict with ``success: False`` + and an ``error`` key on failure. + """ + item_data = self.storage.get(item_id) + if not item_data: + return { + "success": False, + "error": f"Item {item_id!r} not found in recycle bin", + } + + if "children" not in item_data: + return {"success": False, "error": "Item has no children"} + + if target_container is None: + return { + "success": False, + "error": "You must specify a target_container to restore a child item", + } + if not restore_id: + return { + "success": False, + "error": "You must provide restore_id to restore a child item", + } + + child_data, parent_dict, child_key = self._find_child_by_restore_id( + item_data["children"], restore_id + ) + if child_data is None: + return { + "success": False, + "error": f"Child with restore_id {restore_id!r} not found", + } + + temp_id = str(uuid.uuid4()) + self.storage[temp_id] = child_data + try: + result = self.restore_item(temp_id, target_container) + except Exception: + if temp_id in self.storage: + del self.storage[temp_id] + raise + + if isinstance(result, dict) and not result.get("success", True): + return result + + # Remove child from parent tree and persist the updated parent record + del parent_dict[child_key] + self.storage[item_id] = item_data + return result + + def _cleanup_child_references(self, item_data): + """Clean up any child items associated with a parent that was restored""" + if "children" in item_data and isinstance(item_data["children"], dict): + logger.info( + f"Cleaning up {len(item_data['children'])} child items from recyclebin" + ) + + # Define a function to recursively process nested folders + def cleanup_children(children_dict): + for child_id, child_data in children_dict.items(): + # Clean up any entries that might match this child + child_path = child_data.get("path") + child_orig_id = child_data.get("id") + + for storage_id, storage_data in list(self.storage.get_items()): + if ( + storage_data.get("path") == child_path + or storage_data.get("id") == child_orig_id + ): + logger.info( + f"Removing child item {child_orig_id} from recyclebin" + ) + if storage_id in self.storage: + del self.storage[storage_id] + + # If this child is also a folder, recursively process its children + if "children" in child_data and isinstance( + child_data["children"], dict + ): + cleanup_children(child_data["children"]) + + # Start the recursive cleanup + cleanup_children(item_data["children"]) + + def _handle_existing_object(self, obj_id, target_container, obj): + """Handle cases where an object with the same ID already exists in target""" + if obj_id in target_container: + raise ValueError( + f"Cannot restore item '{obj_id}' because an item with this ID already exists in the target location. " + f"To replace the existing item with the recycled one, use the recycle bin interface." + ) + + def restore_item(self, item_id, target_container=None): + """Restore item to original location or specified container""" + if item_id not in self.storage: + raise KeyError(f"Item with ID '{item_id}' not found in recycle bin") + + item_data = self.storage[item_id] + obj = item_data["object"] + obj_id = item_data["id"] + + # Regular content object restoration + # Find the container to restore to + success, target_container, error_message = self._find_target_container( + target_container, item_data["parent_path"] + ) + + # If we couldn't find the target container, return the error message + if not success: + return {"success": False, "error": error_message} + + # Make sure we don't overwrite existing content + self._handle_existing_object(obj_id, target_container, obj) + + # Set the new ID if it was changed + if obj_id != item_data["id"]: + obj.id = obj_id + + # Update __parent__ and __name__ on the persistent object BEFORE + # inserting it into the container. OFS does not do this automatically, + # and CMFCatalogAware.manage_afterAdd calls indexObject() during + # _setObject, so the catalog would otherwise record the old path. + # We keep the acquisition-wrapped target_container as __parent__ so + # that getPhysicalPath() can traverse the full portal path during the + # current transaction; ZODB will persist only the underlying reference. + if hasattr(obj, "__parent__"): + obj.__parent__ = target_container + if hasattr(obj, "__name__"): + obj.__name__ = obj_id + + # Add object to the target container + target_container[obj_id] = obj + + # Add a workflow history entry about the restoration + restored_obj = target_container[obj_id] + self._update_workflow_history(restored_obj, "restoration", item_data) + + # Reset workflow state to initial state if the setting is enabled + self._reset_workflow_state_if_needed(restored_obj) + + # Also reset workflow states of children if this is a folder + self._reset_folder_children_workflow_if_needed(restored_obj) + + # Reindex the restored object and all its descendants so the catalog + # reflects the new path and security. + self._reindex_recursive(restored_obj) + + # Remove from recycle bin + del self.storage[item_id] + + return restored_obj + + def purge_item(self, item_id) -> bool: + """Permanently delete an item from the recycle bin + + Args: + item_id: The ID of the item in the recycle bin + + Returns: + Boolean indicating success + """ + if item_id not in self.storage: + logger.warning(f"Cannot purge item {item_id}: not found in recycle bin") + return False + + try: + # Remove only the requested entry. Do not cascade based on child path/id, + # because separate recycle-bin entries may legitimately share those values. + item = self.storage.pop(item_id) + logger.info(f"Item {item['path']} ({item_id}) purged from recycle bin") + return True + except Exception as e: + logger.error(f"Error purging item {item_id}: {str(e)}") + return False + + def _purge_expired_items(self): + """Purge items that exceed the retention period + + Returns: + Number of items purged + """ + try: + settings = self._get_settings() + retention_days = settings.retention_period + + # If retention_period is 0, auto-purging is disabled + if retention_days <= 0: + logger.debug("Auto-purging is disabled (retention_period = 0)") + return 0 + + cutoff_date = datetime.now() - timedelta(days=retention_days) + purge_count = 0 + + # Use sorted index for efficient date-based removal (oldest first) + for item_id, data in list( + self.storage.get_items_sorted_by_date(reverse=False) + ): + deletion_date = data.get("deletion_date") + + # If item is older than retention period, purge it + if deletion_date and deletion_date < cutoff_date: + if self.purge_item(item_id): + purge_count += 1 + logger.info( + f"Item {item_id} purged due to retention policy (deleted on {deletion_date})" + ) + else: + # Since items are sorted by date, once we find an item newer than + # the cutoff date, we can stop checking + break + + return purge_count + + except Exception as e: + logger.error(f"Error purging expired items: {str(e)}") + return 0 + + def clear(self): + """Clear all items from the recycle bin""" + self.storage.clear() + + def _get_deletion_date(self): + return datetime.now() diff --git a/src/Products/CMFPlone/tests/robot/test_controlpanel_actions.robot b/src/Products/CMFPlone/tests/robot/test_controlpanel_actions.robot index 3bc55367aa..eefa652284 100644 --- a/src/Products/CMFPlone/tests/robot/test_controlpanel_actions.robot +++ b/src/Products/CMFPlone/tests/robot/test_controlpanel_actions.robot @@ -91,7 +91,7 @@ I add a new action Type Text //input[@name="form.widgets.id"] favorites Click //div[contains(@class,'pattern-modal-buttons')]/button Wait For Condition Text //body contains favorites - Click //*[@id="content-core"]/section[6]/section/ol/li[8]/form/a + Click //*[@id="content-core"]/section[6]/section/ol/li[9]/form/a Wait For Condition Text //body contains Action Settings Type Text //input[@name="form.widgets.title"] My favorites Type Text //input[@name="form.widgets.url_expr"] string:\${globals_view/navigationRootUrl}/favorites @@ -111,7 +111,7 @@ I delete an action Click //*[@id="content-core"]/section[2]/section/ol/li[1]/form/button[@name="delete"] I change category of an action - Click //*[@id="content-core"]/section[6]/section/ol/li[7]/form/a + Click //*[@id="content-core"]/section[6]/section/ol/li[8]/form/a Wait For Condition Text //body contains Action Settings Select Options By //select[@name="form.widgets.category:list"] value portal_tabs Click //div[contains(@class,'pattern-modal-buttons')]/button diff --git a/src/Products/CMFPlone/tests/test_recyclebin.py b/src/Products/CMFPlone/tests/test_recyclebin.py new file mode 100644 index 0000000000..c3a7979746 --- /dev/null +++ b/src/Products/CMFPlone/tests/test_recyclebin.py @@ -0,0 +1,1918 @@ +from datetime import datetime +from datetime import timedelta +from plone.app.testing import login +from plone.app.testing import PLONE_INTEGRATION_TESTING +from plone.app.testing import setRoles +from plone.app.testing import TEST_USER_ID +from plone.app.testing import TEST_USER_NAME +from plone.base.interfaces.recyclebin import IRecycleBin +from plone.registry.interfaces import IRegistry +from Products.CMFPlone.browser.recyclebin import RecycleBinView +from Products.CMFPlone.controlpanel.browser.recyclebin import ( + IRecycleBinControlPanelSettings, +) +from zope.component import getUtility + +import unittest + + +# Test content factory functions for reusability +def create_test_content(portal, content_type, id_suffix="", title_suffix=""): + """Factory function for creating test content""" + import time + + content_map = { + "Document": ("test-page", "Test Page"), + "News Item": ("test-news", "Test News"), + "Folder": ("test-folder", "Test Folder"), + } + + base_id, base_title = content_map.get(content_type, ("test-item", "Test Item")) + + # If no suffix provided, add timestamp to ensure uniqueness + if not id_suffix: + id_suffix = f"-{int(time.time() * 1000000) % 1000000}" + + obj_id = f"{base_id}{id_suffix}" + obj_title = f"{base_title}{title_suffix}" + + portal.invokeFactory(content_type, obj_id, title=obj_title) + return portal[obj_id] + + +# Helper assertion mixins +class RecycleBinAssertionMixin: + """Mixin providing common assertion methods for recycle bin tests""" + + def assertItemInRecycleBin( + self, item_id, obj_id=None, obj_title=None, obj_type=None + ): + """Assert that an item is properly stored in the recycle bin""" + self.assertIn(item_id, self.recyclebin.storage) + + if obj_id or obj_title or obj_type: + item_data = self.recyclebin.storage[item_id] + if obj_id: + self.assertEqual(item_data["id"], obj_id) + if obj_title: + self.assertEqual(item_data["title"], obj_title) + if obj_type: + self.assertEqual(item_data["portal_type"], obj_type) + + def assertItemNotInRecycleBin(self, item_id): + """Assert that an item is not in the recycle bin""" + self.assertNotIn(item_id, self.recyclebin.storage) + + def assertRecycleBinEmpty(self): + """Assert that the recycle bin is empty""" + items = self.recyclebin.get_items() + self.assertEqual(len(items), 0) + + def assertRecycleBinCount(self, expected_count): + """Assert the number of items in the recycle bin""" + items = self.recyclebin.get_items() + self.assertEqual(len(items), expected_count) + + def assertItemRestored(self, restored_obj, original_id, original_title, container): + """Assert that an item was successfully restored""" + self.assertIsNotNone(restored_obj) + self.assertEqual(restored_obj.getId(), original_id) + self.assertEqual(restored_obj.Title(), original_title) + self.assertIn(original_id, container) + + def assertFolderContentsRestored(self, folder, expected_children): + """Assert that folder contents were properly restored""" + for child_id, child_title in expected_children.items(): + self.assertIn(child_id, folder) + self.assertEqual(folder[child_id].Title(), child_title) + + +# Content creation utilities +class ContentTestHelper: + """Helper class for creating and managing test content""" + + @staticmethod + def create_nested_folder_structure(portal, depth=2): + """Create a nested folder structure for testing""" + current_container = portal + folders = [] + + for i in range(depth): + folder_id = f"folder-level-{i}" + folder_title = f"Folder Level {i}" + current_container.invokeFactory("Folder", folder_id, title=folder_title) + folder = current_container[folder_id] + folders.append(folder) + + # Add some content to each folder + folder.invokeFactory("Document", f"page-{i}", title=f"Page {i}") + folder.invokeFactory("News Item", f"news-{i}", title=f"News {i}") + + current_container = folder + + return folders + + @staticmethod + def create_workflow_content( + portal, workflow_tool, content_type="Document", state="published" + ): + """Create content with specific workflow state""" + obj_id = f"workflow-{state}-{content_type.lower().replace(' ', '-')}" + title = f"Workflow {state.title()} {content_type}" + + portal.invokeFactory(content_type, obj_id, title=title) + obj = portal[obj_id] + + # Only try to transition if workflow tool is available and workflows exist + if workflow_tool and state != "private": + try: + # Check if there are any workflows configured for this content type + workflow_chain = workflow_tool.getChainFor(obj) + if workflow_chain: + # Try to get available transitions + transitions = workflow_tool.getTransitionsFor(obj) + if transitions: + # Look for a publish transition + for transition in transitions: + if transition["id"] == "publish": + workflow_tool.doActionFor(obj, "publish") + break + except Exception: + # If workflow operations fail, just continue with default state + pass + + return obj + + +class RecycleBinTestCase(unittest.TestCase, RecycleBinAssertionMixin): + """Base test case for RecycleBin tests with optimized setup and helper methods""" + + layer = PLONE_INTEGRATION_TESTING + + def setUp(self): + """Set up the test environment""" + self.portal = self.layer["portal"] + self.request = self.layer["request"] + + # Log in as a manager + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + login(self.portal, TEST_USER_NAME) + + # Get the registry to access recycle bin settings + self.registry = getUtility(IRegistry) + + # Get the recycle bin utility + self.recyclebin = getUtility(IRecycleBin) + + # Configure recycle bin with test-optimized settings + self._configure_recyclebin_settings() + + # Clear any existing items from the recycle bin + self._clear_recyclebin() + + def tearDown(self): + """Clean up after the test""" + self._clear_recyclebin() + + def _configure_recyclebin_settings(self, **overrides): + """Configure recycle bin settings with sensible test defaults""" + settings = self.registry.forInterface( + IRecycleBinControlPanelSettings, prefix="recyclebin-controlpanel" + ) + + # Default test settings + default_settings = { + "recycling_enabled": True, + "retention_period": 30, + "restore_to_initial_state": False, + } + + # Apply overrides + default_settings.update(overrides) + + # Set the settings + for key, value in default_settings.items(): + setattr(settings, key, value) + + def _clear_recyclebin(self): + """Clear all items from the recycle bin""" + self.recyclebin.clear() + + def _add_item_to_recyclebin(self, obj, container=None): + """Helper method to add an item to the recycle bin""" + if container is None: + container = self.portal + obj_path = "/".join(obj.getPhysicalPath()) + return self.recyclebin.add_item(obj, container, obj_path) + + def _test_basic_recycle_restore_cycle(self, obj, container=None): + """Test the basic cycle of recycling and restoring an object""" + if container is None: + container = self.portal + + # Store original info + obj_id = obj.getId() + obj_title = obj.Title() + # Use the same logic as the recyclebin implementation for consistency + obj_type = getattr(obj, "portal_type", "Unknown") + + # Add to recycle bin + recycle_id = self._add_item_to_recyclebin(obj, container) + + # Verify it was added correctly + self.assertItemInRecycleBin(recycle_id, obj_id, obj_title, obj_type) + + # Simulate deletion + del container[obj_id] + self.assertNotIn(obj_id, container) + + # Restore the item + restored_obj = self.recyclebin.restore_item(recycle_id) + + # Verify restoration + self.assertItemRestored(restored_obj, obj_id, obj_title, container) + self.assertItemNotInRecycleBin(recycle_id) + + return restored_obj + + +class RecycleBinSetupTests(RecycleBinTestCase): + """Tests for RecycleBin setup and configuration""" + + def test_recyclebin_enabled(self): + """Test that the recycle bin is initialized and enabled""" + self.assertTrue(self.recyclebin.is_enabled()) + + def test_recyclebin_storage(self): + """Test that the storage is correctly initialized""" + storage = self.recyclebin.storage + self.assertEqual(len(storage), 0) + self.assertEqual(list(storage.keys()), []) + + def test_recyclebin_settings(self): + """Test that the settings are correctly initialized""" + settings = self.recyclebin._get_settings() + self.assertTrue(settings.recycling_enabled) + self.assertEqual(settings.retention_period, 30) + self.assertFalse(settings.restore_to_initial_state) + + +class RecycleBinContentTests(RecycleBinTestCase): + """Tests for deleting and restoring basic content items""" + + def setUp(self): + """Set up test content""" + super().setUp() + # Create test content using helper functions + self.page = create_test_content(self.portal, "Document") + self.news = create_test_content(self.portal, "News Item") + + def test_delete_restore_page(self): + """Test deleting and restoring a page""" + self._test_basic_recycle_restore_cycle(self.page) + + def test_delete_restore_news(self): + """Test deleting and restoring a news item""" + restored_news = self._test_basic_recycle_restore_cycle(self.news) + + # Verify additional news-specific behavior + self.assertEqual(restored_news.portal_type, "News Item") + + def test_content_types_metadata_storage(self): + """Test that different content types store metadata correctly""" + content_items = [(self.page, "Document"), (self.news, "News Item")] + + for obj, expected_type in content_items: + with self.subTest(content_type=expected_type): + recycle_id = self._add_item_to_recyclebin(obj) + item_data = self.recyclebin.storage[recycle_id] + + # Common metadata assertions + self.assertEqual(item_data["id"], obj.getId()) + self.assertEqual(item_data["title"], obj.Title()) + self.assertEqual(item_data["portal_type"], expected_type) + self.assertIsInstance(item_data["deletion_date"], datetime) + self.assertIn("deleted_by", item_data) + self.assertEqual(item_data["deleted_by"], TEST_USER_ID) + + # Clean up for next iteration + del self.recyclebin.storage[recycle_id] + + def test_purge_item(self): + """Test purging an item from the recycle bin""" + recycle_id = self._add_item_to_recyclebin(self.page) + + # Verify it was added to the recycle bin + self.assertItemInRecycleBin(recycle_id) + + # Purge the item + result = self.recyclebin.purge_item(recycle_id) + + # Verify the item was purged + self.assertTrue(result) + self.assertItemNotInRecycleBin(recycle_id) + self.assertRecycleBinEmpty() + + def test_deleted_by_field(self): + """Test that deleted_by field is properly stored and retrieved""" + recycle_id = self._add_item_to_recyclebin(self.page) + + # Test deleted_by field in various access methods + access_methods = [ + ("storage", lambda: self.recyclebin.storage[recycle_id]), + ("get_items", lambda: self.recyclebin.get_items()[0]), + ("get_item", lambda: self.recyclebin.get_item(recycle_id)), + ] + + for method_name, get_item_data in access_methods: + with self.subTest(access_method=method_name): + item_data = get_item_data() + self.assertIn("deleted_by", item_data) + self.assertIsInstance(item_data["deleted_by"], str) + self.assertEqual(item_data["deleted_by"], TEST_USER_ID) + + +class RecycleBinFolderTests(RecycleBinTestCase): + """Tests for deleting and restoring folder structures""" + + def setUp(self): + """Set up test content""" + super().setUp() + + # Create a folder with content using helper + self.folder = create_test_content(self.portal, "Folder") + + # Add content to the folder + self.folder.invokeFactory("Document", "folder-page", title="Folder Page") + self.folder.invokeFactory("News Item", "folder-news", title="Folder News") + + # Store expected children for easier testing + self.expected_children = { + "folder-page": "Folder Page", + "folder-news": "Folder News", + } + + def test_delete_restore_folder(self): + """Test deleting and restoring a folder with content""" + restored_folder = self._test_basic_recycle_restore_cycle(self.folder) + + # Verify folder-specific behavior: children were restored + self.assertFolderContentsRestored(restored_folder, self.expected_children) + + def test_folder_children_tracking(self): + """Test that folder children are properly tracked in recycle bin""" + recycle_id = self._add_item_to_recyclebin(self.folder) + item_data = self.recyclebin.storage[recycle_id] + + # Verify children tracking + self.assertIn("children", item_data) + self.assertEqual(len(item_data["children"]), 2) + + # Verify each child is tracked + for child_id in self.expected_children.keys(): + self.assertIn(child_id, item_data["children"]) + + def test_purge_folder_with_contents(self): + """Test purging a folder does not remove standalone child entries""" + # Get the original path + folder_path = "/".join(self.folder.getPhysicalPath()) + page_path = "/".join(self.folder["folder-page"].getPhysicalPath()) + news_path = "/".join(self.folder["folder-news"].getPhysicalPath()) + + # Delete the folder and its contents by adding them individually to the recycle bin + # This simulates how the recycle bin typically receives items when a folder is deleted + folder_recycle_id = self.recyclebin.add_item( + self.folder, self.portal, folder_path + ) + page_recycle_id = self.recyclebin.add_item( + self.folder["folder-page"], self.folder, page_path + ) + news_recycle_id = self.recyclebin.add_item( + self.folder["folder-news"], self.folder, news_path + ) + + # Verify all items were added to the recycle bin + self.assertIn(folder_recycle_id, self.recyclebin.storage) + self.assertIn(page_recycle_id, self.recyclebin.storage) + self.assertIn(news_recycle_id, self.recyclebin.storage) + + # Get all items before purging + before_items = self.recyclebin.get_items() + self.assertEqual(len(before_items), 3) + + # Purge just the folder item + result = self.recyclebin.purge_item(folder_recycle_id) + self.assertTrue(result) + + # Verify only the folder entry was purged. + # Standalone entries for child items are separate delete operations. + self.assertNotIn(folder_recycle_id, self.recyclebin.storage) + self.assertIn(page_recycle_id, self.recyclebin.storage) + self.assertIn(news_recycle_id, self.recyclebin.storage) + + # Verify child standalone entries remain in the listing + after_items = self.recyclebin.get_items() + self.assertEqual(len(after_items), 2) + + +class RecycleBinNestedFolderTests(RecycleBinTestCase): + """Tests for deleting and restoring nested folder structures""" + + def setUp(self): + """Set up test content""" + super().setUp() + + # Use helper to create nested structure + self.folders = ContentTestHelper.create_nested_folder_structure( + self.portal, depth=3 + ) + self.parent_folder = self.folders[0] + self.child_folder = self.folders[1] if len(self.folders) > 1 else None + self.grandchild_folder = self.folders[2] if len(self.folders) > 2 else None + + def test_delete_restore_nested_folder(self): + """Test deleting and restoring a nested folder structure""" + # Get the original paths + parent_path = "/".join(self.parent_folder.getPhysicalPath()) + parent_id = self.parent_folder.getId() + + # Delete the parent folder by adding it to the recycle bin + recycle_id = self.recyclebin.add_item( + self.parent_folder, self.portal, parent_path + ) + + # Verify it was added to the recycle bin + self.assertIsNotNone(recycle_id) + self.assertIn(recycle_id, self.recyclebin.storage) + + # Verify the parent folder metadata was stored correctly + item_data = self.recyclebin.storage[recycle_id] + self.assertEqual(item_data["id"], parent_id) + self.assertEqual(item_data["portal_type"], "Folder") + + # Verify the children were tracked + self.assertIn("children", item_data) + self.assertEqual(len(item_data["children"]), 3) + self.assertIn("page-0", item_data["children"]) + self.assertIn("news-0", item_data["children"]) + self.assertIn("folder-level-1", item_data["children"]) + + # Verify the nested children were tracked + child_data = item_data["children"]["folder-level-1"] + self.assertIn("children", child_data) + self.assertEqual(len(child_data["children"]), 3) + self.assertIn("page-1", child_data["children"]) + self.assertIn("news-1", child_data["children"]) + self.assertIn("folder-level-2", child_data["children"]) + + # Verify the deepest level was tracked + grandchild_data = child_data["children"]["folder-level-2"] + self.assertIn("children", grandchild_data) + self.assertEqual(len(grandchild_data["children"]), 2) + self.assertIn("page-2", grandchild_data["children"]) + self.assertIn("news-2", grandchild_data["children"]) + + # Remove the parent folder from the portal to simulate deletion + del self.portal[parent_id] + self.assertNotIn(parent_id, self.portal) + + # Restore the parent folder + restored_folder = self.recyclebin.restore_item(recycle_id) + + # Verify the parent folder was restored + self.assertIsNotNone(restored_folder) + self.assertEqual(restored_folder.getId(), parent_id) + self.assertIn(parent_id, self.portal) + + # Verify the child folder was restored + self.assertIn("folder-level-1", restored_folder) + restored_child = restored_folder["folder-level-1"] + + # Verify the nested content was restored + self.assertIn("page-1", restored_child) + self.assertIn("news-1", restored_child) + self.assertIn("folder-level-2", restored_child) + + # Verify the deepest level was restored + restored_grandchild = restored_child["folder-level-2"] + self.assertIn("page-2", restored_grandchild) + self.assertIn("news-2", restored_grandchild) + + # Verify the item was removed from the recycle bin + self.assertNotIn(recycle_id, self.recyclebin.storage) + + def test_delete_restore_middle_folder(self): + """Test deleting and restoring a middle-level folder""" + # Get the original paths + child_path = "/".join(self.child_folder.getPhysicalPath()) + child_id = self.child_folder.getId() + + # Delete the child folder by adding it to the recycle bin + recycle_id = self.recyclebin.add_item( + self.child_folder, self.parent_folder, child_path + ) + + # Verify it was added to the recycle bin + self.assertIsNotNone(recycle_id) + self.assertIn(recycle_id, self.recyclebin.storage) + + # Verify the child folder metadata was stored correctly + item_data = self.recyclebin.storage[recycle_id] + self.assertEqual(item_data["id"], child_id) + self.assertEqual(item_data["portal_type"], "Folder") + + # Verify the nested children were tracked + self.assertIn("children", item_data) + self.assertEqual(len(item_data["children"]), 3) + + # Remove the child folder from the parent folder to simulate deletion + del self.parent_folder[child_id] + self.assertNotIn(child_id, self.parent_folder) + + # Restore the child folder + restored_folder = self.recyclebin.restore_item(recycle_id) + + # Verify the child folder was restored + self.assertIsNotNone(restored_folder) + self.assertEqual(restored_folder.getId(), child_id) + self.assertIn(child_id, self.parent_folder) + + # Verify the nested content was restored + self.assertIn("page-1", restored_folder) + self.assertIn("news-1", restored_folder) + self.assertIn("folder-level-2", restored_folder) + + # Verify the deepest level was restored + restored_grandchild = restored_folder["folder-level-2"] + self.assertIn("page-2", restored_grandchild) + self.assertIn("news-2", restored_grandchild) + + # Verify the item was removed from the recycle bin + self.assertNotIn(recycle_id, self.recyclebin.storage) + + +class RecycleBinExpirationTests(RecycleBinTestCase): + """Tests for recyclebin expiration and size limit functionality""" + + def test_purge_expired_items(self): + """Test purging expired items based on retention period""" + # Create a page + self.portal.invokeFactory("Document", "expired-page", title="Expired Page") + page = self.portal["expired-page"] + page_path = "/".join(page.getPhysicalPath()) + + # Add it to the recycle bin + recycle_id = self.recyclebin.add_item(page, self.portal, page_path) + + # Verify it was added + self.assertIn(recycle_id, self.recyclebin.storage) + + # Modify the deletion date to be older than the retention period + old_date = datetime.now() - timedelta(days=31) + # Get the item data, modify it, and set it back to update the index + item_data = self.recyclebin.storage[recycle_id].copy() + item_data["deletion_date"] = old_date + self.recyclebin.storage[recycle_id] = item_data + + # Call _purge_expired_items + purged_count = self.recyclebin._purge_expired_items() + + # Verify the item was purged + self.assertEqual(purged_count, 1) + self.assertNotIn(recycle_id, self.recyclebin.storage) + + +class RecycleBinRestoreEdgeCaseTests(RecycleBinTestCase): + """Tests for edge cases when restoring items""" + + def test_restore_with_parent_gone(self): + """Test restoring an item when its parent container is gone""" + # Create a folder and a document inside it + self.portal.invokeFactory("Folder", "temp-folder", title="Temporary Folder") + folder = self.portal["temp-folder"] + folder.invokeFactory("Document", "orphan-page", title="Orphan Page") + page = folder["orphan-page"] + page_path = "/".join(page.getPhysicalPath()) + + # Add the page to the recycle bin + recycle_id = self.recyclebin.add_item(page, folder, page_path) + + # Delete the folder to simulate parent container being gone + del self.portal["temp-folder"] + + # Trying to restore without a target container should return an error dictionary + result = self.recyclebin.restore_item(recycle_id) + self.assertIsInstance(result, dict) + self.assertFalse( + result.get("success", True) + ) # Should be marked as unsuccessful + self.assertIn("error", result) # Should contain an error message + + # Now restore with an explicit target container + restored_page = self.recyclebin.restore_item( + recycle_id, target_container=self.portal + ) + + # Verify the page was restored to the portal + self.assertIsNotNone(restored_page) + self.assertEqual(restored_page.getId(), "orphan-page") + self.assertIn("orphan-page", self.portal) + + def test_restore_with_name_conflict(self): + """Test restoring an item when an item with same id already exists""" + # Create a page + self.portal.invokeFactory("Document", "conflict-page2", title="Original Page") + page = self.portal["conflict-page2"] + page_path = "/".join(page.getPhysicalPath()) + page_id = page.getId() + + # Add it to the recycle bin + recycle_id = self.recyclebin.add_item(page, self.portal, page_path) + + # Remove the original page from the portal to simulate deletion + del self.portal[page_id] + self.assertNotIn(page_id, self.portal) + + # Create another page with the same ID + self.portal.invokeFactory( + "Document", "conflict-page2", title="Replacement Page" + ) + + # Since the ID already exists, it should raise an error + with self.assertRaises(ValueError): + # Restore the item + self.recyclebin.restore_item(recycle_id) + + +class RecycleBinWorkflowTests(RecycleBinTestCase): + """Tests for workflow state restoration functionality""" + + def setUp(self): + """Set up test content with workflow states""" + super().setUp() + + # Import here to avoid module resolution issues + try: + from Products.CMFCore.utils import getToolByName + + self.workflow_tool = getToolByName(self.portal, "portal_workflow") + except ImportError: + # Fallback for testing without full Plone environment + self.workflow_tool = None + + # Check if workflows are actually available before proceeding + if self.workflow_tool: + try: + # Create a simple test document first + self.portal.invokeFactory( + "Document", "test-workflow-doc", title="Test Doc" + ) + test_obj = self.portal["test-workflow-doc"] + + # Check if workflows are configured + workflow_chain = self.workflow_tool.getChainFor(test_obj) + if not workflow_chain: + self.workflow_tool = None + else: + # Clean up test object + del self.portal["test-workflow-doc"] + except Exception: + self.workflow_tool = None + + if not self.workflow_tool: + self.skipTest("Workflow tool not available or no workflows configured") + + # Create workflow content using helper + self.page = ContentTestHelper.create_workflow_content( + self.portal, self.workflow_tool, "Document", "published" + ) + + def test_workflow_state_restoration_scenarios(self): + """Parameterized test for different workflow restoration scenarios""" + if not self.workflow_tool: + self.skipTest("Workflow tool not available") + + # Create a simple test document without trying to change workflow state + self.portal.invokeFactory( + "Document", "workflow-test-doc", title="Workflow Test" + ) + test_page = self.portal["workflow-test-doc"] + + # Get the current workflow state (whatever it is) + try: + current_state = self.workflow_tool.getInfoFor(test_page, "review_state") + except Exception: + current_state = None + + # Test scenarios based on actual workflow availability + test_scenarios = [ + (False, current_state), # Don't reset workflow, expect current state + ( + True, + None, + ), # Reset workflow, expect initial state (determined dynamically) + ] + + for reset_workflow, expected_state in test_scenarios: + with self.subTest(reset_workflow=reset_workflow): + # Configure workflow reset setting + self._configure_recyclebin_settings( + restore_to_initial_state=reset_workflow + ) + + # Test the recycle/restore cycle + recycle_id = self._add_item_to_recyclebin(test_page) + test_page_id = test_page.getId() + del self.portal[test_page_id] + restored_page = self.recyclebin.restore_item(recycle_id) + + # Verify the page was restored + self.assertIsNotNone(restored_page) + self.assertEqual(restored_page.getId(), test_page_id) + + # Verify workflow state if workflows are available + if current_state is not None: + try: + actual_state = self.workflow_tool.getInfoFor( + restored_page, "review_state" + ) + if expected_state: + self.assertEqual(actual_state, expected_state) + elif reset_workflow: + # For reset workflow, check it matches initial state + workflow_chain = self.workflow_tool.getChainFor( + restored_page + ) + if workflow_chain: + workflow = self.workflow_tool.getWorkflowById( + workflow_chain[0] + ) + initial_state = workflow.initial_state + self.assertEqual(actual_state, initial_state) + except Exception: + # If workflow operations fail, just verify basic restoration + pass + + # Clean up for next iteration - create fresh content + if restored_page.getId() in self.portal: + del self.portal[restored_page.getId()] + + # Recreate test page for next iteration if there is one + if reset_workflow != test_scenarios[-1][0]: # Not the last iteration + self.portal.invokeFactory( + "Document", "workflow-test-doc", title="Workflow Test" + ) + test_page = self.portal["workflow-test-doc"] + + +class RecycleBinStorageTests(RecycleBinTestCase): + """Tests for RecycleBinStorage functionality including BTrees and sorting""" + + def test_storage_initialization(self): + """Test that storage is properly initialized with BTrees""" + storage = self.recyclebin.storage + self.assertIsNotNone(storage.items) + self.assertIsNotNone(storage._sorted_index) + self.assertEqual(len(storage), 0) + + def test_storage_sorted_index(self): + """Test that storage maintains sorted index by deletion date""" + # Create items with different deletion dates + items = [] + for i in range(3): + obj = create_test_content(self.portal, "Document", f"-sorted-{i}") + recycle_id = self._add_item_to_recyclebin(obj) + items.append((recycle_id, obj)) + + # Modify deletion date to create a sequence + import time + + time.sleep(0.001) # Small delay to ensure different timestamps + + # Test sorted retrieval (newest first by default) + sorted_items = list( + self.recyclebin.storage.get_items_sorted_by_date(reverse=True) + ) + self.assertEqual(len(sorted_items), 3) + + # Verify items are sorted by date (newest first) + for i in range(len(sorted_items) - 1): + current_date = sorted_items[i][1]["deletion_date"] + next_date = sorted_items[i + 1][1]["deletion_date"] + self.assertGreaterEqual(current_date, next_date) + + # Test reverse sorting (oldest first) + sorted_items_reverse = list( + self.recyclebin.storage.get_items_sorted_by_date(reverse=False) + ) + self.assertEqual(len(sorted_items_reverse), 3) + + # Verify reverse order + for i in range(len(sorted_items_reverse) - 1): + current_date = sorted_items_reverse[i][1]["deletion_date"] + next_date = sorted_items_reverse[i + 1][1]["deletion_date"] + self.assertLessEqual(current_date, next_date) + + def test_storage_index_maintenance(self): + """Test that sorted index is properly maintained during operations""" + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Verify item is in index + index_items = list(self.recyclebin.storage._sorted_index) + self.assertEqual(len(index_items), 1) + + # Delete item and verify index is cleaned up + del self.recyclebin.storage[recycle_id] + index_items = list(self.recyclebin.storage._sorted_index) + self.assertEqual(len(index_items), 0) + + +class RecycleBinSecurityTests(RecycleBinTestCase): + """Tests for security and user tracking in recycle bin""" + + def test_user_tracking(self): + """Test that deleted_by field correctly tracks the user""" + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + item_data = self.recyclebin.storage[recycle_id] + self.assertEqual(item_data["deleted_by"], TEST_USER_ID) + + def test_different_user_deletions(self): + """Test tracking different users' deletions""" + from plone.app.testing import login + from plone.app.testing import logout + + # Create content as first user + obj1 = create_test_content(self.portal, "Document", "-user1") + recycle_id1 = self._add_item_to_recyclebin(obj1) + + # Change to different user + logout() + # Create a test user + self.portal.acl_users.userFolderAddUser("testuser2", "secret", ["Member"], []) + login(self.portal, "testuser2") + setRoles(self.portal, "testuser2", ["Manager"]) + + obj2 = create_test_content(self.portal, "Document", "-user2") + recycle_id2 = self._add_item_to_recyclebin(obj2) + + # Verify different users are tracked + item1_data = self.recyclebin.storage[recycle_id1] + item2_data = self.recyclebin.storage[recycle_id2] + + self.assertEqual(item1_data["deleted_by"], TEST_USER_ID) + self.assertEqual(item2_data["deleted_by"], "testuser2") + + # Switch back to original user + logout() + login(self.portal, TEST_USER_NAME) + setRoles(self.portal, TEST_USER_ID, ["Manager"]) + + +class RecycleBinRetentionTests(RecycleBinTestCase): + """Tests for retention period and auto-purging""" + + def test_retention_period_enforcement(self): + """Test that items are auto-purged after retention period""" + # Set short retention period + self._configure_recyclebin_settings(retention_period=1) # 1 day + + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Modify the deletion date to be older than retention period + old_date = datetime.now() - timedelta(days=2) + # Get the item data, modify it, and set it back to update the index + item_data = self.recyclebin.storage[recycle_id].copy() + item_data["deletion_date"] = old_date + self.recyclebin.storage[recycle_id] = item_data + + # Trigger expiration check + purged_count = self.recyclebin._purge_expired_items() + + # Verify item was purged + self.assertEqual(purged_count, 1) + self.assertItemNotInRecycleBin(recycle_id) + + def test_retention_period_disabled(self): + """Test that auto-purging can be disabled""" + # Disable retention period + self._configure_recyclebin_settings(retention_period=0) + + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Mock very old deletion date + old_date = datetime.now() - timedelta(days=365) + # Get the item data, modify it, and set it back to update the index + item_data = self.recyclebin.storage[recycle_id].copy() + item_data["deletion_date"] = old_date + self.recyclebin.storage[recycle_id] = item_data + + # Trigger expiration check + purged_count = self.recyclebin._purge_expired_items() + + # Verify no items were purged + self.assertEqual(purged_count, 0) + self.assertItemInRecycleBin(recycle_id) + + +class RecycleBinWorkflowHistoryTests(RecycleBinTestCase): + """Tests for workflow history tracking""" + + def test_workflow_history_on_deletion(self): + """Test that workflow history is updated on deletion""" + obj = create_test_content(self.portal, "Document") + + # Add to recycle bin + recycle_id = self._add_item_to_recyclebin(obj) + + # Verify item was added to recycle bin + self.assertItemInRecycleBin(recycle_id) + + # Check if workflow history was updated (this depends on workflow being available) + if hasattr(obj, "workflow_history"): + # Verify some workflow history exists + self.assertTrue(len(obj.workflow_history) >= 0) + + def test_workflow_history_on_restoration(self): + """Test that workflow history is updated on restoration""" + obj = create_test_content(self.portal, "Document") + obj_id = obj.getId() + + # Add to recycle bin and simulate deletion + recycle_id = self._add_item_to_recyclebin(obj) + del self.portal[obj_id] + + # Restore the item + restored_obj = self.recyclebin.restore_item(recycle_id) + + # Verify restoration was successful + self.assertIsNotNone(restored_obj) + self.assertEqual(restored_obj.getId(), obj_id) + + +class RecycleBinPathTests(RecycleBinTestCase): + """Tests for path handling and resolution""" + + def test_path_storage_and_retrieval(self): + """Test that paths are correctly stored and can be used for restoration""" + # Create nested structure + folder = create_test_content(self.portal, "Folder") + folder.invokeFactory("Document", "nested-doc", title="Nested Document") + nested_doc = folder["nested-doc"] + + # Get original paths + folder_path = "/".join(folder.getPhysicalPath()) + doc_path = "/".join(nested_doc.getPhysicalPath()) + + # Add to recycle bin + folder_recycle_id = self._add_item_to_recyclebin(folder) + doc_recycle_id = self._add_item_to_recyclebin(nested_doc, folder) + + # Verify paths are stored correctly + folder_data = self.recyclebin.storage[folder_recycle_id] + doc_data = self.recyclebin.storage[doc_recycle_id] + + self.assertEqual(folder_data["path"], folder_path) + self.assertEqual(doc_data["path"], doc_path) + self.assertEqual(doc_data["parent_path"], folder_path) + + def test_path_resolution_for_restoration(self): + """Test path resolution during restoration""" + folder = create_test_content(self.portal, "Folder") + folder.invokeFactory("Document", "path-test-doc", title="Path Test Document") + doc = folder["path-test-doc"] + + # Add document to recycle bin + doc_recycle_id = self._add_item_to_recyclebin(doc, folder) + + # Simulate deletion of document only (folder remains) + del folder["path-test-doc"] + + # Restore document + restored_doc = self.recyclebin.restore_item(doc_recycle_id) + + # Verify document was restored to correct location + self.assertIsNotNone(restored_doc) + self.assertIn("path-test-doc", folder) + self.assertEqual(folder["path-test-doc"].Title(), "Path Test Document") + + +class RecycleBinMetadataTests(RecycleBinTestCase): + """Tests for comprehensive metadata storage and retrieval""" + + def test_complete_metadata_storage(self): + """Test that all expected metadata fields are stored""" + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + item_data = self.recyclebin.storage[recycle_id] + + # Required fields + required_fields = [ + "id", + "title", + "portal_type", + "path", + "parent_path", + "deletion_date", + "deleted_by", + "object", + ] + + for field in required_fields: + with self.subTest(field=field): + self.assertIn(field, item_data) + self.assertIsNotNone(item_data[field]) + + +class RecycleBinSpecialContentTests(RecycleBinTestCase): + """Tests for special content types and edge cases""" + + def test_title_fallback_mechanisms(self): + """Test different title fallback mechanisms""" + obj = create_test_content(self.portal, "Document") + + # Test with Title method (normal case) + recycle_id = self._add_item_to_recyclebin(obj) + item_data = self.recyclebin.storage[recycle_id] + self.assertEqual(item_data["title"], obj.Title()) + + +class RecycleBinConcurrencyTests(RecycleBinTestCase): + """Tests for concurrent operations and data integrity""" + + def test_concurrent_additions(self): + """Test that concurrent additions work correctly""" + # Simulate concurrent additions + items = [] + for i in range(10): + obj = create_test_content(self.portal, "Document", f"-concurrent-{i}") + recycle_id = self._add_item_to_recyclebin(obj) + items.append(recycle_id) + + # Verify all items were added + self.assertRecycleBinCount(10) + + # Verify each item is properly stored + for recycle_id in items: + self.assertItemInRecycleBin(recycle_id) + + def test_concurrent_modifications(self): + """Test that storage handles concurrent modifications properly""" + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Modify item data + original_data = self.recyclebin.storage[recycle_id].copy() + original_data["custom_field"] = "test_value" + self.recyclebin.storage[recycle_id] = original_data + + # Verify modification was preserved + modified_data = self.recyclebin.storage[recycle_id] + self.assertEqual(modified_data["custom_field"], "test_value") + + +class OptimizedRecycleBinTests(RecycleBinTestCase): + """Demonstration of optimized test patterns and comprehensive scenarios""" + + def test_multiple_content_types_cycle(self): + """Test recycle/restore cycle for multiple content types in one test""" + content_types = ["Document", "News Item", "Folder"] + + for content_type in content_types: + with self.subTest(content_type=content_type): + # Create content + obj = create_test_content( + self.portal, content_type, f"-{content_type.lower()}" + ) + + # Test full cycle + self._test_basic_recycle_restore_cycle(obj) + + def test_bulk_operations_with_decorator(self): + """Test bulk operations with multiple content types""" + # Create test content directly + doc = create_test_content(self.portal, "Document", "-bulk") + news = create_test_content(self.portal, "News Item", "-bulk") + folder = create_test_content(self.portal, "Folder", "-bulk") + + test_objects = [doc, news, folder] + recycle_ids = [] + + # Add all items to recycle bin + for obj in test_objects: + recycle_id = self._add_item_to_recyclebin(obj) + recycle_ids.append(recycle_id) + + # Verify all items are in recycle bin + self.assertRecycleBinCount(len(recycle_ids)) + + # Purge all items + for recycle_id in recycle_ids: + self.assertTrue(self.recyclebin.purge_item(recycle_id)) + + # Verify recycle bin is empty + self.assertRecycleBinEmpty() + + def test_edge_cases_comprehensive(self): + """Test various edge cases in a single comprehensive test""" + # Test with empty recycle bin + self.assertRecycleBinEmpty() + + # Test invalid operations + self.assertFalse(self.recyclebin.purge_item("non-existent-id")) + self.assertIsNone(self.recyclebin.get_item("non-existent-id")) + + # Test settings configuration + self._configure_recyclebin_settings( + recycling_enabled=False, retention_period=60 + ) + + settings = self.recyclebin._get_settings() + self.assertFalse(settings.recycling_enabled) + self.assertEqual(settings.retention_period, 60) + + def test_disabled_recyclebin_behavior(self): + """Test behavior when recycle bin is disabled""" + # Disable recycling + self._configure_recyclebin_settings(recycling_enabled=False) + + # Verify is_enabled returns False + self.assertFalse(self.recyclebin.is_enabled()) + + # Try to add item - should return None + obj = create_test_content(self.portal, "Document") + result = self.recyclebin.add_item( + obj, self.portal, "/".join(obj.getPhysicalPath()) + ) + + self.assertIsNone(result) + self.assertRecycleBinEmpty() + + def test_error_handling_in_settings(self): + """Test error handling when settings are not available""" + # Test that recyclebin gracefully handles missing settings + # In normal operation, is_enabled() should work correctly + enabled_state = self.recyclebin.is_enabled() + self.assertIsInstance(enabled_state, bool) + + def test_comprehensive_item_lifecycle(self): + """Test complete item lifecycle from creation to final purge""" + # Create and track content through entire lifecycle + obj = create_test_content(self.portal, "Document", "-lifecycle") + obj_id = obj.getId() + obj_title = obj.Title() + + # Stage 1: Add to recycle bin + recycle_id = self._add_item_to_recyclebin(obj) + self.assertIsNotNone(recycle_id) + self.assertItemInRecycleBin(recycle_id) + + # Stage 2: Verify all metadata is present + item_data = self.recyclebin.get_item(recycle_id) + self.assertEqual(item_data["id"], obj_id) + self.assertEqual(item_data["title"], obj_title) + self.assertIn("deletion_date", item_data) + self.assertIn("deleted_by", item_data) + + # Stage 3: Verify item appears in listings + all_items = self.recyclebin.get_items() + self.assertEqual(len(all_items), 1) + self.assertEqual(all_items[0]["id"], obj_id) + + # Stage 4: Simulate deletion from portal + del self.portal[obj_id] + self.assertNotIn(obj_id, self.portal) + + # Stage 5: Restore item + restored_obj = self.recyclebin.restore_item(recycle_id) + self.assertIsNotNone(restored_obj) + self.assertEqual(restored_obj.getId(), obj_id) + self.assertIn(obj_id, self.portal) + + # Stage 6: Item should be removed from recycle bin after restoration + self.assertItemNotInRecycleBin(recycle_id) + # Note: Other tests may have left items in recycle bin, so check specific item only + + # Stage 7: Add back to recycle bin and permanently purge + recycle_id2 = self._add_item_to_recyclebin(restored_obj) + self.assertItemInRecycleBin(recycle_id2) + + purged = self.recyclebin.purge_item(recycle_id2) + self.assertTrue(purged) + self.assertItemNotInRecycleBin(recycle_id2) + + +class RecycleBinNestedContentTests(RecycleBinTestCase): + """Tests for nested content handling""" + + def test_nested_folder_structure(self): + """Test handling of nested folder structures""" + # Create nested structure + folder1 = create_test_content(self.portal, "Folder", "-level1") + folder1.invokeFactory("Folder", "level2", title="Level 2 Folder") + folder2 = folder1["level2"] + folder2.invokeFactory("Document", "nested-doc", title="Nested Document") + nested_doc = folder2["nested-doc"] + + # Test adding nested items to recycle bin + doc_recycle_id = self._add_item_to_recyclebin(nested_doc, folder2) + folder2_recycle_id = self._add_item_to_recyclebin(folder2, folder1) + folder1_recycle_id = self._add_item_to_recyclebin(folder1, self.portal) + + # Verify all items are in recycle bin + self.assertItemInRecycleBin(doc_recycle_id) + self.assertItemInRecycleBin(folder2_recycle_id) + self.assertItemInRecycleBin(folder1_recycle_id) + + # Verify parent paths are correctly stored + doc_data = self.recyclebin.storage[doc_recycle_id] + folder2_data = self.recyclebin.storage[folder2_recycle_id] + folder1_data = self.recyclebin.storage[folder1_recycle_id] + + # Check that paths contain the expected components + self.assertIn("level2", doc_data["parent_path"]) + self.assertIn("level1", folder2_data["parent_path"]) + self.assertIn("plone", folder1_data["parent_path"]) + + def test_parent_path_resolution(self): + """Test parent path resolution for deeply nested content""" + # Create deep nesting + current_container = self.portal + containers = [] + + for i in range(3): + folder_id = f"folder-{i}" + current_container.invokeFactory("Folder", folder_id, title=f"Folder {i}") + current_container = current_container[folder_id] + containers.append(current_container) + + # Add final document + current_container.invokeFactory("Document", "deep-doc", title="Deep Document") + deep_doc = current_container["deep-doc"] + + # Add to recycle bin + recycle_id = self._add_item_to_recyclebin(deep_doc, current_container) + + # Verify path information + item_data = self.recyclebin.storage[recycle_id] + self.assertIn("folder-0", item_data["parent_path"]) + self.assertIn("folder-1", item_data["parent_path"]) + self.assertIn("folder-2", item_data["parent_path"]) + + +class RecycleBinBulkOperationsTests(RecycleBinTestCase): + """Tests for bulk operations and performance""" + + def test_bulk_addition_performance(self): + """Test performance of bulk additions""" + import time + + # Create multiple items + items = [] + start_time = time.time() + + for i in range(20): + obj = create_test_content(self.portal, "Document", f"-bulk-{i}") + recycle_id = self._add_item_to_recyclebin(obj) + items.append(recycle_id) + + end_time = time.time() + duration = end_time - start_time + + # Should complete in reasonable time (less than 2 seconds for 20 items) + self.assertLess(duration, 2.0) + + # Verify all items were added + self.assertEqual(len(items), 20) + for recycle_id in items: + self.assertItemInRecycleBin(recycle_id) + + def test_bulk_retrieval_operations(self): + """Test bulk retrieval operations""" + # Add multiple items + items = [] + for i in range(10): + obj = create_test_content(self.portal, "Document", f"-retrieval-{i}") + recycle_id = self._add_item_to_recyclebin(obj) + items.append(recycle_id) + + # Test get_items performance + import time + + start_time = time.time() + all_items = self.recyclebin.get_items() + end_time = time.time() + + # Should be fast + self.assertLess(end_time - start_time, 1.0) + self.assertGreaterEqual(len(all_items), 10) + + def test_bulk_purge_operations(self): + """Test bulk purging operations""" + # Add items + items = [] + for i in range(5): + obj = create_test_content(self.portal, "Document", f"-purge-{i}") + recycle_id = self._add_item_to_recyclebin(obj) + items.append(recycle_id) + + # Purge all items + purged_count = 0 + for recycle_id in items: + if self.recyclebin.purge_item(recycle_id): + purged_count += 1 + + # Verify all were purged + self.assertEqual(purged_count, 5) + for recycle_id in items: + self.assertItemNotInRecycleBin(recycle_id) + + +class RecycleBinDataIntegrityTests(RecycleBinTestCase): + """Tests for data integrity and consistency""" + + def test_storage_consistency_after_operations(self): + """Test that storage remains consistent after various operations""" + # Add item + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Verify storage consistency + self.assertEqual(len(self.recyclebin.storage), len(self.recyclebin.get_items())) + + # Purge item + self.recyclebin.purge_item(recycle_id) + + # Verify consistency after purge + self.assertEqual(len(self.recyclebin.storage), len(self.recyclebin.get_items())) + + def test_metadata_integrity(self): + """Test that metadata remains intact throughout operations""" + obj = create_test_content(self.portal, "Document") + original_title = obj.Title() + original_id = obj.getId() + + recycle_id = self._add_item_to_recyclebin(obj) + + # Verify metadata integrity + item_data = self.recyclebin.storage[recycle_id] + self.assertEqual(item_data["title"], original_title) + self.assertEqual(item_data["id"], original_id) + self.assertIn("deletion_date", item_data) + self.assertIn("deleted_by", item_data) + + # Get item through different methods and verify consistency + get_item_result = self.recyclebin.get_item(recycle_id) + get_items_result = [ + item for item in self.recyclebin.get_items() if item["id"] == original_id + ][0] + + key_fields = ["id", "title", "portal_type", "deleted_by"] + for field in key_fields: + self.assertEqual(item_data[field], get_item_result[field]) + self.assertEqual(item_data[field], get_items_result[field]) + + +class RecycleBinBrowserViewTests(RecycleBinTestCase): + """Tests for browser view integration and functionality""" + + def test_recyclebin_view_integration(self): + """Test integration with browser views""" + # This tests the integration points that browser views would use + obj = create_test_content(self.portal, "Document") + recycle_id = self._add_item_to_recyclebin(obj) + + # Test methods that browser views typically use + items = self.recyclebin.get_items() + self.assertGreater(len(items), 0) + + item = self.recyclebin.get_item(recycle_id) + self.assertIsNotNone(item) + + # Test that view-related metadata is present + self.assertIn("deletion_date", item) + self.assertIn("portal_type", item) + + def test_item_filtering_support(self): + """Test support for filtering that browser views might use""" + # Create items of different types + doc = create_test_content(self.portal, "Document", "-filter-doc") + folder = create_test_content(self.portal, "Folder", "-filter-folder") + + self._add_item_to_recyclebin(doc) + self._add_item_to_recyclebin(folder) + + # Get all items + all_items = self.recyclebin.get_items() + + # Filter by type (simulating what a browser view might do) + doc_items = [item for item in all_items if item["portal_type"] == "Document"] + folder_items = [item for item in all_items if item["portal_type"] == "Folder"] + + self.assertGreater(len(doc_items), 0) + self.assertGreater(len(folder_items), 0) + + # Verify specific items are found + doc_found = any(item["id"].endswith("-filter-doc") for item in doc_items) + folder_found = any( + item["id"].endswith("-filter-folder") for item in folder_items + ) + + self.assertTrue(doc_found) + self.assertTrue(folder_found) + + +class RecycleBinEventDispatchTests(RecycleBinTestCase): + """Tests for handle_content_removal with OFS sub-location event dispatch.""" + + def _make_folder_with_children(self, folder_id="evt-folder"): + """Create a folder with two leaf documents and return it.""" + self.portal.invokeFactory("Folder", folder_id, title="Event Test Folder") + folder = self.portal[folder_id] + folder.invokeFactory("Document", "child-a", title="Child A") + folder.invokeFactory("Document", "child-b", title="Child B") + return folder + + def test_deleting_folder_creates_single_entry(self): + """Deleting a folder must produce exactly one recycle-bin entry.""" + folder = self._make_folder_with_children() + folder_path = "/".join(folder.getPhysicalPath()) + + self.recyclebin.add_item(folder, self.portal, folder_path) + + # Only one top-level entry must exist + self.assertRecycleBinCount(1) + + def test_children_stored_inside_folder_entry(self): + """Children must be stored inside the folder entry, not as standalone entries.""" + folder = self._make_folder_with_children() + folder_path = "/".join(folder.getPhysicalPath()) + + recycle_id = self.recyclebin.add_item(folder, self.portal, folder_path) + item_data = self.recyclebin.storage[recycle_id] + + self.assertIn("children", item_data) + self.assertIn("child-a", item_data["children"]) + self.assertIn("child-b", item_data["children"]) + + def test_simulated_ofs_dispatch_does_not_add_children_as_standalone(self): + """Simulates OFS dispatching the event to a child (event.object != obj). + + The handler must skip children when event.object is the parent folder. + """ + from Products.CMFPlone.events import handle_content_removal + from unittest.mock import MagicMock + + folder = self._make_folder_with_children("dispatch-folder") + child = folder["child-a"] + + # Build a fake IObjectRemovedEvent where event.object == folder (parent) + fake_event = MagicMock() + fake_event.newParent = None + fake_event.oldParent = self.portal + fake_event.object = folder # <-- parent, not the child + + handle_content_removal(child, fake_event) + + # No entry must have been created for the child + self.assertRecycleBinEmpty() + + +class RecycleBinFindChildByRestoreIdTests(RecycleBinTestCase): + """Unit tests for RecycleBin._find_child_by_restore_id.""" + + def _make_tree(self): + return { + "a": { + "id": "a", + "restore_id": "rid-a", + "children": { + "b": { + "id": "b", + "restore_id": "rid-b", + "children": { + "c": { + "id": "c", + "restore_id": "rid-c", + } + }, + } + }, + }, + "d": { + "id": "d", + "restore_id": "rid-d", + }, + } + + def test_finds_direct_child_by_restore_id(self): + tree = self._make_tree() + data, _, key = self.recyclebin._find_child_by_restore_id(tree, "rid-d") + self.assertIsNotNone(data) + self.assertEqual(data["id"], "d") + self.assertEqual(key, "d") + + def test_finds_deeply_nested_child_by_restore_id(self): + tree = self._make_tree() + data, parent, key = self.recyclebin._find_child_by_restore_id(tree, "rid-c") + self.assertIsNotNone(data) + self.assertEqual(data["id"], "c") + self.assertEqual(key, "c") + + def test_returns_none_for_missing_restore_id(self): + tree = self._make_tree() + data, parent, key = self.recyclebin._find_child_by_restore_id( + tree, "rid-missing" + ) + self.assertIsNone(data) + self.assertIsNone(parent) + self.assertIsNone(key) + + +class RecycleBinFlattenChildrenTests(RecycleBinTestCase): + """Unit tests for RecycleBinView._flatten_children.""" + + def setUp(self): + super().setUp() + self.view = RecycleBinView(self.portal, self.request) + + def _make_nested(self): + return { + "a": { + "id": "a", + "path": "/root/a", + "children": { + "b": { + "id": "b", + "path": "/root/a/b", + "children": { + "c": { + "id": "c", + "path": "/root/a/b/c", + } + }, + } + }, + }, + "d": { + "id": "d", + "path": "/root/d", + }, + } + + def test_all_descendants_are_yielded(self): + tree = self._make_nested() + flat = list(self.view._flatten_children(tree)) + ids = [e["id"] for e in flat] + self.assertIn("a", ids) + self.assertIn("b", ids) + self.assertIn("c", ids) + self.assertIn("d", ids) + self.assertEqual(len(flat), 4) + + def test_depth_increases_with_nesting(self): + tree = self._make_nested() + flat = {e["id"]: e for e in self.view._flatten_children(tree)} + self.assertEqual(flat["a"]["depth"], 0) + self.assertEqual(flat["b"]["depth"], 1) + self.assertEqual(flat["c"]["depth"], 2) + self.assertEqual(flat["d"]["depth"], 0) + + def test_children_key_stripped_from_entries(self): + tree = self._make_nested() + for entry in self.view._flatten_children(tree): + self.assertNotIn("children", entry) + + def test_children_count_set_for_nodes_with_sub_children(self): + tree = self._make_nested() + flat = {e["id"]: e for e in self.view._flatten_children(tree)} + # "a" has one direct child "b" which itself has child "c" → count = 2 + self.assertIn("children_count", flat["a"]) + self.assertEqual(flat["a"]["children_count"], 2) + # "d" has no children → no children_count key + self.assertNotIn("children_count", flat["d"]) + + +class RecycleBinCountDescendantsTests(RecycleBinTestCase): + """Unit tests for RecycleBinView._count_descendants.""" + + def setUp(self): + super().setUp() + self.view = RecycleBinView(self.portal, self.request) + + def test_empty_dict_returns_zero(self): + self.assertEqual(self.view._count_descendants({}), 0) + + def test_flat_children(self): + children = { + "a": {"id": "a"}, + "b": {"id": "b"}, + } + self.assertEqual(self.view._count_descendants(children), 2) + + def test_deeply_nested_children(self): + children = { + "a": { + "id": "a", + "children": { + "b": { + "id": "b", + "children": { + "c": {"id": "c"}, + }, + } + }, + } + } + # a + b + c = 3 + self.assertEqual(self.view._count_descendants(children), 3) + + def test_mixed_flat_and_nested(self): + children = { + "flat": {"id": "flat"}, + "nested": { + "id": "nested", + "children": { + "child": {"id": "child"}, + }, + }, + } + # flat + nested + child = 3 + self.assertEqual(self.view._count_descendants(children), 3) + + +class RecycleBinChildRestoreReindexTests(RecycleBinTestCase): + """Integration tests for RecycleBin.restore_child_item. + + These tests use Plone's manage_delObjects() to delete content so that + the real event handlers (handle_content_removal) are triggered and the + item lands in the recycle bin exactly as it would in production. + """ + + def _delete(self, container, obj_id): + """Delete an object via manage_delObjects (fires Plone event handlers).""" + container.manage_delObjects([obj_id]) + + def _find_child_restore_id(self, children, child_id): + """Find restore_id for a child id in a nested children tree.""" + for child_data in children.values(): + if child_data.get("id") == child_id: + return child_data.get("restore_id") + nested = child_data.get("children", {}) + if isinstance(nested, dict) and nested: + found = self._find_child_restore_id(nested, child_id) + if found: + return found + return None + + def test_restore_child_item_error_when_parent_not_found(self): + """restore_child_item must return an error dict when item_id is unknown.""" + result = self.recyclebin.restore_child_item( + "nonexistent-id", restore_id="some-id", target_container=self.portal + ) + self.assertFalse(result.get("success", True)) + self.assertIn("error", result) + + def test_restore_child_item_error_when_restore_id_not_found(self): + """restore_child_item must return an error dict when restore_id is unknown.""" + self.portal.invokeFactory("Folder", "err-folder", title="Err Folder") + err_folder = self.portal["err-folder"] + err_folder.invokeFactory("Document", "err-doc", title="Err Doc") + + # Delete the whole folder — event handler stores it with its children + self._delete(self.portal, "err-folder") + + # There must be exactly one item in the bin (the folder) + items = self.recyclebin.get_items() + self.assertEqual(len(items), 1) + recycle_id = items[0]["recycle_id"] + + result = self.recyclebin.restore_child_item( + recycle_id, + restore_id="missing-restore-id", + target_container=self.portal, + ) + self.assertFalse(result.get("success", True)) + self.assertIn("error", result) + + def test_restore_child_item_success_direct_child(self): + """restore_child_item must restore a direct child by restore_id.""" + self.portal.invokeFactory("Folder", "src-folder", title="Src Folder") + src_folder = self.portal["src-folder"] + src_folder.invokeFactory("Document", "a-doc", title="A Doc") + + # Delete the whole folder via Plone (fires events, fills recycle bin) + self._delete(self.portal, "src-folder") + + items = self.recyclebin.get_items() + self.assertEqual(len(items), 1) + recycle_id = items[0]["recycle_id"] + parent_data = self.recyclebin.get_item(recycle_id) + restore_id = self._find_child_restore_id( + parent_data.get("children", {}), "a-doc" + ) + self.assertIsNotNone(restore_id) + + # Create a separate target container + self.portal.invokeFactory("Folder", "target-folder", title="Target Folder") + target = self.portal["target-folder"] + + result = self.recyclebin.restore_child_item( + recycle_id, + restore_id=restore_id, + target_container=target, + ) + + self.assertIsNotNone(result) + self.assertTrue( + hasattr(result, "getId"), "result should be the restored object" + ) + self.assertEqual(result.getId(), "a-doc") + self.assertIn("a-doc", target.objectIds()) + + def test_restore_child_item_removes_child_from_parent_tree(self): + """After restore_child_item the child must be removed from the parent's children.""" + self.portal.invokeFactory("Folder", "src2", title="Src2") + src2 = self.portal["src2"] + src2.invokeFactory("Document", "child-doc", title="Child Doc") + doc_path = "/".join(src2["child-doc"].getPhysicalPath()) + + self._delete(self.portal, "src2") + + items = self.recyclebin.get_items() + self.assertEqual(len(items), 1) + recycle_id = items[0]["recycle_id"] + parent_data = self.recyclebin.get_item(recycle_id) + restore_id = self._find_child_restore_id( + parent_data.get("children", {}), "child-doc" + ) + self.assertIsNotNone(restore_id) + + self.portal.invokeFactory("Folder", "dest2", title="Dest2") + dest = self.portal["dest2"] + + self.recyclebin.restore_child_item( + recycle_id, + restore_id=restore_id, + target_container=dest, + ) + + # Parent entry must still exist but without the restored child + parent_data = self.recyclebin.get_item(recycle_id) + self.assertIsNotNone(parent_data) + child_paths = [c.get("path") for c in parent_data.get("children", {}).values()] + self.assertNotIn(doc_path, child_paths) + + def test_restored_child_is_catalogued_at_new_path(self): + """Restored child must be findable in the catalog at its new location.""" + self.portal.invokeFactory("Folder", "orig", title="Orig") + orig = self.portal["orig"] + orig.invokeFactory("Document", "catalogued-doc", title="Catalogued Doc") + + self._delete(self.portal, "orig") + + items = self.recyclebin.get_items() + self.assertEqual(len(items), 1) + recycle_id = items[0]["recycle_id"] + parent_data = self.recyclebin.get_item(recycle_id) + restore_id = self._find_child_restore_id( + parent_data.get("children", {}), "catalogued-doc" + ) + self.assertIsNotNone(restore_id) + + self.portal.invokeFactory("Folder", "new-home", title="New Home") + new_home = self.portal["new-home"] + + self.recyclebin.restore_child_item( + recycle_id, + restore_id=restore_id, + target_container=new_home, + ) + + catalog = self.portal.portal_catalog + expected_path = "/".join(new_home.getPhysicalPath()) + "/catalogued-doc" + results = catalog.searchResults(path={"query": expected_path, "depth": 0}) + self.assertEqual( + len(results), 1, "Restored child must appear in catalog at new path" + ) + + def test_restore_child_item_error_when_restore_id_missing(self): + """restore_child_item must return an error dict when restore_id is empty.""" + self.portal.invokeFactory("Folder", "missing-id-folder", title="Missing ID") + missing_id_folder = self.portal["missing-id-folder"] + missing_id_folder.invokeFactory("Document", "doc", title="Doc") + + self._delete(self.portal, "missing-id-folder") + + items = self.recyclebin.get_items() + self.assertEqual(len(items), 1) + recycle_id = items[0]["recycle_id"] + + result = self.recyclebin.restore_child_item( + recycle_id, + restore_id="", + target_container=self.portal, + ) + + self.assertFalse(result.get("success", True)) + self.assertIn("error", result) + + +class RecycleBinSearchChildrenTests(RecycleBinTestCase): + """Tests that search() also matches titles and paths of nested children.""" + + def setUp(self): + super().setUp() + # Create a 3-level deep structure: + # search-root/ (Folder) + # top-doc (Document, title="Top Doc") + # sub-folder/ (Folder) + # nested-news (News Item, title="Deeply Nested News") + # deep-folder/ (Folder) + # deep-doc (Document, title="Deepest Page") + self.portal.invokeFactory("Folder", "search-root", title="Search Root") + root = self.portal["search-root"] + root.invokeFactory("Document", "top-doc", title="Top Doc") + root.invokeFactory("Folder", "sub-folder", title="Sub Folder") + sub = root["sub-folder"] + sub.invokeFactory("News Item", "nested-news", title="Deeply Nested News") + sub.invokeFactory("Folder", "deep-folder", title="Deep Folder") + deep = sub["deep-folder"] + deep.invokeFactory("Document", "deep-doc", title="Deepest Page") + + root_path = "/".join(root.getPhysicalPath()) + self.recycle_id = self.recyclebin.add_item(root, self.portal, root_path) + self.portal.manage_delObjects(["search-root"]) + + def test_search_title_matches_root(self): + """Searching for the root title returns the item.""" + results = self.recyclebin.search(title="Search Root") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_title_matches_direct_child(self): + """Searching for a direct child's title returns the parent item.""" + results = self.recyclebin.search(title="Top Doc") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_title_matches_deeply_nested_child(self): + """Searching for a deeply nested child's title returns the parent item.""" + results = self.recyclebin.search(title="Deepest Page") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_title_partial_match_in_child(self): + """Partial title match in a child returns the parent item.""" + results = self.recyclebin.search(title="Nested News") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_title_no_match(self): + """Searching for a title not present anywhere returns nothing.""" + results = self.recyclebin.search(title="Nonexistent Title XYZ") + ids = [r["recycle_id"] for r in results] + self.assertNotIn(self.recycle_id, ids) + + def test_search_path_matches_nested_child(self): + """Searching for a path fragment of a nested child returns the parent item.""" + results = self.recyclebin.search(path="deep-folder/deep-doc") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_path_no_match(self): + """Searching for a path not present anywhere returns nothing.""" + results = self.recyclebin.search(path="/nonexistent/path/xyz") + ids = [r["recycle_id"] for r in results] + self.assertNotIn(self.recycle_id, ids) + + def test_search_portal_type_matches_nested_child(self): + """Searching for a portal_type present only in a nested child returns the parent.""" + results = self.recyclebin.search(portal_type="News Item") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_portal_type_matches_deeply_nested_child(self): + """Searching for Document matches the deeply nested deep-doc child.""" + # The root is a Folder, but deep-doc (3 levels down) is a Document. + results = self.recyclebin.search(portal_type="Document") + ids = [r["recycle_id"] for r in results] + self.assertIn(self.recycle_id, ids) + + def test_search_portal_type_no_match(self): + """Searching for a portal_type not present anywhere returns nothing.""" + results = self.recyclebin.search(portal_type="Event") + ids = [r["recycle_id"] for r in results] + self.assertNotIn(self.recycle_id, ids) + + def test_search_portal_type_is_exact_match(self): + """portal_type filter uses exact match, not substring.""" + # "News" is a substring of "News Item" but must NOT match. + results = self.recyclebin.search(portal_type="News") + ids = [r["recycle_id"] for r in results] + self.assertNotIn(self.recycle_id, ids)