diff --git a/src/backend/core/admin.py b/src/backend/core/admin.py index 25dab1778..eadbeabbc 100644 --- a/src/backend/core/admin.py +++ b/src/backend/core/admin.py @@ -4,12 +4,15 @@ from django.contrib.auth import admin as auth_admin from django.utils.translation import gettext_lazy as _ +from treebeard.admin import TreeAdmin + from . import models class TemplateAccessInline(admin.TabularInline): """Inline admin class for template accesses.""" + autocomplete_fields = ["user"] model = models.TemplateAccess extra = 0 @@ -111,14 +114,46 @@ class TemplateAdmin(admin.ModelAdmin): class DocumentAccessInline(admin.TabularInline): """Inline admin class for template accesses.""" + autocomplete_fields = ["user"] model = models.DocumentAccess extra = 0 @admin.register(models.Document) -class DocumentAdmin(admin.ModelAdmin): +class DocumentAdmin(TreeAdmin): """Document admin interface declaration.""" + fieldsets = ( + ( + None, + { + "fields": ( + "id", + "title", + ) + }, + ), + ( + _("Permissions"), + { + "fields": ( + "creator", + "link_reach", + "link_role", + ) + }, + ), + ( + _("Tree structure"), + { + "fields": ( + "path", + "depth", + "numchild", + ) + }, + ), + ) inlines = (DocumentAccessInline,) list_display = ( "id", @@ -128,6 +163,14 @@ class DocumentAdmin(admin.ModelAdmin): "created_at", "updated_at", ) + readonly_fields = ( + "creator", + "depth", + "id", + "numchild", + "path", + ) + search_fields = ("id", "title") @admin.register(models.Invitation) diff --git a/src/backend/core/api/permissions.py b/src/backend/core/api/permissions.py index 445a2c16e..f20801929 100644 --- a/src/backend/core/api/permissions.py +++ b/src/backend/core/api/permissions.py @@ -8,7 +8,8 @@ from core.models import DocumentAccess, RoleChoices ACTION_FOR_METHOD_TO_PERMISSION = { - "versions_detail": {"DELETE": "versions_destroy", "GET": "versions_retrieve"} + "versions_detail": {"DELETE": "versions_destroy", "GET": "versions_retrieve"}, + "children": {"GET": "children_list", "POST": "children_create"}, } diff --git a/src/backend/core/api/serializers.py b/src/backend/core/api/serializers.py index f8dbb22d2..3903ee31f 100644 --- a/src/backend/core/api/serializers.py +++ b/src/backend/core/api/serializers.py @@ -147,34 +147,54 @@ class ListDocumentSerializer(BaseResourceSerializer): is_favorite = serializers.BooleanField(read_only=True) nb_accesses = serializers.IntegerField(read_only=True) + user_roles = serializers.SerializerMethodField(read_only=True) class Meta: model = models.Document fields = [ "id", "abilities", - "content", "created_at", "creator", + "depth", + "excerpt", "is_favorite", "link_role", "link_reach", "nb_accesses", + "numchild", + "path", "title", "updated_at", + "user_roles", ] read_only_fields = [ "id", "abilities", "created_at", "creator", + "depth", + "excerpt", "is_favorite", "link_role", "link_reach", "nb_accesses", + "numchild", + "path", "updated_at", + "user_roles", ] + def get_user_roles(self, document): + """ + Return roles of the logged-in user for the current document, + taking into account ancestors. + """ + request = self.context.get("request") + if request: + return document.get_roles(request.user) + return [] + class DocumentSerializer(ListDocumentSerializer): """Serialize documents with all fields for display in detail views.""" @@ -189,23 +209,32 @@ class Meta: "content", "created_at", "creator", + "depth", + "excerpt", "is_favorite", "link_role", "link_reach", "nb_accesses", + "numchild", + "path", "title", "updated_at", + "user_roles", ] read_only_fields = [ "id", "abilities", "created_at", "creator", - "is_avorite", + "depth", + "is_favorite", "link_role", "link_reach", "nb_accesses", + "numchild", + "path", "updated_at", + "user_roles", ] def get_fields(self): @@ -281,7 +310,7 @@ def create(self, validated_data): except ConversionError as err: raise exceptions.APIException(detail="could not convert content") from err - document = models.Document.objects.create( + document = models.Document.add_root( title=validated_data["title"], content=document_content, creator=user, @@ -529,3 +558,44 @@ def validate_text(self, value): if len(value.strip()) == 0: raise serializers.ValidationError("Text field cannot be empty.") return value + + +class MoveDocumentSerializer(serializers.Serializer): + """ + Serializer for validating input data to move a document within the tree structure. + + Fields: + - target_document_id (UUIDField): The ID of the target parent document where the + document should be moved. This field is required and must be a valid UUID. + - position (ChoiceField): Specifies the position of the document in relation to + the target parent's children. + Choices: + - "first-child": Place the document as the first child of the target parent. + - "last-child": Place the document as the last child of the target parent (default). + - "left": Place the document as the left sibling of the target parent. + - "right": Place the document as the right sibling of the target parent. + + Example: + Input payload for moving a document: + { + "target_document_id": "123e4567-e89b-12d3-a456-426614174000", + "position": "first-child" + } + + Notes: + - The `target_document_id` is mandatory. + - The `position` defaults to "last-child" if not provided. + """ + + target_document_id = serializers.UUIDField(required=True) + position = serializers.ChoiceField( + choices=[ + "first-child", + "last-child", + "first-sibling", + "last-sibling", + "left", + "right", + ], + default="last-child", + ) diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index b217194d9..bfdcec23d 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -4,28 +4,25 @@ import logging import re import uuid +from datetime import timedelta from urllib.parse import urlparse from django.conf import settings from django.contrib.postgres.aggregates import ArrayAgg +from django.contrib.postgres.fields import ArrayField from django.contrib.postgres.search import TrigramSimilarity from django.core.exceptions import ValidationError from django.core.files.storage import default_storage from django.db import models as db -from django.db.models import ( - Count, - Exists, - OuterRef, - Q, - Subquery, - Value, -) +from django.db import transaction +from django.db.models.functions import Left, Length from django.http import Http404 +from django.utils import timezone import rest_framework as drf from botocore.exceptions import ClientError from django_filters import rest_framework as drf_filters -from rest_framework import filters, status +from rest_framework import filters, status, viewsets from rest_framework import response as drf_response from rest_framework.permissions import AllowAny @@ -52,7 +49,7 @@ # pylint: disable=too-many-ancestors -class NestedGenericViewSet(drf.viewsets.GenericViewSet): +class NestedGenericViewSet(viewsets.GenericViewSet): """ A generic Viewset aims to be used in a nested route context. e.g: `/api/v1.0/resource_1//resource_2//` @@ -133,7 +130,7 @@ class Pagination(drf.pagination.PageNumberPagination): class UserViewSet( - drf.mixins.UpdateModelMixin, drf.viewsets.GenericViewSet, drf.mixins.ListModelMixin + drf.mixins.UpdateModelMixin, viewsets.GenericViewSet, drf.mixins.ListModelMixin ): """User ViewSet""" @@ -301,28 +298,105 @@ def determine_metadata(self, request, view): return simple_metadata +# pylint: disable=too-many-public-methods class DocumentViewSet( drf.mixins.CreateModelMixin, drf.mixins.DestroyModelMixin, + drf.mixins.ListModelMixin, drf.mixins.UpdateModelMixin, - drf.viewsets.GenericViewSet, + viewsets.GenericViewSet, ): """ - Document ViewSet for managing documents. - - Provides endpoints for creating, updating, and deleting documents, - along with filtering options. - - Filtering: + DocumentViewSet API. + + This view set provides CRUD operations and additional actions for managing documents. + Supports filtering, ordering, and annotations for enhanced querying capabilities. + + ### API Endpoints: + 1. **List**: Retrieve a paginated list of documents. + Example: GET /documents/?page=2 + 2. **Retrieve**: Get a specific document by its ID. + Example: GET /documents/{id}/ + 3. **Create**: Create a new document. + Example: POST /documents/ + 4. **Update**: Update a document by its ID. + Example: PUT /documents/{id}/ + 5. **Delete**: Soft delete a document by its ID. + Example: DELETE /documents/{id}/ + + ### Additional Actions: + 1. **Children**: List or create child documents. + Example: GET, POST /documents/{id}/children/ + + 2. **Versions List**: Retrieve version history of a document. + Example: GET /documents/{id}/versions/ + + 3. **Version Detail**: Get or delete a specific document version. + Example: GET, DELETE /documents/{id}/versions/{version_id}/ + + 4. **Favorite**: Mark or unmark a document as favorite. + Example: POST, DELETE /documents/{id}/favorite/ + + 5. **Attachment Upload**: Upload a file attachment for the document. + Example: POST /documents/{id}/attachment-upload/ + + 6. **Create for Owner**: Create a document via server-to-server on behalf of a user. + Example: POST /documents/create-for-owner/ + + 7. **Link Configuration**: Update document link configuration. + Example: PUT /documents/{id}/link-configuration/ + + 8. **Media Auth**: Authorize access to document media. + Example: GET /documents/media-auth/ + + 9. **Media Auth**: Authorize access to the collaboration server for a document. + Example: GET /documents/collaboration-auth/ + + 10. **AI Transform**: Apply a transformation action on a piece of text with AI. + Example: POST /documents/{id}/ai-transform/ + Expected data: + - text (str): The input text. + - action (str): The transformation type, one of [prompt, correct, rephrase, summarize]. + Returns: JSON response with the processed text. + Throttled by: AIDocumentRateThrottle, AIUserRateThrottle. + + 11. **AI Translate**: Translate a piece of text with AI. + Example: POST /documents/{id}/ai-translate/ + Expected data: + - text (str): The input text. + - language (str): The target language, chosen from settings.LANGUAGES. + Returns: JSON response with the translated text. + Throttled by: AIDocumentRateThrottle, AIUserRateThrottle. + + ### Ordering: created_at, updated_at, is_favorite, title + + Example: + - Ascending: GET /api/v1.0/documents/?ordering=created_at + - Desceding: GET /api/v1.0/documents/?ordering=-title + + ### Filtering: - `is_creator_me=true`: Returns documents created by the current user. - `is_creator_me=false`: Returns documents created by other users. - `is_favorite=true`: Returns documents marked as favorite by the current user - `is_favorite=false`: Returns documents not marked as favorite by the current user + - `is_deleted=true`: Returns documents that were soft deleted left than x days ago + - `is_deleted=false`: Returns documents that were not deleted - `title=hello`: Returns documents which title contains the "hello" string - Example Usage: + Example: - GET /api/v1.0/documents/?is_creator_me=true&is_favorite=true - GET /api/v1.0/documents/?is_creator_me=false&title=hello + + ### Annotations: + 1. **nb_accesses**: Number of accesses related to the document or its ancestors. + 2. **is_favorite**: Indicates whether the document is marked as favorite by the current user. + 3. **user_roles**: Roles the current user has on the document or its ancestors. + 4. **is_traced**: Indicates if the document has been accessed by the current user. + 5. **ancestors_deleted_at**: Date when the document or one of its ancestors was soft deleted. + + ### Notes: + - Only the highest ancestor in a document hierarchy is shown in list views. + - Implements soft delete logic to retain document tree structures. """ filter_backends = [drf_filters.DjangoFilterBackend, filters.OrderingFilter] @@ -338,66 +412,148 @@ class DocumentViewSet( def get_serializer_class(self): """ - Use ListDocumentSerializer for list actions, otherwise use DocumentSerializer. + Use ListDocumentSerializer for list actions; otherwise, use DocumentSerializer. """ - if self.action == "list": - return serializers.ListDocumentSerializer - return self.serializer_class + return ( + serializers.ListDocumentSerializer + if self.action == "list" + else self.serializer_class + ) - def get_queryset(self): - """Optimize queryset to include favorite status for the current user.""" - queryset = super().get_queryset() + def annotate_nb_accesses(self, queryset): + """Annotate document queryset with number of accesses, taking into account ancestors.""" + + ancestor_accesses_query = ( + models.DocumentAccess.objects.filter( + document__path=Left(db.OuterRef("path"), Length("document__path")), + ) + .order_by() + .annotate(total_accesses=db.Func(db.Value("id"), function="COUNT")) + .values("total_accesses") + ) + return queryset.annotate(nb_accesses=db.Subquery(ancestor_accesses_query)) + + def annotate_is_favorite(self, queryset): + """ + Annotate document queryset with the favorite status for the current user. + """ user = self.request.user - # Annotate the number of accesses associated with each document - queryset = queryset.annotate(nb_accesses=Count("accesses", distinct=True)) + if user.is_authenticated: + favorite_exists_subquery = models.DocumentFavorite.objects.filter( + document_id=db.OuterRef("pk"), user=user + ) + return queryset.annotate(is_favorite=db.Exists(favorite_exists_subquery)) - if not user.is_authenticated: - # If the user is not authenticated, annotate `is_favorite` as False - return queryset.annotate(is_favorite=Value(False)) + return queryset.annotate(is_favorite=db.Value(False)) - # Annotate the queryset to indicate if the document is favorited by the current user - favorite_exists = models.DocumentFavorite.objects.filter( - document_id=OuterRef("pk"), user=user - ) - queryset = queryset.annotate(is_favorite=Exists(favorite_exists)) + def annotate_user_roles(self, queryset): + """ + Annotate document queryset with the roles of the current user + on the document or its ancestors. + """ + user = self.request.user - # Annotate the queryset with the logged-in user roles - user_roles_query = ( - models.DocumentAccess.objects.filter( - Q(user=user) | Q(team__in=user.teams), - document_id=OuterRef("pk"), + if user.is_authenticated: + user_roles_subquery = models.DocumentAccess.objects.filter( + db.Q(user=user) | db.Q(team__in=user.teams), + document__path=Left(db.OuterRef("path"), Length("document__path")), + ).values_list("role", flat=True) + + return queryset.annotate( + user_roles=db.Func(user_roles_subquery, function="ARRAY") ) - .values("document") - .annotate(roles_array=ArrayAgg("role")) - .values("roles_array") + + return queryset.annotate( + user_roles=db.Value([], output_field=ArrayField(base_field=db.CharField())), ) - return queryset.annotate(user_roles=Subquery(user_roles_query)).distinct() - def list(self, request, *args, **kwargs): - """Restrict resources returned by the list endpoint""" - queryset = self.filter_queryset(self.get_queryset()) - user = self.request.user + def get_queryset(self): + """Get queryset performing all annotation and filtering on the document tree structure.""" + request = self.request + user = request.user + is_deleted = request.GET.get("is_deleted", "false").lower() in ["true", "1"] + + queryset = super().get_queryset() + # Annotate link trace to indicate if the user has already visited the document + # and filter based on user access if user.is_authenticated: - queryset = queryset.filter( - db.Q(accesses__user=user) - | db.Q(accesses__team__in=user.teams) - | ( - db.Q(link_traces__user=user) - & ~db.Q(link_reach=models.LinkReachChoices.RESTRICTED) + if not is_deleted: + link_trace_subquery = models.LinkTrace.objects.filter( + document=db.OuterRef("pk"), user=user ) - ) + queryset = queryset.annotate(is_traced=db.Exists(link_trace_subquery)) + + if not self.detail: + queryset = queryset.filter( + db.Q(accesses__user=user) + | db.Q(accesses__team__in=user.teams) + | ( + db.Q(is_traced=True) + & ~db.Q(link_reach=models.LinkReachChoices.RESTRICTED) + ) + ) + elif self.detail: + queryset = queryset.annotate(is_traced=db.Value(False)) else: - queryset = queryset.none() + return queryset.none() + + # Among the results, we may have documents that are ancestors/children of each other + # In this case we want to keep only the highest ancestor. + # As for deletion, as soon as a document is deleted, all its descendants are considered + # deleted as well. + ancestors_subquery = queryset.filter( + path=Left(db.OuterRef("path"), Length("path")) + ) - page = self.paginate_queryset(queryset) - if page is not None: - serializer = self.get_serializer(page, many=True) - return self.get_paginated_response(serializer.data) + # Annotate with the oldest `deleted_at` date among ancestors using the `Min` function` + queryset = queryset.annotate( + ancestors_deleted_at=db.Subquery( + ancestors_subquery.filter(deleted_at__isnull=False) + .annotate(min_deleted_at=db.Min("deleted_at")) + .values("min_deleted_at")[:1] + ) + ) - serializer = self.get_serializer(queryset, many=True) - return drf.response.Response(serializer.data) + if not user.is_authenticated or (not self.detail and not is_deleted): + queryset = queryset.filter(ancestors_deleted_at__isnull=True) + + queryset = self.annotate_user_roles(queryset) + + if user.is_authenticated: + trashbin_threshold = timezone.now() - timedelta( + days=settings.SOFT_DELETE_KEEP_DAYS + ) + owner_trashbin_clause = ( + db.Q(user_roles__contains=models.RoleChoices.OWNER) + & db.Q(ancestors_deleted_at__isnull=False) + & db.Q(ancestors_deleted_at__gte=trashbin_threshold) + ) + if self.detail: + queryset = queryset.filter( + owner_trashbin_clause | db.Q(ancestors_deleted_at__isnull=True) + ) + elif is_deleted: + queryset = queryset.filter(owner_trashbin_clause) + + if not self.detail: + # Keep only documents who are the annotated highest ancestor + queryset = queryset.annotate( + root_path=db.Subquery( + ancestors_subquery.filter(deleted_at__isnull=not is_deleted) + .order_by("path") + .values("path")[:1] + ) + ).filter(root_path=db.F("path")) + + return queryset.distinct() + + def filter_queryset(self, queryset): + """Apply annotations and filters sequentially.""" + queryset = self.annotate_is_favorite(queryset) + queryset = super().filter_queryset(queryset) + return self.annotate_nb_accesses(queryset) def retrieve(self, request, *args, **kwargs): """ @@ -408,29 +564,28 @@ def retrieve(self, request, *args, **kwargs): instance = self.get_object() serializer = self.get_serializer(instance) - if self.request.user.is_authenticated: - try: - # Add a trace that the user visited the document (this is needed to include - # the document in the user's list view) - models.LinkTrace.objects.create( - document=instance, - user=self.request.user, - ) - except ValidationError: - # The trace already exists, so we just pass without doing anything - pass + if self.request.user.is_authenticated and not instance.is_traced: + models.LinkTrace.objects.create(document=instance, user=request.user) return drf.response.Response(serializer.data) def perform_create(self, serializer): """Set the current user as creator and owner of the newly created object.""" - obj = serializer.save(creator=self.request.user) + obj = models.Document.add_root( + creator=self.request.user, + **serializer.validated_data, + ) + serializer.instance = obj models.DocumentAccess.objects.create( document=obj, user=self.request.user, role=models.RoleChoices.OWNER, ) + def perform_destroy(self, instance): + """Override to implement a soft delete instead of dumping the record in database.""" + instance.soft_delete() + @drf.decorators.action( authentication_classes=[authentication.ServerToServerAuthentication], detail=False, @@ -455,6 +610,110 @@ def create_for_owner(self, request): {"id": str(document.id)}, status=status.HTTP_201_CREATED ) + @drf.decorators.action(detail=True, methods=["post"]) + def move(self, request, *args, **kwargs): + """ + Move a document to another location within the document tree. + + The user must be an administrator or owner of both the document being moved + and the target parent document. + """ + user = request.user + document = self.get_object() # including permission checks + + # Validate the input payload + serializer = serializers.MoveDocumentSerializer(data=request.data) + serializer.is_valid(raise_exception=True) + validated_data = serializer.validated_data + + target_document_id = validated_data["target_document_id"] + try: + target_document = models.Document.objects.get(id=target_document_id) + except models.Document.DoesNotExist: + return drf.response.Response( + {"target_document_id": "Target parent document does not exist."}, + status=status.HTTP_400_BAD_REQUEST, + ) + + # Check permission for the target parent document + if not target_document.get_abilities(user).get("move"): + message = "You do not have permission to move documents to this target." + return drf.response.Response( + {"target_document_id": message}, + status=status.HTTP_400_BAD_REQUEST, + ) + + document.move(target_document, pos=validated_data["position"]) + + return drf.response.Response( + {"message": "Document moved successfully."}, status=status.HTTP_200_OK + ) + + @drf.decorators.action( + detail=True, + methods=["post"], + ) + def restore(self, request, *args, **kwargs): + """ + Restore a soft-deleted document if it was deleted less than x days ago. + """ + document = self.get_object() + document.restore() + + return drf_response.Response( + {"detail": "Document has been successfully restored."}, + status=status.HTTP_200_OK, + ) + + @drf.decorators.action( + detail=True, + methods=["get", "post"], + ordering=["path"], + serializer_class=serializers.ListDocumentSerializer, + url_path="children", + ) + def children(self, request, *args, **kwargs): + """Handle listing and creating children of a document""" + document = self.get_object() + + if request.method == "POST": + # Create a child document + serializer = serializers.DocumentSerializer( + data=request.data, context=self.get_serializer_context() + ) + serializer.is_valid(raise_exception=True) + + with transaction.atomic(): + child_document = document.add_child( + creator=request.user, + **serializer.validated_data, + ) + models.DocumentAccess.objects.create( + document=child_document, + user=request.user, + role=models.RoleChoices.OWNER, + ) + # Set the created instance to the serializer + serializer.instance = child_document + + headers = self.get_success_headers(serializer.data) + return drf.response.Response( + serializer.data, status=status.HTTP_201_CREATED, headers=headers + ) + + # GET: List children + queryset = document.get_children().filter(deleted_at__isnull=True) + queryset = self.filter_queryset(queryset) + queryset = self.annotate_user_roles(queryset) + + page = self.paginate_queryset(queryset) + if page is not None: + serializer = self.get_serializer(page, many=True) + return self.get_paginated_response(serializer.data) + + serializer = self.get_serializer(queryset, many=True) + return drf.response.Response(serializer.data) + @drf.decorators.action(detail=True, methods=["get"], url_path="versions") def versions_list(self, request, *args, **kwargs): """ @@ -473,8 +732,9 @@ def versions_list(self, request, *args, **kwargs): # Users should not see version history dating from before they gained access to the # document. Filter to get the minimum access date for the logged-in user - access_queryset = document.accesses.filter( - db.Q(user=user) | db.Q(team__in=user.teams) + access_queryset = models.DocumentAccess.objects.filter( + db.Q(user=user) | db.Q(team__in=user.teams), + document__path=Left(db.Value(document.path), Length("document__path")), ).aggregate(min_date=db.Min("created_at")) # Handle the case where the user has no accesses @@ -512,10 +772,12 @@ def versions_detail(self, request, pk, version_id, *args, **kwargs): user = request.user min_datetime = min( access.created_at - for access in document.accesses.filter( + for access in models.DocumentAccess.objects.filter( db.Q(user=user) | db.Q(team__in=user.teams), + document__path=Left(db.Value(document.path), Length("document__path")), ) ) + if response["LastModified"] < min_datetime: raise Http404 @@ -800,7 +1062,7 @@ class DocumentAccessViewSet( drf.mixins.ListModelMixin, drf.mixins.RetrieveModelMixin, drf.mixins.UpdateModelMixin, - drf.viewsets.GenericViewSet, + viewsets.GenericViewSet, ): """ API ViewSet for all interactions with document accesses. @@ -873,7 +1135,7 @@ class TemplateViewSet( drf.mixins.DestroyModelMixin, drf.mixins.RetrieveModelMixin, drf.mixins.UpdateModelMixin, - drf.viewsets.GenericViewSet, + viewsets.GenericViewSet, ): """Template ViewSet""" @@ -897,14 +1159,14 @@ def get_queryset(self): user_roles_query = ( models.TemplateAccess.objects.filter( - Q(user=user) | Q(team__in=user.teams), - template_id=OuterRef("pk"), + db.Q(user=user) | db.Q(team__in=user.teams), + template_id=db.OuterRef("pk"), ) .values("template") .annotate(roles_array=ArrayAgg("role")) .values("roles_array") ) - return queryset.annotate(user_roles=Subquery(user_roles_query)).distinct() + return queryset.annotate(user_roles=db.Subquery(user_roles_query)).distinct() def list(self, request, *args, **kwargs): """Restrict templates returned by the list endpoint""" @@ -978,7 +1240,7 @@ class TemplateAccessViewSet( drf.mixins.ListModelMixin, drf.mixins.RetrieveModelMixin, drf.mixins.UpdateModelMixin, - drf.viewsets.GenericViewSet, + viewsets.GenericViewSet, ): """ API ViewSet for all interactions with template accesses. @@ -1018,7 +1280,7 @@ class InvitationViewset( drf.mixins.RetrieveModelMixin, drf.mixins.DestroyModelMixin, drf.mixins.UpdateModelMixin, - drf.viewsets.GenericViewSet, + viewsets.GenericViewSet, ): """API ViewSet for user invitations to document. diff --git a/src/backend/core/factories.py b/src/backend/core/factories.py index f1ce85909..2a6ad53a4 100644 --- a/src/backend/core/factories.py +++ b/src/backend/core/factories.py @@ -46,6 +46,23 @@ def with_owned_template(self, create, extracted, **kwargs): UserTemplateAccessFactory(user=self, role="owner") +class ParentNodeFactory(factory.declarations.ParameteredAttribute): + """Custom factory attribute for setting the parent node.""" + + def generate(self, step, params): + """ + Generate a parent node for the factory. + + This method is invoked during the factory's build process to determine the parent + node of the current object being created. If `params` is provided, it uses the factory's + metadata to recursively create or fetch the parent node. Otherwise, it returns `None`. + """ + if not params: + return None + subfactory = step.builder.factory_meta.factory + return step.recurse(subfactory, params) + + class DocumentFactory(factory.django.DjangoModelFactory): """A factory to create documents""" @@ -54,7 +71,10 @@ class Meta: django_get_or_create = ("title",) skip_postgeneration_save = True + parent = ParentNodeFactory() + title = factory.Sequence(lambda n: f"document{n}") + excerpt = factory.Sequence(lambda n: f"excerpt{n}") content = factory.Sequence(lambda n: f"content{n}") creator = factory.SubFactory(UserFactory) link_reach = factory.fuzzy.FuzzyChoice( @@ -64,6 +84,21 @@ class Meta: [r[0] for r in models.LinkRoleChoices.choices] ) + @classmethod + def _create(cls, model_class, *args, **kwargs): + """ + Custom creation logic for the factory: creates a document as a child node if + a parent is provided; otherwise, creates it as a root node. + """ + parent = kwargs.pop("parent", None) + + if parent: + # Add as a child node + return parent.add_child(instance=model_class(**kwargs)) + + # Add as a root node + return model_class.add_root(instance=model_class(**kwargs)) + @factory.post_generation def users(self, create, extracted, **kwargs): """Add users to document from a given list of users with or without roles.""" diff --git a/src/backend/core/migrations/0013_add_tree_structure_to_documents.py b/src/backend/core/migrations/0013_add_tree_structure_to_documents.py new file mode 100644 index 000000000..446e85e66 --- /dev/null +++ b/src/backend/core/migrations/0013_add_tree_structure_to_documents.py @@ -0,0 +1,31 @@ +# Generated by Django 5.1.2 on 2024-12-07 09:43 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0012_make_document_creator_and_invitation_issuer_optional'), + ] + + operations = [ + migrations.AddField( + model_name='document', + name='depth', + field=models.PositiveIntegerField(default=0), + preserve_default=False, + ), + migrations.AddField( + model_name='document', + name='numchild', + field=models.PositiveIntegerField(default=0), + ), + migrations.AddField( + model_name='document', + name='path', + # Allow null values pending the next datamigration to populate the field + field=models.CharField(max_length=255, null=True, unique=True), + preserve_default=False, + ), + ] diff --git a/src/backend/core/migrations/0014_set_path_on_existing_documents.py b/src/backend/core/migrations/0014_set_path_on_existing_documents.py new file mode 100644 index 000000000..fd9f414c5 --- /dev/null +++ b/src/backend/core/migrations/0014_set_path_on_existing_documents.py @@ -0,0 +1,52 @@ +# Generated by Django 5.1.2 on 2024-12-07 10:33 + +from django.db import migrations, models + +from treebeard.numconv import NumConv + +ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" +STEPLEN = 7 + +def set_path_on_existing_documents(apps, schema_editor): + """ + Updates the `path` and `depth` fields for all existing Document records + to ensure valid materialized paths. + + This function assigns a unique `path` to each Document as a root node + + Note: After running this migration, we quickly modify the schema to make + the `path` field required as it should. + """ + Document = apps.get_model("core", "Document") + + # Iterate over all existing documents and make them root nodes + documents = Document.objects.order_by("created_at").values_list("id", flat=True) + numconv = NumConv(len(ALPHABET), ALPHABET) + + updates = [] + for i, pk in enumerate(documents): + key = numconv.int2str(i) + path = "{0}{1}".format( + ALPHABET[0] * (STEPLEN - len(key)), + key + ) + updates.append(Document(pk=pk, path=path, depth=1)) + + # Bulk update using the prepared updates list + Document.objects.bulk_update(updates, ['depth', 'path']) + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0013_add_tree_structure_to_documents'), + ] + + operations = [ + migrations.RunPython(set_path_on_existing_documents, reverse_code=migrations.RunPython.noop), + migrations.AlterField( + model_name='document', + name='path', + field=models.CharField(max_length=255, unique=True), + ), + ] diff --git a/src/backend/core/migrations/0015_add_document_excerpt.py b/src/backend/core/migrations/0015_add_document_excerpt.py new file mode 100644 index 000000000..916e4be24 --- /dev/null +++ b/src/backend/core/migrations/0015_add_document_excerpt.py @@ -0,0 +1,23 @@ +# Generated by Django 5.1.4 on 2024-12-18 08:56 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0014_set_path_on_existing_documents'), + ] + + operations = [ + migrations.AddField( + model_name='document', + name='excerpt', + field=models.TextField(blank=True, max_length=300, null=True, verbose_name='excerpt'), + ), + migrations.AlterField( + model_name='user', + name='language', + field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'), ('de-de', 'German'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'), + ), + ] diff --git a/src/backend/core/migrations/0016_add_document_deleted_at.py b/src/backend/core/migrations/0016_add_document_deleted_at.py new file mode 100644 index 000000000..d189d138c --- /dev/null +++ b/src/backend/core/migrations/0016_add_document_deleted_at.py @@ -0,0 +1,23 @@ +# Generated by Django 5.1.4 on 2024-12-30 09:56 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('core', '0015_add_document_excerpt'), + ] + + operations = [ + migrations.AddField( + model_name='document', + name='deleted_at', + field=models.DateTimeField(blank=True, null=True), + ), + migrations.AlterField( + model_name='user', + name='language', + field=models.CharField(choices="(('en-us', 'English'), ('fr-fr', 'French'), ('de-de', 'German'))", default='en-us', help_text='The language in which the user wants to see the interface.', max_length=10, verbose_name='language'), + ), + ] diff --git a/src/backend/core/models.py b/src/backend/core/models.py index b7fb8e797..182c05111 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -1,6 +1,7 @@ """ Declare and configure the models for the impress core application """ +# pylint: disable=too-many-lines import hashlib import smtplib @@ -20,6 +21,7 @@ from django.core.files.storage import default_storage from django.core.mail import send_mail from django.db import models +from django.db.models.functions import Left, Length from django.http import FileResponse from django.template.base import Template as DjangoTemplate from django.template.context import Context @@ -35,27 +37,11 @@ import weasyprint from botocore.exceptions import ClientError from timezone_field import TimeZoneField +from treebeard.mp_tree import MP_Node logger = getLogger(__name__) -def get_resource_roles(resource, user): - """Compute the roles a user has on a resource.""" - if not user.is_authenticated: - return [] - - try: - roles = resource.user_roles or [] - except AttributeError: - try: - roles = resource.accesses.filter( - models.Q(user=user) | models.Q(team__in=user.teams), - ).values_list("role", flat=True) - except (models.ObjectDoesNotExist, IndexError): - roles = [] - return roles - - class LinkRoleChoices(models.TextChoices): """Defines the possible roles a link can offer on a document.""" @@ -336,10 +322,36 @@ def _get_abilities(self, resource, user): } -class Document(BaseModel): +class DocumentQuerySet(models.QuerySet): + """Custom queryset for Document model.""" + + def active(self): + """Return only active (non-deleted) documents.""" + return self.filter(deleted_at__isnull=True) + + def soft_deleted(self): + """Return only soft-deleted documents.""" + limit_datetime = timezone.now() - timedelta(days=settings.SOFT_DELETE_KEEP_DAYS) + return self.filter(deleted_at__isnull=False, deleted_at__gte=limit_datetime) + + def hard_deleted(self): + """Return only hard-deleted documents.""" + limit_datetime = timezone.now() - timedelta(days=settings.SOFT_DELETE_KEEP_DAYS) + return self.filter(deleted_at__isnull=False, deleted_at__lt=limit_datetime) + + def not_hard_deleted(self): + """Return active or soft-deleted documents. Used for detailed views.""" + limit_datetime = timezone.now() - timedelta(days=settings.SOFT_DELETE_KEEP_DAYS) + return self.filter( + models.Q(deleted_at__isnull=True) | models.Q(deleted_at__gte=limit_datetime) + ) + + +class Document(MP_Node, BaseModel): """Pad document carrying the content.""" title = models.CharField(_("title"), max_length=255, null=True, blank=True) + excerpt = models.TextField(_("excerpt"), max_length=300, null=True, blank=True) link_reach = models.CharField( max_length=20, choices=LinkReachChoices.choices, @@ -355,9 +367,18 @@ class Document(BaseModel): blank=True, null=True, ) + deleted_at = models.DateTimeField(null=True, blank=True) _content = None + # Tree structure + alphabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ" + steplen = 7 # nb siblings max: 3,521,614,606,208 / max depth: 255/7=36 + node_order_by = None # Manual ordering + + # Custom manager + objects = DocumentQuerySet.as_manager() + class Meta: db_table = "impress_document" ordering = ("title",) @@ -501,11 +522,46 @@ def delete_version(self, version_id): Bucket=default_storage.bucket_name, Key=self.file_key, VersionId=version_id ) + def get_roles(self, user): + """Return the roles a user has on a document.""" + if not user.is_authenticated: + return [] + + try: + roles = self.user_roles or [] + except AttributeError: + try: + roles = DocumentAccess.objects.filter( + models.Q(user=user) | models.Q(team__in=user.teams), + document__path=Left( + models.Value(self.path), Length("document__path") + ), + ).values_list("role", flat=True) + except (models.ObjectDoesNotExist, IndexError): + roles = [] + return roles + + @cached_property + def links_definitions(self): + """Get links reach/role definitions for the current document and its ancestors.""" + links_definitions = [ + {"link_reach": self.link_reach, "link_role": self.link_role} + ] + + # Ancestors links definitions are only interesting if the document is not the highest + # ancestor to which the current user has access. Look for the annotation: + if getattr(self, "root_path", None) != self.path: + links_definitions.extend( + self.get_ancestors().values("link_reach", "link_role") + ) + + return links_definitions + def get_abilities(self, user): """ Compute and return abilities for a given user on the document. """ - roles = set(get_resource_roles(self, user)) + roles = set(self.get_roles(user)) # Compute version roles before adding link roles because we don't # want anonymous users to access versions (we wouldn't know from @@ -513,15 +569,20 @@ def get_abilities(self, user): # Anonymous users should also not see document accesses has_role = bool(roles) - # Add role provided by the document link - if self.link_reach == LinkReachChoices.PUBLIC or ( - self.link_reach == LinkReachChoices.AUTHENTICATED and user.is_authenticated - ): - roles.add(self.link_role) + # Add roles provided by the document link, taking into account its ancestors + links_definitions = self.links_definitions + for lr in links_definitions: + if lr["link_reach"] == LinkReachChoices.PUBLIC: + roles.add(lr["link_role"]) + + if user.is_authenticated: + for lr in links_definitions: + if lr["link_reach"] == LinkReachChoices.AUTHENTICATED: + roles.add(lr["link_role"]) + + is_owner = RoleChoices.OWNER in roles + is_owner_or_admin = is_owner or RoleChoices.ADMIN in roles - is_owner_or_admin = bool( - roles.intersection({RoleChoices.OWNER, RoleChoices.ADMIN}) - ) can_get = bool(roles) can_update = is_owner_or_admin or RoleChoices.EDITOR in roles @@ -531,12 +592,16 @@ def get_abilities(self, user): "ai_transform": can_update, "ai_translate": can_update, "attachment_upload": can_update, + "children_list": can_get, + "children_create": can_update and user.is_authenticated, "collaboration_auth": can_get, "destroy": RoleChoices.OWNER in roles, "favorite": can_get and user.is_authenticated, "link_configuration": is_owner_or_admin, "invite_owner": RoleChoices.OWNER in roles, + "move": is_owner_or_admin, "partial_update": can_update, + "restore": is_owner, "retrieve": can_get, "media_auth": can_get, "update": can_update, @@ -603,6 +668,31 @@ def send_invitation_email(self, email, role, sender, language=None): self.send_email(subject, [email], context, language) + def soft_delete(self): + """We still keep the .delete() method untouched for programmatic purposes.""" + self.deleted_at = timezone.now() + self.save() + + def restore(self): + """Cancelling a soft delete with checks.""" + if self.deleted_at is None: + raise exceptions.ValidationError( + {"deleted_at": _("This document is not deleted.")} + ) + + limit_datetime = timezone.now() - timedelta(days=settings.SOFT_DELETE_KEEP_DAYS) + if self.deleted_at < limit_datetime: + raise exceptions.ValidationError( + { + "deleted_at": _( + "This document was hard deleted and cannot be restored." + ) + } + ) + + self.deleted_at = None + self.save() + class LinkTrace(BaseModel): """ @@ -734,11 +824,27 @@ class Meta: def __str__(self): return self.title + def get_roles(self, user): + """Return the roles a user has on a resource.""" + if not user.is_authenticated: + return [] + + try: + roles = self.user_roles or [] + except AttributeError: + try: + roles = self.accesses.filter( + models.Q(user=user) | models.Q(team__in=user.teams), + ).values_list("role", flat=True) + except (models.ObjectDoesNotExist, IndexError): + roles = [] + return roles + def get_abilities(self, user): """ Compute and return abilities for a given user on the template. """ - roles = get_resource_roles(self, user) + roles = self.get_roles(user) is_owner_or_admin = bool( set(roles).intersection({RoleChoices.OWNER, RoleChoices.ADMIN}) ) diff --git a/src/backend/core/tests/documents/test_api_document_versions.py b/src/backend/core/tests/documents/test_api_document_versions.py index e91012b29..ecb27d511 100644 --- a/src/backend/core/tests/documents/test_api_document_versions.py +++ b/src/backend/core/tests/documents/test_api_document_versions.py @@ -185,6 +185,83 @@ def test_api_document_versions_list_authenticated_related_pagination( assert content["versions"][0]["version_id"] == all_version_ids[2] +@pytest.mark.parametrize("via", VIA) +def test_api_document_versions_list_authenticated_related_pagination_parent( + via, mock_user_teams +): + """ + When a user gains access to a document's versions via an ancestor, the date of access + to the parent should be used to filter versions that were created prior to the + user gaining access to the document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory() + parent = factories.DocumentFactory(parent=grand_parent) + document = factories.DocumentFactory(parent=parent) + for i in range(3): + document.content = f"before {i:d}" + document.save() + + if via == USER: + models.DocumentAccess.objects.create( + document=grand_parent, + user=user, + role=random.choice(models.RoleChoices.choices)[0], + ) + elif via == TEAM: + mock_user_teams.return_value = ["lasuite", "unknown"] + models.DocumentAccess.objects.create( + document=grand_parent, + team="lasuite", + role=random.choice(models.RoleChoices.choices)[0], + ) + + for i in range(4): + document.content = f"after {i:d}" + document.save() + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/", + ) + + content = response.json() + + assert content["is_truncated"] is False + # The current version is not listed + assert content["count"] == 3 + assert content["next_version_id_marker"] == "" + all_version_ids = [version["version_id"] for version in content["versions"]] + + # - set page size + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/?page_size=2", + ) + + content = response.json() + assert content["count"] == 2 + assert content["is_truncated"] is True + marker = content["next_version_id_marker"] + assert marker == all_version_ids[1] + assert [ + version["version_id"] for version in content["versions"] + ] == all_version_ids[:2] + + # - get page 2 + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/?page_size=2&version_id={marker:s}", + ) + + content = response.json() + assert content["count"] == 1 + assert content["is_truncated"] is False + assert content["next_version_id_marker"] == "" + assert content["versions"][0]["version_id"] == all_version_ids[2] + + def test_api_document_versions_list_exceeds_max_page_size(): """Page size should not exceed the limit set on the serializer""" user = factories.UserFactory() @@ -314,6 +391,74 @@ def test_api_document_versions_retrieve_authenticated_related(via, mock_user_tea assert response.json()["content"] == "new content 1" +@pytest.mark.parametrize("via", VIA) +def test_api_document_versions_retrieve_authenticated_related_parent( + via, mock_user_teams +): + """ + A user who gains access to a document's versions via one of its ancestors, should be able to + retrieve the document versions. The date of access to the parent should be used to filter + versions that were created prior to the user gaining access to the document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory() + parent = factories.DocumentFactory(parent=grand_parent) + document = factories.DocumentFactory(parent=parent) + document.content = "new content" + document.save() + + assert len(document.get_versions_slice()["versions"]) == 1 + version_id = document.get_versions_slice()["versions"][0]["version_id"] + + if via == USER: + factories.UserDocumentAccessFactory(document=grand_parent, user=user) + elif via == TEAM: + mock_user_teams.return_value = ["lasuite", "unknown"] + factories.TeamDocumentAccessFactory(document=grand_parent, team="lasuite") + + time.sleep(1) # minio stores datetimes with the precision of a second + + # Versions created before the document was shared should not be seen by the user + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/", + ) + + assert response.status_code == 404 + + # Create a new version should not make it available to the user because + # only the current version is available to the user but it is excluded + # from the list + document.content = "new content 1" + document.save() + + assert len(document.get_versions_slice()["versions"]) == 2 + version_id = document.get_versions_slice()["versions"][0]["version_id"] + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/", + ) + + assert response.status_code == 404 + + # Adding one more version should make the previous version available to the user + document.content = "new content 2" + document.save() + + assert len(document.get_versions_slice()["versions"]) == 3 + version_id = document.get_versions_slice()["versions"][0]["version_id"] + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/versions/{version_id:s}/", + ) + + assert response.status_code == 200 + assert response.json()["content"] == "new content 1" + + def test_api_document_versions_create_anonymous(): """Anonymous users should not be allowed to create document versions.""" document = factories.DocumentFactory() @@ -458,15 +603,19 @@ def test_api_document_versions_update_authenticated_related(via, mock_user_teams # Delete -def test_api_document_versions_delete_anonymous(): +@pytest.mark.parametrize("reach", models.LinkReachChoices.values) +def test_api_document_versions_delete_anonymous(reach): """Anonymous users should not be allowed to destroy a document version.""" - access = factories.UserDocumentAccessFactory() + access = factories.UserDocumentAccessFactory(document__link_reach=reach) response = APIClient().delete( f"/api/v1.0/documents/{access.document_id!s}/versions/{access.id!s}/", ) assert response.status_code == 401 + assert response.json() == { + "detail": "Authentication credentials were not provided." + } @pytest.mark.parametrize("reach", models.LinkReachChoices.values) diff --git a/src/backend/core/tests/documents/test_api_documents_children_create.py b/src/backend/core/tests/documents/test_api_documents_children_create.py new file mode 100644 index 000000000..6cfb3764b --- /dev/null +++ b/src/backend/core/tests/documents/test_api_documents_children_create.py @@ -0,0 +1,251 @@ +""" +Tests for Documents API endpoint in impress's core app: create +""" + +from uuid import uuid4 + +import pytest +from rest_framework.test import APIClient + +from core import factories +from core.models import Document, LinkReachChoices, LinkRoleChoices + +pytestmark = pytest.mark.django_db + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("role", LinkRoleChoices.values) +@pytest.mark.parametrize("reach", LinkReachChoices.values) +def test_api_documents_children_create_anonymous(reach, role, depth): + """Anonymous users should not be allowed to create children documents.""" + for i in range(depth): + if i == 0: + document = factories.DocumentFactory(link_reach=reach, link_role=role) + else: + document = factories.DocumentFactory(parent=document) + + response = APIClient().post( + f"/api/v1.0/documents/{document.id!s}/children/", + { + "title": "my document", + }, + ) + + assert Document.objects.count() == depth + assert response.status_code == 401 + assert response.json() == { + "detail": "Authentication credentials were not provided." + } + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize( + "reach,role", + [ + ["restricted", "editor"], + ["restricted", "reader"], + ["public", "reader"], + ["authenticated", "reader"], + ], +) +def test_api_documents_children_create_authenticated_forbidden(reach, role, depth): + """ + Authenticated users with no write access on a document should not be allowed + to create a nested document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + for i in range(depth): + if i == 0: + document = factories.DocumentFactory(link_reach=reach, link_role=role) + else: + document = factories.DocumentFactory(parent=document, link_role="reader") + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/children/", + { + "title": "my document", + }, + ) + + assert response.status_code == 403 + assert Document.objects.count() == depth + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize( + "reach,role", + [ + ["public", "editor"], + ["authenticated", "editor"], + ], +) +def test_api_documents_children_create_authenticated_success(reach, role, depth): + """ + Authenticated users with write access on a document should be able + to create a nested document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + for i in range(depth): + if i == 0: + document = factories.DocumentFactory(link_reach=reach, link_role=role) + else: + document = factories.DocumentFactory(parent=document, link_role="reader") + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/children/", + { + "title": "my child", + }, + ) + + assert response.status_code == 201 + + child = Document.objects.get(id=response.json()["id"]) + assert child.title == "my child" + assert child.link_reach == "restricted" + assert child.accesses.filter(role="owner", user=user).exists() + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +def test_api_documents_children_create_related_forbidden(depth): + """ + Authenticated users with a specific read access on a document should not be allowed + to create a nested document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + for i in range(depth): + if i == 0: + document = factories.DocumentFactory(link_reach="restricted") + factories.UserDocumentAccessFactory( + user=user, document=document, role="reader" + ) + else: + document = factories.DocumentFactory( + parent=document, link_reach="restricted" + ) + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/children/", + { + "title": "my document", + }, + ) + + assert response.status_code == 403 + assert Document.objects.count() == depth + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("role", ["editor", "administrator", "owner"]) +def test_api_documents_children_create_related_success(role, depth): + """ + Authenticated users with a specific write access on a document should be + able to create a nested document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + for i in range(depth): + if i == 0: + document = factories.DocumentFactory(link_reach="restricted") + factories.UserDocumentAccessFactory(user=user, document=document, role=role) + else: + document = factories.DocumentFactory( + parent=document, link_reach="restricted" + ) + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/children/", + { + "title": "my child", + }, + ) + + assert response.status_code == 201 + child = Document.objects.get(id=response.json()["id"]) + assert child.title == "my child" + assert child.link_reach == "restricted" + assert child.accesses.filter(role="owner", user=user).exists() + + +def test_api_documents_children_create_authenticated_title_null(): + """It should be possible to create several nested documents with a null title.""" + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + parent = factories.DocumentFactory( + title=None, link_reach="authenticated", link_role="editor" + ) + factories.DocumentFactory(title=None, parent=parent) + + response = client.post( + f"/api/v1.0/documents/{parent.id!s}/children/", {}, format="json" + ) + + assert response.status_code == 201 + assert Document.objects.filter(title__isnull=True).count() == 3 + + +def test_api_documents_children_create_force_id_success(): + """It should be possible to force the document ID when creating a nested document.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + access = factories.UserDocumentAccessFactory(user=user, role="editor") + forced_id = uuid4() + + response = client.post( + f"/api/v1.0/documents/{access.document.id!s}/children/", + { + "id": str(forced_id), + "title": "my document", + }, + format="json", + ) + + assert response.status_code == 201 + assert Document.objects.count() == 2 + assert response.json()["id"] == str(forced_id) + + +def test_api_documents_children_create_force_id_existing(): + """ + It should not be possible to use the ID of an existing document when forcing ID on creation. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + access = factories.UserDocumentAccessFactory(user=user, role="editor") + document = factories.DocumentFactory() + + response = client.post( + f"/api/v1.0/documents/{access.document.id!s}/children/", + { + "id": str(document.id), + "title": "my document", + }, + format="json", + ) + + assert response.status_code == 400 + assert response.json() == { + "id": ["A document with this ID already exists. You cannot override it."] + } diff --git a/src/backend/core/tests/documents/test_api_documents_children_list.py b/src/backend/core/tests/documents/test_api_documents_children_list.py new file mode 100644 index 000000000..7862a2cf8 --- /dev/null +++ b/src/backend/core/tests/documents/test_api_documents_children_list.py @@ -0,0 +1,542 @@ +""" +Tests for Documents API endpoint in impress's core app: retrieve +""" + +import random + +from django.contrib.auth.models import AnonymousUser + +import pytest +from rest_framework.test import APIClient + +from core import factories + +pytestmark = pytest.mark.django_db + + +def test_api_documents_children_list_anonymous_public_standalone(): + """Anonymous users should be allowed to retrieve the children of a public documents.""" + document = factories.DocumentFactory(link_reach="public") + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/children/") + + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(AnonymousUser()), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 2, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + { + "abilities": child2.get_abilities(AnonymousUser()), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 2, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 0, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + ], + } + + +def test_api_documents_children_list_anonymous_public_parent(): + """ + Anonymous users should be allowed to retrieve the children of a document who + has a public ancestor. + """ + grand_parent = factories.DocumentFactory(link_reach="public") + parent = factories.DocumentFactory( + parent=grand_parent, link_reach=random.choice(["authenticated", "restricted"]) + ) + document = factories.DocumentFactory( + link_reach=random.choice(["authenticated", "restricted"]), parent=parent + ) + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/children/") + + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(AnonymousUser()), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 4, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + { + "abilities": child2.get_abilities(AnonymousUser()), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 4, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 0, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + ], + } + + +@pytest.mark.parametrize("reach", ["restricted", "authenticated"]) +def test_api_documents_children_list_anonymous_restricted_or_authenticated(reach): + """ + Anonymous users should not be able to retrieve children of a document that is not public. + """ + document = factories.DocumentFactory(link_reach=reach) + factories.DocumentFactory.create_batch(2, parent=document) + + response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/children/") + + assert response.status_code == 401 + assert response.json() == { + "detail": "Authentication credentials were not provided." + } + + +@pytest.mark.parametrize("reach", ["public", "authenticated"]) +def test_api_documents_children_list_authenticated_unrelated_public_or_authenticated( + reach, +): + """ + Authenticated users should be able to retrieve the children of a public/authenticated + document to which they are not related. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach=reach) + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/children/", + ) + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(user), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 2, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + { + "abilities": child2.get_abilities(user), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 2, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 0, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + ], + } + + +@pytest.mark.parametrize("reach", ["public", "authenticated"]) +def test_api_documents_children_list_authenticated_public_or_authenticated_parent( + reach, +): + """ + Authenticated users should be allowed to retrieve the children of a document who + has a public or authenticated ancestor. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory(link_reach=reach) + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(link_reach="restricted", parent=parent) + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = client.get(f"/api/v1.0/documents/{document.id!s}/children/") + + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(user), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 4, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + { + "abilities": child2.get_abilities(user), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 4, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 0, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + }, + ], + } + + +def test_api_documents_children_list_authenticated_unrelated_restricted(): + """ + Authenticated users should not be allowed to retrieve the children of a document that is + restricted and to which they are not related. + """ + user = factories.UserFactory(with_owned_document=True) + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + child1, _child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/children/", + ) + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." + } + + +def test_api_documents_children_list_authenticated_related_direct(): + """ + Authenticated users should be allowed to retrieve the children of a document + to which they are directly related whatever the role. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory() + access = factories.UserDocumentAccessFactory(document=document, user=user) + factories.UserDocumentAccessFactory(document=document) + + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/children/", + ) + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(user), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 2, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 3, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + }, + { + "abilities": child2.get_abilities(user), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 2, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 2, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + }, + ], + } + + +def test_api_documents_children_list_authenticated_related_parent(): + """ + Authenticated users should be allowed to retrieve the children of a document if they + are related to one of its ancestors whatever the role. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory(link_reach="restricted") + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + factories.UserDocumentAccessFactory(document=child1) + + grand_parent_access = factories.UserDocumentAccessFactory( + document=grand_parent, user=user + ) + factories.UserDocumentAccessFactory(document=grand_parent) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/children/", + ) + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(user), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 4, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 3, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [grand_parent_access.role], + }, + { + "abilities": child2.get_abilities(user), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 4, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 2, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [grand_parent_access.role], + }, + ], + } + + +def test_api_documents_children_list_authenticated_related_child(): + """ + Authenticated users should not be allowed to retrieve all the children of a document + as a result of being related to one of its children. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + child1, _child2 = factories.DocumentFactory.create_batch(2, parent=document) + + factories.UserDocumentAccessFactory(document=child1, user=user) + factories.UserDocumentAccessFactory(document=document) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/children/", + ) + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." + } + + +def test_api_documents_children_list_authenticated_related_team_none(mock_user_teams): + """ + Authenticated users should not be able to retrieve the children of a restricted document + related to teams in which the user is not. + """ + mock_user_teams.return_value = [] + + user = factories.UserFactory(with_owned_document=True) + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + factories.DocumentFactory.create_batch(2, parent=document) + + factories.TeamDocumentAccessFactory(document=document, team="myteam") + + response = client.get(f"/api/v1.0/documents/{document.id!s}/children/") + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." + } + + +def test_api_documents_children_list_authenticated_related_team_members( + mock_user_teams, +): + """ + Authenticated users should be allowed to retrieve the children of a document to which they + are related via a team whatever the role. + """ + mock_user_teams.return_value = ["myteam"] + + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + child1, child2 = factories.DocumentFactory.create_batch(2, parent=document) + + access = factories.TeamDocumentAccessFactory(document=document, team="myteam") + + response = client.get(f"/api/v1.0/documents/{document.id!s}/children/") + + # pylint: disable=R0801 + assert response.status_code == 200 + assert response.json() == { + "count": 2, + "next": None, + "previous": None, + "results": [ + { + "abilities": child1.get_abilities(user), + "created_at": child1.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child1.creator.id), + "depth": 2, + "excerpt": child1.excerpt, + "id": str(child1.id), + "is_favorite": False, + "link_reach": child1.link_reach, + "link_role": child1.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child1.path, + "title": child1.title, + "updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + }, + { + "abilities": child2.get_abilities(user), + "created_at": child2.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(child2.creator.id), + "depth": 2, + "excerpt": child2.excerpt, + "id": str(child2.id), + "is_favorite": False, + "link_reach": child2.link_reach, + "link_role": child2.link_role, + "numchild": 0, + "nb_accesses": 1, + "path": child2.path, + "title": child2.title, + "updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + }, + ], + } diff --git a/src/backend/core/tests/documents/test_api_documents_delete.py b/src/backend/core/tests/documents/test_api_documents_delete.py index ba12c8ff6..f8cd2205c 100644 --- a/src/backend/core/tests/documents/test_api_documents_delete.py +++ b/src/backend/core/tests/documents/test_api_documents_delete.py @@ -77,6 +77,38 @@ def test_api_documents_delete_authenticated_not_owner(via, role, mock_user_teams assert models.Document.objects.count() == 2 +@pytest.mark.parametrize("depth", [1, 2, 3]) +def test_api_documents_delete_authenticated_owner_of_ancestor(depth): + """ + Authenticated users should not be able to delete a document for which + they are only owner of an ancestor. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + documents = [] + for i in range(depth): + documents.append( + factories.UserDocumentAccessFactory(role="owner", user=user).document + if i == 0 + else factories.DocumentFactory(parent=documents[-1]) + ) + assert models.Document.objects.count() == depth + + response = client.delete( + f"/api/v1.0/documents/{documents[-1].id}/", + ) + + assert response.status_code == 204 + assert models.Document.objects.active().count() == depth - 1 + + # Make sure it is only a soft delete + assert models.Document.objects.soft_deleted().exists() is True + assert models.Document.objects.not_hard_deleted().exists() is True + assert models.Document.objects.hard_deleted().exists() is False + + @pytest.mark.parametrize("via", VIA) def test_api_documents_delete_authenticated_owner(via, mock_user_teams): """ @@ -101,4 +133,9 @@ def test_api_documents_delete_authenticated_owner(via, mock_user_teams): ) assert response.status_code == 204 - assert models.Document.objects.exists() is False + assert models.Document.objects.active().exists() is False + + # Make sure it is only a soft delete + assert models.Document.objects.soft_deleted().exists() is True + assert models.Document.objects.not_hard_deleted().exists() is True + assert models.Document.objects.hard_deleted().exists() is False diff --git a/src/backend/core/tests/documents/test_api_documents_list.py b/src/backend/core/tests/documents/test_api_documents_list.py index 8793b34f6..6e0c762e5 100644 --- a/src/backend/core/tests/documents/test_api_documents_list.py +++ b/src/backend/core/tests/documents/test_api_documents_list.py @@ -2,10 +2,11 @@ Tests for Documents API endpoint in impress's core app: list """ -import operator import random +from datetime import timedelta from unittest import mock -from urllib.parse import urlencode + +from django.utils import timezone import pytest from faker import Faker @@ -37,16 +38,16 @@ def test_api_documents_list_anonymous(reach, role): def test_api_documents_list_format(): """Validate the format of documents as returned by the list view.""" user = factories.UserFactory() - client = APIClient() client.force_login(user) other_users = factories.UserFactory.create_batch(3) document = factories.DocumentFactory( - users=[user, *factories.UserFactory.create_batch(2)], + users=factories.UserFactory.create_batch(2), favorited_by=[user, *other_users], link_traces=other_users, ) + access = factories.UserDocumentAccessFactory(document=document, user=user) response = client.get("/api/v1.0/documents/") @@ -62,15 +63,19 @@ def test_api_documents_list_format(): assert results[0] == { "id": str(document.id), "abilities": document.get_abilities(user), - "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": True, "link_reach": document.link_reach, "link_role": document.link_role, "nb_accesses": 3, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], } @@ -81,11 +86,10 @@ def test_api_documents_list_authenticated_direct(django_assert_num_queries): than restricted. """ user = factories.UserFactory() - client = APIClient() client.force_login(user) - documents = [ + document1, document2 = [ access.document for access in factories.UserDocumentAccessFactory.create_batch(2, user=user) ] @@ -95,16 +99,49 @@ def test_api_documents_list_authenticated_direct(django_assert_num_queries): for role in models.LinkRoleChoices: factories.DocumentFactory(link_reach=reach, link_role=role) - expected_ids = {str(document.id) for document in documents} + # Children of visible documents should not get listed even with a specific access + factories.DocumentFactory(parent=document1) + + child1_with_access = factories.DocumentFactory(parent=document1) + factories.UserDocumentAccessFactory(user=user, document=child1_with_access) + + middle_document = factories.DocumentFactory(parent=document2) + child2_with_access = factories.DocumentFactory(parent=middle_document) + factories.UserDocumentAccessFactory(user=user, document=child2_with_access) + + # Children of hidden documents should get listed when visible by the logged-in user + hidden_root = factories.DocumentFactory() + child3_with_access = factories.DocumentFactory(parent=hidden_root) + factories.UserDocumentAccessFactory(user=user, document=child3_with_access) + + # Documents that are soft deleted and children of a soft deleted document should not be listed + soft_deleted_document = factories.DocumentFactory( + users=[user], deleted_at=timezone.now() + ) + child_of_soft_deleted_document = factories.DocumentFactory( + users=[user], parent=soft_deleted_document + ) + factories.DocumentFactory(users=[user], parent=child_of_soft_deleted_document) + + # Documents that are hard deleted and children of a hard deleted document should not be listed + hard_deleted_document = factories.DocumentFactory( + users=[user], deleted_at=timezone.now() - timedelta(days=40) + ) + child_of_hard_deleted_document = factories.DocumentFactory( + users=[user], parent=hard_deleted_document + ) + factories.DocumentFactory(users=[user], parent=child_of_hard_deleted_document) + + expected_ids = {str(document1.id), str(document2.id), str(child3_with_access.id)} with django_assert_num_queries(3): response = client.get("/api/v1.0/documents/") assert response.status_code == 200 results = response.json()["results"] - assert len(results) == 2 - results_id = {result["id"] for result in results} - assert expected_ids == results_id + results_ids = {result["id"] for result in results} + assert len(results) == 3 + assert expected_ids == results_ids def test_api_documents_list_authenticated_via_team( @@ -186,12 +223,27 @@ def test_api_documents_list_authenticated_link_reach_public_or_authenticated( client = APIClient() client.force_login(user) - documents = [ + document1, document2 = [ factories.DocumentFactory(link_traces=[user], link_reach=reach) for reach in models.LinkReachChoices if reach != "restricted" ] - expected_ids = {str(document.id) for document in documents} + factories.DocumentFactory( + link_reach=random.choice(["public", "authenticated"]), + link_traces=[user], + parent=document1, + ) + + hidden_document = factories.DocumentFactory( + link_reach=random.choice(["public", "authenticated"]) + ) + visible_child = factories.DocumentFactory( + link_traces=[user], + link_reach=random.choice(["public", "authenticated"]), + parent=hidden_document, + ) + + expected_ids = {str(document1.id), str(document2.id), str(visible_child.id)} with django_assert_num_queries(3): response = client.get( @@ -200,7 +252,6 @@ def test_api_documents_list_authenticated_link_reach_public_or_authenticated( assert response.status_code == 200 results = response.json()["results"] - assert len(results) == 2 results_id = {result["id"] for result in results} assert expected_ids == results_id @@ -314,361 +365,3 @@ def test_api_documents_list_favorites_no_extra_queries(django_assert_num_queries assert result["is_favorite"] is True else: assert result["is_favorite"] is False - - -def test_api_documents_list_filter_and_access_rights(): - """Filtering on querystring parameters should respect access rights.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - other_user = factories.UserFactory() - - def random_favorited_by(): - return random.choice([[], [user], [other_user]]) - - # Documents that should be listed to this user - listed_documents = [ - factories.DocumentFactory( - link_reach="public", - link_traces=[user], - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ), - factories.DocumentFactory( - link_reach="authenticated", - link_traces=[user], - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ), - factories.DocumentFactory( - link_reach="restricted", - users=[user], - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ), - ] - listed_ids = [str(doc.id) for doc in listed_documents] - word_list = [word for doc in listed_documents for word in doc.title.split(" ")] - - # Documents that should not be listed to this user - factories.DocumentFactory( - link_reach="public", - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ) - factories.DocumentFactory( - link_reach="authenticated", - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ) - factories.DocumentFactory( - link_reach="restricted", - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ) - factories.DocumentFactory( - link_reach="restricted", - link_traces=[user], - favorited_by=random_favorited_by(), - creator=random.choice([user, other_user]), - ) - - filters = { - "link_reach": random.choice([None, *models.LinkReachChoices.values]), - "title": random.choice([None, *word_list]), - "favorite": random.choice([None, True, False]), - "creator": random.choice([None, user, other_user]), - "ordering": random.choice( - [ - None, - "created_at", - "-created_at", - "is_favorite", - "-is_favorite", - "nb_accesses", - "-nb_accesses", - "title", - "-title", - "updated_at", - "-updated_at", - ] - ), - } - query_params = {key: value for key, value in filters.items() if value is not None} - querystring = urlencode(query_params) - - response = client.get(f"/api/v1.0/documents/?{querystring:s}") - - assert response.status_code == 200 - results = response.json()["results"] - - # Ensure all documents in results respect expected access rights - for result in results: - assert result["id"] in listed_ids - - -# Filters: ordering - - -def test_api_documents_list_ordering_default(): - """Documents should be ordered by descending "updated_at" by default""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(5, users=[user]) - - response = client.get("/api/v1.0/documents/") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 5 - - # Check that results are sorted by descending "updated_at" as expected - for i in range(4): - assert operator.ge(results[i]["updated_at"], results[i + 1]["updated_at"]) - - -def test_api_documents_list_ordering_by_fields(): - """It should be possible to order by several fields""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(5, users=[user]) - - for parameter in [ - "created_at", - "-created_at", - "is_favorite", - "-is_favorite", - "nb_accesses", - "-nb_accesses", - "title", - "-title", - "updated_at", - "-updated_at", - ]: - is_descending = parameter.startswith("-") - field = parameter.lstrip("-") - querystring = f"?ordering={parameter}" - - response = client.get(f"/api/v1.0/documents/{querystring:s}") - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 5 - - # Check that results are sorted by the field in querystring as expected - compare = operator.ge if is_descending else operator.le - for i in range(4): - assert compare(results[i][field], results[i + 1][field]) - - -# Filters: is_creator_me - - -def test_api_documents_list_filter_is_creator_me_true(): - """ - Authenticated users should be able to filter documents they created. - """ - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], creator=user) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_creator_me=true") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 3 - - # Ensure all results are created by the current user - for result in results: - assert result["creator"] == str(user.id) - - -def test_api_documents_list_filter_is_creator_me_false(): - """ - Authenticated users should be able to filter documents created by others. - """ - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], creator=user) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_creator_me=false") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 2 - - # Ensure all results are created by other users - for result in results: - assert result["creator"] != str(user.id) - - -def test_api_documents_list_filter_is_creator_me_invalid(): - """Filtering with an invalid `is_creator_me` value should do nothing.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], creator=user) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_creator_me=invalid") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 5 - - -# Filters: is_favorite - - -def test_api_documents_list_filter_is_favorite_true(): - """ - Authenticated users should be able to filter documents they marked as favorite. - """ - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_favorite=true") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 3 - - # Ensure all results are marked as favorite by the current user - for result in results: - assert result["is_favorite"] is True - - -def test_api_documents_list_filter_is_favorite_false(): - """ - Authenticated users should be able to filter documents they didn't mark as favorite. - """ - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_favorite=false") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 2 - - # Ensure all results are not marked as favorite by the current user - for result in results: - assert result["is_favorite"] is False - - -def test_api_documents_list_filter_is_favorite_invalid(): - """Filtering with an invalid `is_favorite` value should do nothing.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) - factories.DocumentFactory.create_batch(2, users=[user]) - - response = client.get("/api/v1.0/documents/?is_favorite=invalid") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == 5 - - -# Filters: link_reach - - -@pytest.mark.parametrize("reach", models.LinkReachChoices.values) -def test_api_documents_list_filter_link_reach(reach): - """Authenticated users should be able to filter documents by link reach.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(5, users=[user]) - - response = client.get(f"/api/v1.0/documents/?link_reach={reach:s}") - - assert response.status_code == 200 - results = response.json()["results"] - - # Ensure all results have the chosen link reach - for result in results: - assert result["link_reach"] == reach - - -def test_api_documents_list_filter_link_reach_invalid(): - """Filtering with an invalid `link_reach` value should raise an error.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - factories.DocumentFactory.create_batch(3, users=[user]) - - response = client.get("/api/v1.0/documents/?link_reach=invalid") - - assert response.status_code == 400 - assert response.json() == { - "link_reach": [ - "Select a valid choice. invalid is not one of the available choices." - ] - } - - -# Filters: title - - -@pytest.mark.parametrize( - "query,nb_results", - [ - ("Project Alpha", 1), # Exact match - ("project", 2), # Partial match (case-insensitive) - ("Guide", 1), # Word match within a title - ("Special", 0), # No match (nonexistent keyword) - ("2024", 2), # Match by numeric keyword - ("", 5), # Empty string - ], -) -def test_api_documents_list_filter_title(query, nb_results): - """Authenticated users should be able to search documents by their title.""" - user = factories.UserFactory() - client = APIClient() - client.force_login(user) - - # Create documents with predefined titles - titles = [ - "Project Alpha Documentation", - "Project Beta Overview", - "User Guide", - "Financial Report 2024", - "Annual Review 2024", - ] - for title in titles: - factories.DocumentFactory(title=title, users=[user]) - - # Perform the search query - response = client.get(f"/api/v1.0/documents/?title={query:s}") - - assert response.status_code == 200 - results = response.json()["results"] - assert len(results) == nb_results - - # Ensure all results contain the query in their title - for result in results: - assert query.lower().strip() in result["title"].lower() diff --git a/src/backend/core/tests/documents/test_api_documents_list_filters.py b/src/backend/core/tests/documents/test_api_documents_list_filters.py new file mode 100644 index 000000000..d9f557173 --- /dev/null +++ b/src/backend/core/tests/documents/test_api_documents_list_filters.py @@ -0,0 +1,488 @@ +""" +Tests for Documents API endpoint in impress's core app: list +""" + +import operator +import random +from datetime import timedelta +from urllib.parse import urlencode + +from django.utils import timezone + +import pytest +from faker import Faker +from rest_framework.test import APIClient + +from core import factories, models + +fake = Faker() +pytestmark = pytest.mark.django_db + + +def test_api_documents_list_filter_and_access_rights(): + """Filtering on querystring parameters should respect access rights.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + other_user = factories.UserFactory() + + def random_favorited_by(): + return random.choice([[], [user], [other_user]]) + + # Documents that should be listed to this user + listed_documents = [ + factories.DocumentFactory( + link_reach="public", + link_traces=[user], + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ), + factories.DocumentFactory( + link_reach="authenticated", + link_traces=[user], + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ), + factories.DocumentFactory( + link_reach="restricted", + users=[user], + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ), + ] + listed_ids = [str(doc.id) for doc in listed_documents] + word_list = [word for doc in listed_documents for word in doc.title.split(" ")] + + # Documents that should not be listed to this user + factories.DocumentFactory( + link_reach="public", + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ) + factories.DocumentFactory( + link_reach="authenticated", + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ) + factories.DocumentFactory( + link_reach="restricted", + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ) + factories.DocumentFactory( + link_reach="restricted", + link_traces=[user], + favorited_by=random_favorited_by(), + creator=random.choice([user, other_user]), + ) + + filters = { + "link_reach": random.choice([None, *models.LinkReachChoices.values]), + "title": random.choice([None, *word_list]), + "favorite": random.choice([None, True, False]), + "creator": random.choice([None, user, other_user]), + "ordering": random.choice( + [ + None, + "created_at", + "-created_at", + "is_favorite", + "-is_favorite", + "nb_accesses", + "-nb_accesses", + "title", + "-title", + "updated_at", + "-updated_at", + ] + ), + } + query_params = {key: value for key, value in filters.items() if value is not None} + querystring = urlencode(query_params) + + response = client.get(f"/api/v1.0/documents/?{querystring:s}") + + assert response.status_code == 200 + results = response.json()["results"] + + # Ensure all documents in results respect expected access rights + for result in results: + assert result["id"] in listed_ids + + +# Filters: ordering + + +def test_api_documents_list_ordering_default(): + """Documents should be ordered by descending "updated_at" by default""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(5, users=[user]) + + response = client.get("/api/v1.0/documents/") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 5 + + # Check that results are sorted by descending "updated_at" as expected + for i in range(4): + assert operator.ge(results[i]["updated_at"], results[i + 1]["updated_at"]) + + +def test_api_documents_list_ordering_by_fields(): + """It should be possible to order by several fields""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(5, users=[user]) + + for parameter in [ + "created_at", + "-created_at", + "is_favorite", + "-is_favorite", + "nb_accesses", + "-nb_accesses", + "title", + "-title", + "updated_at", + "-updated_at", + ]: + is_descending = parameter.startswith("-") + field = parameter.lstrip("-") + querystring = f"?ordering={parameter}" + + response = client.get(f"/api/v1.0/documents/{querystring:s}") + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 5 + + # Check that results are sorted by the field in querystring as expected + compare = operator.ge if is_descending else operator.le + for i in range(4): + assert compare(results[i][field], results[i + 1][field]) + + +# Filters: is_creator_me + + +def test_api_documents_list_filter_is_creator_me_true(): + """ + Authenticated users should be able to filter documents they created. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], creator=user) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_creator_me=true") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 3 + + # Ensure all results are created by the current user + for result in results: + assert result["creator"] == str(user.id) + + +def test_api_documents_list_filter_is_creator_me_false(): + """ + Authenticated users should be able to filter documents created by others. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], creator=user) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_creator_me=false") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 2 + + # Ensure all results are created by other users + for result in results: + assert result["creator"] != str(user.id) + + +def test_api_documents_list_filter_is_creator_me_invalid(): + """Filtering with an invalid `is_creator_me` value should do nothing.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], creator=user) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_creator_me=invalid") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 5 + + +# Filters: is_favorite + + +def test_api_documents_list_filter_is_favorite_true(): + """ + Authenticated users should be able to filter documents they marked as favorite. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_favorite=true") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 3 + + # Ensure all results are marked as favorite by the current user + for result in results: + assert result["is_favorite"] is True + + +def test_api_documents_list_filter_is_favorite_false(): + """ + Authenticated users should be able to filter documents they didn't mark as favorite. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_favorite=false") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 2 + + # Ensure all results are not marked as favorite by the current user + for result in results: + assert result["is_favorite"] is False + + +def test_api_documents_list_filter_is_favorite_invalid(): + """Filtering with an invalid `is_favorite` value should do nothing.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user], favorited_by=[user]) + factories.DocumentFactory.create_batch(2, users=[user]) + + response = client.get("/api/v1.0/documents/?is_favorite=invalid") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 5 + + +# Filters: link_reach + + +@pytest.mark.parametrize("reach", models.LinkReachChoices.values) +def test_api_documents_list_filter_link_reach(reach): + """Authenticated users should be able to filter documents by link reach.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(5, users=[user]) + + response = client.get(f"/api/v1.0/documents/?link_reach={reach:s}") + + assert response.status_code == 200 + results = response.json()["results"] + + # Ensure all results have the chosen link reach + for result in results: + assert result["link_reach"] == reach + + +def test_api_documents_list_filter_link_reach_invalid(): + """Filtering with an invalid `link_reach` value should raise an error.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(3, users=[user]) + + response = client.get("/api/v1.0/documents/?link_reach=invalid") + + assert response.status_code == 400 + assert response.json() == { + "link_reach": [ + "Select a valid choice. invalid is not one of the available choices." + ] + } + + +# Filters: title + + +@pytest.mark.parametrize( + "query,nb_results", + [ + ("Project Alpha", 1), # Exact match + ("project", 2), # Partial match (case-insensitive) + ("Guide", 1), # Word match within a title + ("Special", 0), # No match (nonexistent keyword) + ("2024", 2), # Match by numeric keyword + ("", 5), # Empty string + ], +) +def test_api_documents_list_filter_title(query, nb_results): + """Authenticated users should be able to search documents by their title.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + # Create documents with predefined titles + titles = [ + "Project Alpha Documentation", + "Project Beta Overview", + "User Guide", + "Financial Report 2024", + "Annual Review 2024", + ] + for title in titles: + factories.DocumentFactory(title=title, users=[user]) + + # Perform the search query + response = client.get(f"/api/v1.0/documents/?title={query:s}") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == nb_results + + # Ensure all results contain the query in their title + for result in results: + assert query.lower().strip() in result["title"].lower() + + +# Filters: is_deleted + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("role", models.RoleChoices.values) +@pytest.mark.parametrize( + "querystring", ["?is_deleted=true", "?is_deleted=True", "?is_deleted=1"] +) +def test_api_documents_list_filter_is_deleted_true(querystring, role, depth): + """ + Authenticated users should be able to filter documents in the trashbin if + they are an owner of the document. + (soft deleted for a period shorter than the limit configured in settings) + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + tree = [] + tree_with_delete = [] + for i in range(depth): + tree.append( + factories.UserDocumentAccessFactory(role=role, user=user).document + if i == 0 + else factories.DocumentFactory(parent=tree[-1]) + ) + tree_with_delete.append( + factories.UserDocumentAccessFactory(role=role, user=user).document + if i == 0 + else factories.DocumentFactory(parent=tree_with_delete[-1]) + ) + + # Soft delete a document + now = timezone.now() + deleted_document = random.choice(tree_with_delete) + deleted_document.deleted_at = now - timedelta(days=15) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{querystring:s}") + + assert response.status_code == 200 + results = response.json()["results"] + + if role == "owner": + assert len(results) == 1 + assert results[0]["id"] == str(deleted_document.id) + else: + assert len(results) == 0 + + # Hard delete the document + deleted_document.deleted_at = now - timedelta(days=40) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{querystring:s}") + + assert response.status_code == 200 + results = response.json()["results"] + + assert len(results) == 0 + + +@pytest.mark.parametrize( + "querystring", ["", "?is_deleted=false", "?is_deleted=False", "?is_deleted=0"] +) +def test_api_documents_list_filter_is_deleted_false(querystring): + """ + Authenticated users should be able to filter documents that are not deleted. + It should be the default filter for `is_deleted`. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + doc_soft, doc_hard, *documents = factories.DocumentFactory.create_batch( + 5, users=[user] + ) + expected_ids = {str(document.id) for document in documents} + + # Soft delete a document + now = timezone.now() + doc_soft.deleted_at = now - timedelta(days=15) + doc_soft.save() + + # Hard delete a document + doc_hard.deleted_at = now - timedelta(days=40) + doc_hard.save() + + response = client.get(f"/api/v1.0/documents/{querystring:s}") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 3 + results_ids = {result["id"] for result in results} + assert results_ids == expected_ids + + +def test_api_documents_list_filter_is_deleted_invalid(): + """Filtering with an invalid `is_deleted` value should do nothing.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + factories.DocumentFactory.create_batch(5, users=[user]) + + response = client.get("/api/v1.0/documents/?is_deleted=invalid") + + assert response.status_code == 200 + results = response.json()["results"] + assert len(results) == 5 diff --git a/src/backend/core/tests/documents/test_api_documents_move.py b/src/backend/core/tests/documents/test_api_documents_move.py new file mode 100644 index 000000000..7b70dc034 --- /dev/null +++ b/src/backend/core/tests/documents/test_api_documents_move.py @@ -0,0 +1,199 @@ +""" +Test moving documents within the document tree via an detail action API endpoint. +""" + +from uuid import uuid4 + +import pytest +from rest_framework.test import APIClient + +from core import factories + +pytestmark = pytest.mark.django_db + + +def test_api_documents_move_anonymous_user(): + """Anonymous users should not be able to move documents.""" + document = factories.DocumentFactory() + target = factories.DocumentFactory() + + response = APIClient().post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": str(target.id)}, + ) + + assert response.status_code == 401 + assert response.json() == { + "detail": "Authentication credentials were not provided." + } + + +@pytest.mark.parametrize("role", [None, "reader", "editor"]) +def test_api_documents_move_authenticated_document_no_permission(role): + """ + Authenticated users should not be able to move documents with insufficient + permissions on the origin document. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory() + target = factories.UserDocumentAccessFactory(user=user, role="owner").document + + if role: + factories.UserDocumentAccessFactory(document=document, user=user, role=role) + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": str(target.id)}, + ) + + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." + } + + +@pytest.mark.parametrize("role", [None, "reader", "editor"]) +def test_api_documents_move_authenticated_target_no_permission(role): + """ + Authenticated users should not be able to move documents with insufficient + permissions on the origin document. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.UserDocumentAccessFactory(user=user, role="owner").document + target = factories.DocumentFactory() + + if role: + factories.UserDocumentAccessFactory(document=target, user=user, role=role) + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": str(target.id)}, + ) + + assert response.status_code == 400 + assert response.json() == { + "target_document_id": "You do not have permission to move documents to this target." + } + + +def test_api_documents_move_invalid_target_string(): + """Test for moving a document to an invalid target as a random string.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.UserDocumentAccessFactory(user=user, role="owner").document + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": "non-existent-id"}, + ) + + assert response.status_code == 400 + assert response.json() == {"target_document_id": ["Must be a valid UUID."]} + + +def test_api_documents_move_invalid_target_uuid(): + """Test for moving a document to an invalid target that looks like a UUID.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.UserDocumentAccessFactory(user=user, role="owner").document + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": str(uuid4())}, + ) + + assert response.status_code == 400 + assert response.json() == { + "target_document_id": "Target parent document does not exist." + } + + +def test_api_documents_move_invalid_position(): + """Test moving a document to an invalid position.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.UserDocumentAccessFactory(user=user, role="owner").document + target = factories.UserDocumentAccessFactory(user=user, role="owner").document + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={ + "target_document_id": str(target.id), + "position": "invalid-position", + }, + ) + + assert response.status_code == 400 + assert response.json() == { + "position": ['"invalid-position" is not a valid choice.'] + } + + +@pytest.mark.parametrize( + "position", + ["first-child", "last-child", "first-sibling", "last-sibling", "left", "right"], +) +@pytest.mark.parametrize("target_role", ["administrator", "owner"]) +@pytest.mark.parametrize("role", ["administrator", "owner"]) +def test_api_documents_move_authenticated_success(role, target_role, position): + """ + Authenticated users with sufficient permissions should be able to move documents + as a child of the target document. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.UserDocumentAccessFactory(user=user, role=role).document + children = factories.DocumentFactory.create_batch(3, parent=document) + + target_parent = factories.UserDocumentAccessFactory( + user=user, role=target_role + ).document + sibling1, target, sibling2 = factories.DocumentFactory.create_batch( + 3, parent=target_parent + ) + target_children = factories.DocumentFactory.create_batch(3, parent=target) + + response = client.post( + f"/api/v1.0/documents/{document.id!s}/move/", + data={"target_document_id": str(target.id), "position": position}, + ) + + assert response.status_code == 200 + assert response.json() == {"message": "Document moved successfully."} + + # Verify that the document has moved as expected in the tree + document.refresh_from_db() + target.refresh_from_db() + + match position: + case "first-child": + assert list(target.get_children()) == [document, *target_children] + case "last-child": + assert list(target.get_children()) == [*target_children, document] + case "first-sibling": + assert list(target.get_siblings()) == [document, sibling1, target, sibling2] + case "last-sibling": + assert list(target.get_siblings()) == [sibling1, target, sibling2, document] + case "left": + assert list(target.get_siblings()) == [sibling1, document, target, sibling2] + case "right": + assert list(target.get_siblings()) == [sibling1, target, document, sibling2] + case _: + raise ValueError(f"Invalid position: {position}") + + # Verify that the document's children have also been moved + assert list(document.get_children()) == children diff --git a/src/backend/core/tests/documents/test_api_documents_restore.py b/src/backend/core/tests/documents/test_api_documents_restore.py new file mode 100644 index 000000000..805319e8b --- /dev/null +++ b/src/backend/core/tests/documents/test_api_documents_restore.py @@ -0,0 +1,84 @@ +""" +Test restoring documents after a soft delete via the detail action API endpoint. +""" + +from datetime import timedelta + +from django.utils import timezone + +import pytest +from rest_framework.test import APIClient + +from core import factories, models + +pytestmark = pytest.mark.django_db + + +def test_api_documents_restore_anonymous_user(): + """Anonymous users should not be able to restore deleted documents.""" + now = timezone.now() - timedelta(days=15) + document = factories.DocumentFactory(deleted_at=now) + + response = APIClient().post(f"/api/v1.0/documents/{document.id!s}/restore/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + assert models.Document.objects.get().deleted_at == now + + +@pytest.mark.parametrize("role", [None, "reader", "editor", "administrator"]) +def test_api_documents_restore_authenticated_no_permission(role): + """ + Authenticated users who are not owners of a deleted document should + not be able to restore it. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + now = timezone.now() - timedelta(days=15) + document = factories.DocumentFactory( + deleted_at=now, link_reach="public", link_role="editor" + ) + if role: + factories.UserDocumentAccessFactory(document=document, user=user, role=role) + + response = client.post(f"/api/v1.0/documents/{document.id!s}/restore/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + + +def test_api_documents_restore_authenticated_success(): + """The owner of a deleted document should be able to restore it.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + now = timezone.now() - timedelta(days=15) + document = factories.DocumentFactory(deleted_at=now) + factories.UserDocumentAccessFactory(document=document, user=user, role="owner") + + response = client.post(f"/api/v1.0/documents/{document.id!s}/restore/") + + assert response.status_code == 200 + assert response.json() == {"detail": "Document has been successfully restored."} + + document.refresh_from_db() + assert document.deleted_at is None + + +def test_api_documents_restore_expired(): + """Attempting to restore a document deleted beyond the allowed time frame should fail.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + now = timezone.now() - timedelta(days=40) + document = factories.DocumentFactory(deleted_at=now) + factories.UserDocumentAccessFactory(document=document, user=user, role="owner") + + response = client.post(f"/api/v1.0/documents/{document.id!s}/restore/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} diff --git a/src/backend/core/tests/documents/test_api_documents_retrieve.py b/src/backend/core/tests/documents/test_api_documents_retrieve.py index 4307ed69e..e81ebfc26 100644 --- a/src/backend/core/tests/documents/test_api_documents_retrieve.py +++ b/src/backend/core/tests/documents/test_api_documents_retrieve.py @@ -2,16 +2,20 @@ Tests for Documents API endpoint in impress's core app: retrieve """ +import random +from datetime import timedelta + +from django.utils import timezone + import pytest from rest_framework.test import APIClient from core import factories, models -from core.api import serializers pytestmark = pytest.mark.django_db -def test_api_documents_retrieve_anonymous_public(): +def test_api_documents_retrieve_anonymous_public_standalone(): """Anonymous users should be allowed to retrieve public documents.""" document = factories.DocumentFactory(link_reach="public") @@ -26,6 +30,8 @@ def test_api_documents_retrieve_anonymous_public(): "ai_transform": document.link_role == "editor", "ai_translate": document.link_role == "editor", "attachment_upload": document.link_role == "editor", + "children_create": False, + "children_list": True, "collaboration_auth": True, "destroy": False, # Anonymous user can't favorite a document even with read access @@ -33,7 +39,9 @@ def test_api_documents_retrieve_anonymous_public(): "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": document.link_role == "editor", + "restore": False, "retrieve": True, "update": document.link_role == "editor", "versions_destroy": False, @@ -43,12 +51,90 @@ def test_api_documents_retrieve_anonymous_public(): "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": "public", "link_role": document.link_role, "nb_accesses": 0, + "numchild": 0, + "path": document.path, + "title": document.title, + "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + } + + +def test_api_documents_retrieve_anonymous_public_parent(): + """Anonymous users should be allowed to retrieve a document who has a public ancestor.""" + grand_parent = factories.DocumentFactory(link_reach="public") + parent = factories.DocumentFactory( + parent=grand_parent, link_reach=random.choice(["authenticated", "restricted"]) + ) + document = factories.DocumentFactory( + link_reach=random.choice(["authenticated", "restricted"]), parent=parent + ) + + response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 200 + assert response.json() == { + "id": str(document.id), + "abilities": { + "accesses_manage": False, + "accesses_view": False, + "ai_transform": grand_parent.link_role == "editor", + "ai_translate": grand_parent.link_role == "editor", + "attachment_upload": grand_parent.link_role == "editor", + "children_create": False, + "children_list": True, + "collaboration_auth": True, + "destroy": False, + # Anonymous user can't favorite a document even with read access + "favorite": False, + "invite_owner": False, + "link_configuration": False, + "media_auth": True, + "move": False, + "partial_update": grand_parent.link_role == "editor", + "restore": False, + "retrieve": True, + "update": grand_parent.link_role == "editor", + "versions_destroy": False, + "versions_list": False, + "versions_retrieve": False, + }, + "content": document.content, + "created_at": document.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(document.creator.id), + "depth": 3, + "excerpt": document.excerpt, + "is_favorite": False, + "link_reach": document.link_reach, + "link_role": document.link_role, + "nb_accesses": 0, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + } + + +def test_api_documents_retrieve_anonymous_public_child(): + """ + Anonymous users having access to a document should not gain access to a parent document. + """ + document = factories.DocumentFactory( + link_reach=random.choice(["authenticated", "restricted"]) + ) + factories.DocumentFactory(link_reach="public", parent=document) + + response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 401 + assert response.json() == { + "detail": "Authentication credentials were not provided." } @@ -68,8 +154,8 @@ def test_api_documents_retrieve_anonymous_restricted_or_authenticated(reach): @pytest.mark.parametrize("reach", ["public", "authenticated"]) def test_api_documents_retrieve_authenticated_unrelated_public_or_authenticated(reach): """ - Authenticated users should be able to retrieve a public document to which they are - not related. + Authenticated users should be able to retrieve a public/authenticated document to + which they are not related. """ user = factories.UserFactory() @@ -90,13 +176,17 @@ def test_api_documents_retrieve_authenticated_unrelated_public_or_authenticated( "ai_transform": document.link_role == "editor", "ai_translate": document.link_role == "editor", "attachment_upload": document.link_role == "editor", + "children_create": document.link_role == "editor", + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": True, "invite_owner": False, - "media_auth": True, "link_configuration": False, + "media_auth": True, + "move": False, "partial_update": document.link_role == "editor", + "restore": False, "retrieve": True, "update": document.link_role == "editor", "versions_destroy": False, @@ -106,18 +196,104 @@ def test_api_documents_retrieve_authenticated_unrelated_public_or_authenticated( "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": reach, "link_role": document.link_role, "nb_accesses": 0, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], } assert ( models.LinkTrace.objects.filter(document=document, user=user).exists() is True ) +@pytest.mark.parametrize("reach", ["public", "authenticated"]) +def test_api_documents_retrieve_authenticated_public_or_authenticated_parent(reach): + """ + Authenticated users should be allowed to retrieve a document who has a public or + authenticated ancestor. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory(link_reach=reach) + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(link_reach="restricted", parent=parent) + + response = client.get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 200 + assert response.json() == { + "id": str(document.id), + "abilities": { + "accesses_manage": False, + "accesses_view": False, + "ai_transform": grand_parent.link_role == "editor", + "ai_translate": grand_parent.link_role == "editor", + "attachment_upload": grand_parent.link_role == "editor", + "children_create": grand_parent.link_role == "editor", + "children_list": True, + "collaboration_auth": True, + "destroy": False, + "favorite": True, + "invite_owner": False, + "link_configuration": False, + "move": False, + "media_auth": True, + "partial_update": grand_parent.link_role == "editor", + "restore": False, + "retrieve": True, + "update": grand_parent.link_role == "editor", + "versions_destroy": False, + "versions_list": False, + "versions_retrieve": False, + }, + "content": document.content, + "created_at": document.created_at.isoformat().replace("+00:00", "Z"), + "creator": str(document.creator.id), + "depth": 3, + "excerpt": document.excerpt, + "is_favorite": False, + "link_reach": document.link_reach, + "link_role": document.link_role, + "nb_accesses": 0, + "numchild": 0, + "path": document.path, + "title": document.title, + "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [], + } + + +@pytest.mark.parametrize("reach", ["public", "authenticated"]) +def test_api_documents_retrieve_authenticated_public_or_authenticated_child(reach): + """ + Authenticated users having access to a document should not gain access to a parent document. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + factories.DocumentFactory(link_reach=reach, parent=document) + + response = client.get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." + } + + @pytest.mark.parametrize("reach", ["public", "authenticated"]) def test_api_documents_retrieve_authenticated_trace_twice(reach): """ @@ -179,10 +355,8 @@ def test_api_documents_retrieve_authenticated_related_direct(): client.force_login(user) document = factories.DocumentFactory() - factories.UserDocumentAccessFactory(document=document, user=user) - access2 = factories.UserDocumentAccessFactory(document=document) - serializers.UserSerializer(instance=user) - serializers.UserSerializer(instance=access2.user) + access = factories.UserDocumentAccessFactory(document=document, user=user) + factories.UserDocumentAccessFactory(document=document) response = client.get( f"/api/v1.0/documents/{document.id!s}/", @@ -194,12 +368,135 @@ def test_api_documents_retrieve_authenticated_related_direct(): "content": document.content, "creator": str(document.creator.id), "created_at": document.created_at.isoformat().replace("+00:00", "Z"), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": document.link_reach, "link_role": document.link_role, "nb_accesses": 2, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + } + + +def test_api_documents_retrieve_authenticated_related_parent(): + """ + Authenticated users should be allowed to retrieve a document if they are related + to one of its ancestors whatever the role. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory(link_reach="restricted") + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + + access = factories.UserDocumentAccessFactory(document=grand_parent, user=user) + factories.UserDocumentAccessFactory(document=grand_parent) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/", + ) + assert response.status_code == 200 + assert response.json() == { + "id": str(document.id), + "abilities": { + "accesses_manage": access.role in ["administrator", "owner"], + "accesses_view": True, + "ai_transform": access.role != "reader", + "ai_translate": access.role != "reader", + "attachment_upload": access.role != "reader", + "children_create": access.role != "reader", + "children_list": True, + "collaboration_auth": True, + "destroy": access.role == "owner", + "favorite": True, + "invite_owner": access.role == "owner", + "link_configuration": access.role in ["administrator", "owner"], + "media_auth": True, + "move": access.role in ["administrator", "owner"], + "partial_update": access.role != "reader", + "restore": access.role == "owner", + "retrieve": True, + "update": access.role != "reader", + "versions_destroy": access.role in ["administrator", "owner"], + "versions_list": True, + "versions_retrieve": True, + }, + "content": document.content, + "creator": str(document.creator.id), + "created_at": document.created_at.isoformat().replace("+00:00", "Z"), + "depth": 3, + "excerpt": document.excerpt, + "is_favorite": False, + "link_reach": "restricted", + "link_role": document.link_role, + "nb_accesses": 2, + "numchild": 0, + "path": document.path, + "title": document.title, + "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": [access.role], + } + + +def test_api_documents_retrieve_authenticated_related_nb_accesses(): + """Validate computation of number of accesses.""" + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory(link_reach="restricted") + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + + factories.UserDocumentAccessFactory(document=grand_parent, user=user) + factories.UserDocumentAccessFactory(document=parent) + factories.UserDocumentAccessFactory(document=document) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/", + ) + assert response.status_code == 200 + assert response.json()["nb_accesses"] == 3 + + factories.UserDocumentAccessFactory(document=grand_parent) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/", + ) + assert response.status_code == 200 + assert response.json()["nb_accesses"] == 4 + + +def test_api_documents_retrieve_authenticated_related_child(): + """ + Authenticated users should not be allowed to retrieve a document as a result of being + related to one of its children. + """ + user = factories.UserFactory() + + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(link_reach="restricted") + child = factories.DocumentFactory(parent=document) + + factories.UserDocumentAccessFactory(document=child, user=user) + factories.UserDocumentAccessFactory(document=document) + + response = client.get( + f"/api/v1.0/documents/{document.id!s}/", + ) + assert response.status_code == 403 + assert response.json() == { + "detail": "You do not have permission to perform this action." } @@ -238,16 +535,16 @@ def test_api_documents_retrieve_authenticated_related_team_none(mock_user_teams) @pytest.mark.parametrize( - "teams", + "teams,roles", [ - ["readers"], - ["unknown", "readers"], - ["editors"], - ["unknown", "editors"], + [["readers"], ["reader"]], + [["unknown", "readers"], ["reader"]], + [["editors"], ["editor"]], + [["unknown", "editors"], ["editor"]], ], ) def test_api_documents_retrieve_authenticated_related_team_members( - teams, mock_user_teams + teams, roles, mock_user_teams ): """ Authenticated users should be allowed to retrieve a document to which they @@ -285,25 +582,30 @@ def test_api_documents_retrieve_authenticated_related_team_members( "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": "restricted", "link_role": document.link_role, "nb_accesses": 5, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": roles, } @pytest.mark.parametrize( - "teams", + "teams,roles", [ - ["administrators"], - ["editors", "administrators"], - ["unknown", "administrators"], + [["administrators"], ["administrator"]], + [["editors", "administrators"], ["administrator", "editor"]], + [["unknown", "administrators"], ["administrator"]], ], ) def test_api_documents_retrieve_authenticated_related_team_administrators( - teams, mock_user_teams + teams, roles, mock_user_teams ): """ Authenticated users should be allowed to retrieve a document to which they @@ -341,26 +643,31 @@ def test_api_documents_retrieve_authenticated_related_team_administrators( "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": "restricted", "link_role": document.link_role, "nb_accesses": 5, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": roles, } @pytest.mark.parametrize( - "teams", + "teams,roles", [ - ["owners"], - ["owners", "administrators"], - ["members", "administrators", "owners"], - ["unknown", "owners"], + [["owners"], ["owner"]], + [["owners", "administrators"], ["owner", "administrator"]], + [["members", "administrators", "owners"], ["owner", "administrator"]], + [["unknown", "owners"], ["owner"]], ], ) def test_api_documents_retrieve_authenticated_related_team_owners( - teams, mock_user_teams + teams, roles, mock_user_teams ): """ Authenticated users should be allowed to retrieve a restricted document to which @@ -369,7 +676,6 @@ def test_api_documents_retrieve_authenticated_related_team_owners( mock_user_teams.return_value = teams user = factories.UserFactory() - client = APIClient() client.force_login(user) @@ -398,10 +704,194 @@ def test_api_documents_retrieve_authenticated_related_team_owners( "content": document.content, "created_at": document.created_at.isoformat().replace("+00:00", "Z"), "creator": str(document.creator.id), + "depth": 1, + "excerpt": document.excerpt, "is_favorite": False, "link_reach": "restricted", "link_role": document.link_role, "nb_accesses": 5, + "numchild": 0, + "path": document.path, "title": document.title, "updated_at": document.updated_at.isoformat().replace("+00:00", "Z"), + "user_roles": roles, } + + +def test_api_documents_retrieve_user_roles(django_assert_num_queries): + """ + Roles should be annotated on querysets taking into account all documents ancestors. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + grand_parent = factories.DocumentFactory( + users=factories.UserFactory.create_batch(2) + ) + parent = factories.DocumentFactory( + parent=grand_parent, users=factories.UserFactory.create_batch(2) + ) + document = factories.DocumentFactory( + parent=parent, users=factories.UserFactory.create_batch(2) + ) + + accesses = ( + factories.UserDocumentAccessFactory(document=grand_parent, user=user), + factories.UserDocumentAccessFactory(document=parent, user=user), + factories.UserDocumentAccessFactory(document=document, user=user), + ) + expected_roles = {access.role for access in accesses} + + with django_assert_num_queries(8): + response = client.get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 200 + + user_roles = response.json()["user_roles"] + assert set(user_roles) == expected_roles + + +def test_api_documents_retrieve_numqueries_with_link_trace(django_assert_num_queries): + """If the link traced already exists, the number of queries should be minimal.""" + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + document = factories.DocumentFactory(users=[user], link_traces=[user]) + + with django_assert_num_queries(2): + response = client.get(f"/api/v1.0/documents/{document.id!s}/") + + assert response.status_code == 200 + + assert response.json()["id"] == str(document.id) + + +# Soft/hard delete + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("reach", models.LinkReachChoices.values) +def test_api_documents_retrieve_soft_deleted_anonymous(reach, depth): + """ + A soft/hard deleted public document should not be accessible via its + detail endpoint for anonymous users, and should return a 404. + """ + documents = [] + for i in range(depth): + documents.append( + factories.DocumentFactory(link_reach=reach) + if i == 0 + else factories.DocumentFactory(parent=documents[-1]) + ) + assert models.Document.objects.count() == depth + + response = APIClient().get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 200 if reach == "public" else 401 + + # Delete any one of the documents... + deleted_document = random.choice(documents) + deleted_document.deleted_at = timezone.now() - timedelta(days=15) + deleted_document.save() + + response = APIClient().get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + + deleted_document.deleted_at = timezone.now() - timedelta(days=40) + deleted_document.save() + + response = APIClient().get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("reach", models.LinkReachChoices.values) +def test_api_documents_retrieve_soft_deleted_authenticated(reach, depth): + """ + A soft/hard deleted document should not be accessible via its detail endpoint for + authenticated users not related to the document. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + documents = [] + for i in range(depth): + documents.append( + factories.DocumentFactory(link_reach=reach) + if i == 0 + else factories.DocumentFactory(parent=documents[-1]) + ) + assert models.Document.objects.count() == depth + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 200 if reach in ["public", "authenticated"] else 403 + + # Delete any one of the documents... + deleted_document = random.choice(documents) + deleted_document.deleted_at = timezone.now() - timedelta(days=15) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + + deleted_document.deleted_at = timezone.now() - timedelta(days=40) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} + + +@pytest.mark.parametrize("depth", [1, 2, 3]) +@pytest.mark.parametrize("role", models.RoleChoices.values) +def test_api_documents_retrieve_soft_deleted_related(role, depth): + """ + A soft deleted document should be accessible via its detail endpoint but only + for owners of the document.Hard deleted documents are not accessible any more. + """ + user = factories.UserFactory() + client = APIClient() + client.force_login(user) + + documents = [] + for i in range(depth): + documents.append( + factories.UserDocumentAccessFactory(role=role, user=user).document + if i == 0 + else factories.DocumentFactory(parent=documents[-1]) + ) + assert models.Document.objects.count() == depth + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 200 + + # Delete any one of the documents + deleted_document = random.choice(documents) + deleted_document.deleted_at = timezone.now() - timedelta(days=15) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + # Only the owner of the document (not the owner of one of its ancestors) can + # see a deleted document (only he could have deleted it...) + assert response.status_code == 200 if role == "owner" and depth == 1 else 404 + + deleted_document.deleted_at = timezone.now() - timedelta(days=40) + deleted_document.save() + + response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/") + + assert response.status_code == 404 + assert response.json() == {"detail": "No Document matches the given query."} diff --git a/src/backend/core/tests/documents/test_api_documents_update.py b/src/backend/core/tests/documents/test_api_documents_update.py index 3724af388..8eaba224f 100644 --- a/src/backend/core/tests/documents/test_api_documents_update.py +++ b/src/backend/core/tests/documents/test_api_documents_update.py @@ -16,6 +16,7 @@ pytestmark = pytest.mark.django_db +@pytest.mark.parametrize("via_parent", [True, False]) @pytest.mark.parametrize( "reach, role", [ @@ -26,12 +27,18 @@ ("public", "reader"), ], ) -def test_api_documents_update_anonymous_forbidden(reach, role): +def test_api_documents_update_anonymous_forbidden(reach, role, via_parent): """ Anonymous users should not be allowed to update a document when link configuration does not allow it. """ - document = factories.DocumentFactory(link_reach=reach, link_role=role) + if via_parent: + grand_parent = factories.DocumentFactory(link_reach=reach, link_role=role) + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + else: + document = factories.DocumentFactory(link_reach=reach, link_role=role) + old_document_values = serializers.DocumentSerializer(instance=document).data new_document_values = serializers.DocumentSerializer( @@ -52,6 +59,7 @@ def test_api_documents_update_anonymous_forbidden(reach, role): assert document_values == old_document_values +@pytest.mark.parametrize("via_parent", [True, False]) @pytest.mark.parametrize( "reach,role", [ @@ -61,7 +69,9 @@ def test_api_documents_update_anonymous_forbidden(reach, role): ("restricted", "editor"), ], ) -def test_api_documents_update_authenticated_unrelated_forbidden(reach, role): +def test_api_documents_update_authenticated_unrelated_forbidden( + reach, role, via_parent +): """ Authenticated users should not be allowed to update a document to which they are not related if the link configuration does not allow it. @@ -71,7 +81,12 @@ def test_api_documents_update_authenticated_unrelated_forbidden(reach, role): client = APIClient() client.force_login(user) - document = factories.DocumentFactory(link_reach=reach, link_role=role) + if via_parent: + grand_parent = factories.DocumentFactory(link_reach=reach, link_role=role) + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + else: + document = factories.DocumentFactory(link_reach=reach, link_role=role) old_document_values = serializers.DocumentSerializer(instance=document).data new_document_values = serializers.DocumentSerializer( @@ -93,6 +108,7 @@ def test_api_documents_update_authenticated_unrelated_forbidden(reach, role): assert document_values == old_document_values +@pytest.mark.parametrize("via_parent", [True, False]) @pytest.mark.parametrize( "is_authenticated,reach,role", [ @@ -102,10 +118,10 @@ def test_api_documents_update_authenticated_unrelated_forbidden(reach, role): ], ) def test_api_documents_update_anonymous_or_authenticated_unrelated( - is_authenticated, reach, role + is_authenticated, reach, role, via_parent ): """ - Authenticated users should be able to update a document to which + Anonymous and authenticated users should be able to update a document to which they are not related if the link configuration allows it. """ client = APIClient() @@ -116,7 +132,12 @@ def test_api_documents_update_anonymous_or_authenticated_unrelated( else: user = AnonymousUser() - document = factories.DocumentFactory(link_reach=reach, link_role=role) + if via_parent: + grand_parent = factories.DocumentFactory(link_reach=reach, link_role=role) + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + else: + document = factories.DocumentFactory(link_reach=reach, link_role=role) old_document_values = serializers.DocumentSerializer(instance=document).data new_document_values = serializers.DocumentSerializer( @@ -137,8 +158,11 @@ def test_api_documents_update_anonymous_or_authenticated_unrelated( "accesses", "created_at", "creator", + "depth", "link_reach", "link_role", + "numchild", + "path", ]: assert value == old_document_values[key] elif key == "updated_at": @@ -147,24 +171,34 @@ def test_api_documents_update_anonymous_or_authenticated_unrelated( assert value == new_document_values[key] +@pytest.mark.parametrize("via_parent", [True, False]) @pytest.mark.parametrize("via", VIA) -def test_api_documents_update_authenticated_reader(via, mock_user_teams): +def test_api_documents_update_authenticated_reader(via, via_parent, mock_user_teams): """ - Users who are reader of a document but not administrators should - not be allowed to update it. + Users who are reader of a document should not be allowed to update it. """ user = factories.UserFactory(with_owned_document=True) client = APIClient() client.force_login(user) - document = factories.DocumentFactory(link_role="reader") + if via_parent: + grand_parent = factories.DocumentFactory(link_reach="restricted") + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + access_document = grand_parent + else: + document = factories.DocumentFactory(link_reach="restricted") + access_document = document + if via == USER: - factories.UserDocumentAccessFactory(document=document, user=user, role="reader") + factories.UserDocumentAccessFactory( + document=access_document, user=user, role="reader" + ) elif via == TEAM: mock_user_teams.return_value = ["lasuite", "unknown"] factories.TeamDocumentAccessFactory( - document=document, team="lasuite", role="reader" + document=access_document, team="lasuite", role="reader" ) old_document_values = serializers.DocumentSerializer(instance=document).data @@ -188,10 +222,11 @@ def test_api_documents_update_authenticated_reader(via, mock_user_teams): assert document_values == old_document_values +@pytest.mark.parametrize("via_parent", [True, False]) @pytest.mark.parametrize("role", ["editor", "administrator", "owner"]) @pytest.mark.parametrize("via", VIA) def test_api_documents_update_authenticated_editor_administrator_or_owner( - via, role, mock_user_teams + via, role, via_parent, mock_user_teams ): """A user who is editor, administrator or owner of a document should be allowed to update it.""" user = factories.UserFactory(with_owned_document=True) @@ -199,13 +234,23 @@ def test_api_documents_update_authenticated_editor_administrator_or_owner( client = APIClient() client.force_login(user) - document = factories.DocumentFactory() + if via_parent: + grand_parent = factories.DocumentFactory(link_reach="restricted") + parent = factories.DocumentFactory(parent=grand_parent, link_reach="restricted") + document = factories.DocumentFactory(parent=parent, link_reach="restricted") + access_document = grand_parent + else: + document = factories.DocumentFactory(link_reach="restricted") + access_document = document + if via == USER: - factories.UserDocumentAccessFactory(document=document, user=user, role=role) + factories.UserDocumentAccessFactory( + document=access_document, user=user, role=role + ) elif via == TEAM: mock_user_teams.return_value = ["lasuite", "unknown"] factories.TeamDocumentAccessFactory( - document=document, team="lasuite", role=role + document=access_document, team="lasuite", role=role ) old_document_values = serializers.DocumentSerializer(instance=document).data @@ -227,55 +272,12 @@ def test_api_documents_update_authenticated_editor_administrator_or_owner( "id", "created_at", "creator", + "depth", "link_reach", "link_role", "nb_accesses", - ]: - assert value == old_document_values[key] - elif key == "updated_at": - assert value > old_document_values[key] - else: - assert value == new_document_values[key] - - -@pytest.mark.parametrize("via", VIA) -def test_api_documents_update_authenticated_owners(via, mock_user_teams): - """Administrators of a document should be allowed to update it.""" - user = factories.UserFactory(with_owned_document=True) - - client = APIClient() - client.force_login(user) - - document = factories.DocumentFactory() - if via == USER: - factories.UserDocumentAccessFactory(document=document, user=user, role="owner") - elif via == TEAM: - mock_user_teams.return_value = ["lasuite", "unknown"] - factories.TeamDocumentAccessFactory( - document=document, team="lasuite", role="owner" - ) - - old_document_values = serializers.DocumentSerializer(instance=document).data - - new_document_values = serializers.DocumentSerializer( - instance=factories.DocumentFactory() - ).data - - response = client.put( - f"/api/v1.0/documents/{document.id!s}/", new_document_values, format="json" - ) - - assert response.status_code == 200 - document = models.Document.objects.get(pk=document.pk) - document_values = serializers.DocumentSerializer(instance=document).data - for key, value in document_values.items(): - if key in [ - "id", - "created_at", - "creator", - "link_reach", - "link_role", - "nb_accesses", + "numchild", + "path", ]: assert value == old_document_values[key] elif key == "updated_at": diff --git a/src/backend/core/tests/test_models_documents.py b/src/backend/core/tests/test_models_documents.py index 8ddaa0182..d19417b86 100644 --- a/src/backend/core/tests/test_models_documents.py +++ b/src/backend/core/tests/test_models_documents.py @@ -34,20 +34,18 @@ def test_models_documents_id_unique(): def test_models_documents_creator_required(): """No field should be required on the Document model.""" - models.Document.objects.create() + models.Document.add_root() def test_models_documents_title_null(): """The "title" field can be null.""" - document = models.Document.objects.create( - title=None, creator=factories.UserFactory() - ) + document = models.Document.add_root(title=None, creator=factories.UserFactory()) assert document.title is None def test_models_documents_title_empty(): """The "title" field can be empty.""" - document = models.Document.objects.create(title="", creator=factories.UserFactory()) + document = models.Document.add_root(title="", creator=factories.UserFactory()) assert document.title == "" @@ -95,13 +93,17 @@ def test_models_documents_get_abilities_forbidden(is_authenticated, reach, role) "ai_transform": False, "ai_translate": False, "attachment_upload": False, + "children_create": False, + "children_list": False, "collaboration_auth": False, "destroy": False, "favorite": False, "invite_owner": False, "media_auth": False, + "move": False, "link_configuration": False, "partial_update": False, + "restore": False, "retrieve": False, "update": False, "versions_destroy": False, @@ -132,13 +134,17 @@ def test_models_documents_get_abilities_reader(is_authenticated, reach): "ai_transform": False, "ai_translate": False, "attachment_upload": False, + "children_create": False, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": is_authenticated, "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": False, + "restore": False, "retrieve": True, "update": False, "versions_destroy": False, @@ -169,13 +175,17 @@ def test_models_documents_get_abilities_editor(is_authenticated, reach): "ai_transform": True, "ai_translate": True, "attachment_upload": True, + "children_create": is_authenticated, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": is_authenticated, "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": True, + "restore": False, "retrieve": True, "update": True, "versions_destroy": False, @@ -195,13 +205,17 @@ def test_models_documents_get_abilities_owner(): "ai_transform": True, "ai_translate": True, "attachment_upload": True, + "children_create": True, + "children_list": True, "collaboration_auth": True, "destroy": True, "favorite": True, "invite_owner": True, "link_configuration": True, "media_auth": True, + "move": True, "partial_update": True, + "restore": True, "retrieve": True, "update": True, "versions_destroy": True, @@ -220,13 +234,17 @@ def test_models_documents_get_abilities_administrator(): "ai_transform": True, "ai_translate": True, "attachment_upload": True, + "children_create": True, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": True, "invite_owner": False, "link_configuration": True, "media_auth": True, + "move": True, "partial_update": True, + "restore": False, "retrieve": True, "update": True, "versions_destroy": True, @@ -248,13 +266,17 @@ def test_models_documents_get_abilities_editor_user(django_assert_num_queries): "ai_transform": True, "ai_translate": True, "attachment_upload": True, + "children_create": True, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": True, "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": True, + "restore": False, "retrieve": True, "update": True, "versions_destroy": False, @@ -278,13 +300,17 @@ def test_models_documents_get_abilities_reader_user(django_assert_num_queries): "ai_transform": False, "ai_translate": False, "attachment_upload": False, + "children_create": False, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": True, "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": False, + "restore": False, "retrieve": True, "update": False, "versions_destroy": False, @@ -309,13 +335,17 @@ def test_models_documents_get_abilities_preset_role(django_assert_num_queries): "ai_transform": False, "ai_translate": False, "attachment_upload": False, + "children_create": False, + "children_list": True, "collaboration_auth": True, "destroy": False, "favorite": True, "invite_owner": False, "link_configuration": False, "media_auth": True, + "move": False, "partial_update": False, + "restore": False, "retrieve": True, "update": False, "versions_destroy": False, diff --git a/src/backend/demo/management/commands/create_demo.py b/src/backend/demo/management/commands/create_demo.py index 4ac9efc71..d07df6133 100644 --- a/src/backend/demo/management/commands/create_demo.py +++ b/src/backend/demo/management/commands/create_demo.py @@ -135,9 +135,14 @@ def create_demo(stdout): users_ids = list(models.User.objects.values_list("id", flat=True)) with Timeit(stdout, "Creating documents"): - for _ in range(defaults.NB_OBJECTS["docs"]): + for i in range(defaults.NB_OBJECTS["docs"]): + # pylint: disable=protected-access + key = models.Document._int2str(i) # noqa: SLF001 + padding = models.Document.alphabet[0] * (models.Document.steplen - len(key)) queue.push( models.Document( + depth=1, + path=f"{padding}{key}", creator_id=random.choice(users_ids), title=fake.sentence(nb_words=4), link_reach=models.LinkReachChoices.AUTHENTICATED diff --git a/src/backend/impress/settings.py b/src/backend/impress/settings.py index f29dce1ac..0818346e2 100755 --- a/src/backend/impress/settings.py +++ b/src/backend/impress/settings.py @@ -299,6 +299,7 @@ class Base(Configuration): "dockerflow.django", "rest_framework", "parler", + "treebeard", "easy_thumbnails", # Django "django.contrib.admin", @@ -350,6 +351,10 @@ class Base(Configuration): "REDOC_DIST": "SIDECAR", } + SOFT_DELETE_KEEP_DAYS = values.Value( + 30, environ_name="SOFT_DELETE_KEEP_DAYS", environ_prefix=None + ) + # Mail EMAIL_BACKEND = values.Value("django.core.mail.backends.smtp.EmailBackend") EMAIL_BRAND_NAME = values.Value(None) diff --git a/src/backend/pyproject.toml b/src/backend/pyproject.toml index dbd54a740..5678b3d91 100644 --- a/src/backend/pyproject.toml +++ b/src/backend/pyproject.toml @@ -38,6 +38,7 @@ dependencies = [ "django-storages[s3]==1.14.4", "django-timezone-field>=5.1", "django==5.1.4", + "django-treebeard==4.7.1", "djangorestframework==3.15.2", "drf_spectacular==0.28.0", "dockerflow==2024.4.2",