studio/docs
Archived
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨(backend) add soft delete to documents and refactor db queryset

Browse Source 
Now that we have introduced a document tree structure, it is not
possible to allow deleting documents anymore as it impacts the whole
subtree below the deleted document and the consequences are too big.

We introduce soft delete in order to give a second thought to the
document's owner (who is the only one to be allowed to delete a
document). After a document is soft deleted, the owner can still
see it in the trashbin (/api/v1.0/documents/trashbin).
After a grace period (30 days be default) the document disappears
from the trashbin and can't be restored anymore. Note that even
then it is still kept in database. Cleaning the database to erase
deleted documents after the grace period can be done as a maintenance
script.
This commit is contained in:
Samuel Paccoud - DINUM
2025-01-02 17:20:09 +01:00
committed by Anthony LC
parent 4de03d292a
commit 8ccfdb3c6a
21 changed files with 1373 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/backend/core/tests/documents/test_api_document_invitations.py
View File

@@ -624,7 +624,9 @@ def test_api_document_invitations_create_cannot_invite_existing_users():
)
assert response.status_code == 400
assert response.json() == ["This email is already associated to a registered user."]
assert response.json() == {
"email": ["This email is already associated to a registered user."]
}
# Update

8
src/backend/core/tests/documents/test_api_document_versions.py
View File

@@ -604,15 +604,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)

5
src/backend/core/tests/documents/test_api_documents_children_create.py
View File

@@ -31,8 +31,11 @@ def test_api_documents_children_create_anonymous(reach, role, depth):
},
)
assert response.status_code == 401
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])

7
src/backend/core/tests/documents/test_api_documents_children_list.py
View File

@@ -305,7 +305,7 @@ def test_api_documents_children_list_authenticated_related_direct():
client.force_login(user)
document = factories.DocumentFactory()
factories.UserDocumentAccessFactory(document=document, user=user)
access = factories.UserDocumentAccessFactory(document=document, user=user)
factories.UserDocumentAccessFactory(document=document)
child1, child2 = factories.DocumentFactory.create_batch(2, parent=document)
@@ -378,7 +378,6 @@ def test_api_documents_children_list_authenticated_related_parent():
grand_parent_access = factories.UserDocumentAccessFactory(
document=grand_parent, user=user
)
factories.UserDocumentAccessFactory(document=grand_parent, user=user)
response = client.get(
f"/api/v1.0/documents/{document.id!s}/children/",
@@ -400,7 +399,7 @@ def test_api_documents_children_list_authenticated_related_parent():
"link_reach": child1.link_reach,
"link_role": child1.link_role,
"numchild": 0,
"nb_accesses": 3,
"nb_accesses": 2,
"path": child1.path,
"title": child1.title,
"updated_at": child1.updated_at.isoformat().replace("+00:00", "Z"),
@@ -417,7 +416,7 @@ def test_api_documents_children_list_authenticated_related_parent():
"link_reach": child2.link_reach,
"link_role": child2.link_role,
"numchild": 0,
"nb_accesses": 2,
"nb_accesses": 1,
"path": child2.path,
"title": child2.title,
"updated_at": child2.updated_at.isoformat().replace("+00:00", "Z"),

37
src/backend/core/tests/documents/test_api_documents_delete.py
View File

@@ -77,6 +77,37 @@ 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
# Make sure it is only a soft delete
assert models.Document.objects.count() == depth
assert models.Document.objects.filter(deleted_at__isnull=True).count() == depth - 1
assert models.Document.objects.filter(deleted_at__isnull=False).count() == 1
@pytest.mark.parametrize("via", VIA)
def test_api_documents_delete_authenticated_owner(via, mock_user_teams):
"""
@@ -101,4 +132,8 @@ def test_api_documents_delete_authenticated_owner(via, mock_user_teams):
)
assert response.status_code == 204
assert models.Document.objects.exists() is False
# Make sure it is only a soft delete
assert models.Document.objects.count() == 1
assert models.Document.objects.filter(deleted_at__isnull=True).exists() is False
assert models.Document.objects.filter(deleted_at__isnull=False).count() == 1

79
src/backend/core/tests/documents/test_api_documents_list.py
View File

@@ -3,8 +3,11 @@ Tests for Documents API endpoint in impress's core app: list
"""
import random
from datetime import timedelta
from unittest import mock
from django.utils import timezone
import pytest
from faker import Faker
from rest_framework.pagination import PageNumberPagination
@@ -21,7 +24,7 @@ pytestmark = pytest.mark.django_db
def test_api_documents_list_anonymous(reach, role):
"""
Anonymous users should not be allowed to list documents whatever the
link reach and the role
link reach and link role
"""
factories.DocumentFactory(link_reach=reach, link_role=role)
@@ -76,6 +79,7 @@ def test_api_documents_list_format():
}
# pylint: disable=too-many-locals
def test_api_documents_list_authenticated_direct(django_assert_num_queries):
"""
Authenticated users should be able to list documents they are a direct
@@ -110,17 +114,50 @@ def test_api_documents_list_authenticated_direct(django_assert_num_queries):
hidden_root = factories.DocumentFactory()
child3_with_access = factories.DocumentFactory(parent=hidden_root)
factories.UserDocumentAccessFactory(user=user, document=child3_with_access)
child4_with_access = factories.DocumentFactory(parent=hidden_root)
factories.UserDocumentAccessFactory(user=user, document=child4_with_access)
expected_ids = {str(document1.id), str(document2.id), str(child3_with_access.id)}
# Documents that are soft deleted and children of a soft deleted document should not be listed
soft_deleted_document = factories.DocumentFactory(users=[user])
child_of_soft_deleted_document = factories.DocumentFactory(
users=[user],
parent=soft_deleted_document,
)
factories.DocumentFactory(users=[user], parent=child_of_soft_deleted_document)
soft_deleted_document.soft_delete()
with django_assert_num_queries(7):
# Documents that are permanently deleted and children of a permanently deleted
# document should not be listed
permanently_deleted_document = factories.DocumentFactory(users=[user])
child_of_permanently_deleted_document = factories.DocumentFactory(
users=[user], parent=permanently_deleted_document
)
factories.DocumentFactory(
users=[user], parent=child_of_permanently_deleted_document
)
fourty_days_ago = timezone.now() - timedelta(days=40)
with mock.patch("django.utils.timezone.now", return_value=fourty_days_ago):
permanently_deleted_document.soft_delete()
expected_ids = {
str(document1.id),
str(document2.id),
str(child3_with_access.id),
str(child4_with_access.id),
}
with django_assert_num_queries(8):
response = client.get("/api/v1.0/documents/")
# nb_accesses should now be cached
with django_assert_num_queries(4):
response = client.get("/api/v1.0/documents/")
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == 3
results_id = {result["id"] for result in results}
assert expected_ids == results_id
results_ids = {result["id"] for result in results}
assert expected_ids == results_ids
def test_api_documents_list_authenticated_via_team(
@@ -148,7 +185,11 @@ def test_api_documents_list_authenticated_via_team(
expected_ids = {str(document.id) for document in documents_team1 + documents_team2}
with django_assert_num_queries(8):
with django_assert_num_queries(9):
response = client.get("/api/v1.0/documents/")
# nb_accesses should now be cached
with django_assert_num_queries(4):
response = client.get("/api/v1.0/documents/")
assert response.status_code == 200
@@ -177,10 +218,12 @@ def test_api_documents_list_authenticated_link_reach_restricted(
other_document = factories.DocumentFactory(link_reach="public")
models.LinkTrace.objects.create(document=other_document, user=user)
with django_assert_num_queries(5):
response = client.get("/api/v1.0/documents/")
# nb_accesses should now be cached
with django_assert_num_queries(4):
response = client.get(
"/api/v1.0/documents/",
)
response = client.get("/api/v1.0/documents/")
assert response.status_code == 200
results = response.json()["results"]
@@ -225,9 +268,11 @@ def test_api_documents_list_authenticated_link_reach_public_or_authenticated(
expected_ids = {str(document1.id), str(document2.id), str(visible_child.id)}
with django_assert_num_queries(7):
response = client.get(
"/api/v1.0/documents/",
)
response = client.get("/api/v1.0/documents/")
# nb_accesses should now be cached
with django_assert_num_queries(4):
response = client.get("/api/v1.0/documents/")
assert response.status_code == 200
results = response.json()["results"]
@@ -317,7 +362,11 @@ def test_api_documents_list_favorites_no_extra_queries(django_assert_num_queries
factories.DocumentFactory.create_batch(2, users=[user])
url = "/api/v1.0/documents/"
with django_assert_num_queries(8):
with django_assert_num_queries(9):
response = client.get(url)
# nb_accesses should now be cached
with django_assert_num_queries(4):
response = client.get(url)
assert response.status_code == 200
@@ -330,7 +379,7 @@ def test_api_documents_list_favorites_no_extra_queries(django_assert_num_queries
for document in special_documents:
models.DocumentFavorite.objects.create(document=document, user=user)
with django_assert_num_queries(8):
with django_assert_num_queries(4):
response = client.get(url)
assert response.status_code == 200

72
src/backend/core/tests/documents/test_api_documents_list_filters.py
View File

@@ -86,8 +86,6 @@ def test_api_documents_list_filter_and_access_rights():
"-created_at",
"is_favorite",
"-is_favorite",
"nb_accesses",
"-nb_accesses",
"title",
"-title",
"updated_at",
@@ -143,8 +141,6 @@ def test_api_documents_list_ordering_by_fields():
"-created_at",
"is_favorite",
"-is_favorite",
"nb_accesses",
"-nb_accesses",
"title",
"-title",
"updated_at",
@@ -165,6 +161,31 @@ def test_api_documents_list_ordering_by_fields():
assert compare(results[i][field], results[i + 1][field])
# Filters: unknown field
def test_api_documents_list_filter_unknown_field():
"""
Trying to filter by an unknown field should raise a 400 error.
"""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
factories.DocumentFactory()
expected_ids = {
str(document.id)
for document in factories.DocumentFactory.create_batch(2, users=[user])
}
response = client.get("/api/v1.0/documents/?unknown=true")
assert response.status_code == 200
results = response.json()["results"]
assert len(results) == 2
assert {result["id"] for result in results} == expected_ids
# Filters: is_creator_me
@@ -291,46 +312,6 @@ def test_api_documents_list_filter_is_favorite_invalid():
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
@@ -360,7 +341,8 @@ def test_api_documents_list_filter_title(query, nb_results):
"Annual Review 2024",
]
for title in titles:
factories.DocumentFactory(title=title, users=[user])
parent = factories.DocumentFactory() if random.choice([True, False]) else None
factories.DocumentFactory(title=title, users=[user], parent=parent)
# Perform the search query
response = client.get(f"/api/v1.0/documents/?title={query:s}")

174
src/backend/core/tests/documents/test_api_documents_retrieve.py
View File

@@ -3,6 +3,10 @@ Tests for Documents API endpoint in impress's core app: retrieve
"""
import random
from datetime import timedelta
from unittest import mock
from django.utils import timezone
import pytest
from rest_framework.test import APIClient
@@ -735,7 +739,7 @@ def test_api_documents_retrieve_user_roles(django_assert_num_queries):
)
expected_roles = {access.role for access in accesses}
with django_assert_num_queries(8):
with django_assert_num_queries(10):
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
assert response.status_code == 200
@@ -752,9 +756,175 @@ def test_api_documents_retrieve_numqueries_with_link_trace(django_assert_num_que
document = factories.DocumentFactory(users=[user], link_traces=[user])
with django_assert_num_queries(2):
with django_assert_num_queries(4):
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
with django_assert_num_queries(3):
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
assert response.status_code == 200
assert response.json()["id"] == str(document.id)
# Soft/permanent 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/permanently 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.soft_delete()
response = APIClient().get(f"/api/v1.0/documents/{documents[-1].id!s}/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
fourty_days_ago = timezone.now() - timedelta(days=40)
deleted_document.deleted_at = fourty_days_ago
deleted_document.ancestors_deleted_at = fourty_days_ago
deleted_document.save()
response = APIClient().get(f"/api/v1.0/documents/{documents[-1].id!s}/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
@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/permanently 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.soft_delete()
response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
fourty_days_ago = timezone.now() - timedelta(days=40)
deleted_document.deleted_at = fourty_days_ago
deleted_document.ancestors_deleted_at = fourty_days_ago
deleted_document.save()
response = client.get(f"/api/v1.0/documents/{documents[-1].id!s}/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
@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 only be accessible via its detail endpoint by
users with specific "owner" access rights.
"""
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
document = documents[-1]
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
assert response.status_code == 200
# Delete any one of the documents
deleted_document = random.choice(documents)
deleted_document.soft_delete()
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
if role == "owner":
assert response.status_code == 200
assert response.json()["id"] == str(document.id)
else:
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}
@pytest.mark.parametrize("depth", [1, 2, 3])
@pytest.mark.parametrize("role", models.RoleChoices.values)
def test_api_documents_retrieve_permanently_deleted_related(role, depth):
"""
A permanently deleted document should not be accessible via its detail endpoint for
authenticated users with specific access rights whatever their role.
"""
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
document = documents[-1]
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
assert response.status_code == 200
# Delete any one of the documents
deleted_document = random.choice(documents)
fourty_days_ago = timezone.now() - timedelta(days=40)
with mock.patch("django.utils.timezone.now", return_value=fourty_days_ago):
deleted_document.soft_delete()
response = client.get(f"/api/v1.0/documents/{document.id!s}/")
assert response.status_code == 404
assert response.json() == {"detail": "Not found."}

275
src/backend/core/tests/documents/test_api_documents_trashbin.py Normal file
View File

@@ -0,0 +1,275 @@
"""
Tests for Documents API endpoint in impress's core app: list
"""
from datetime import timedelta
from unittest import mock
from django.utils import timezone
import pytest
from faker import Faker
from rest_framework.pagination import PageNumberPagination
from rest_framework.test import APIClient
from core import factories, models
fake = Faker()
pytestmark = pytest.mark.django_db
@pytest.mark.parametrize("role", models.LinkRoleChoices.values)
@pytest.mark.parametrize("reach", models.LinkReachChoices.values)
def test_api_documents_trashbin_anonymous(reach, role):
"""
Anonymous users should not be allowed to list documents from the trashbin
whatever the link reach and link role
"""
factories.DocumentFactory(
link_reach=reach, link_role=role, deleted_at=timezone.now()
)
response = APIClient().get("/api/v1.0/documents/trashbin/")
assert response.status_code == 200
assert response.json() == {
"count": 0,
"next": None,
"previous": None,
"results": [],
}
def test_api_documents_trashbin_format():
"""Validate the format of documents as returned by the trashbin view."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
other_users = factories.UserFactory.create_batch(3)
document = factories.DocumentFactory(
deleted_at=timezone.now(),
users=factories.UserFactory.create_batch(2),
favorited_by=[user, *other_users],
link_traces=other_users,
)
factories.UserDocumentAccessFactory(document=document, user=user, role="owner")
response = client.get("/api/v1.0/documents/trashbin/")
assert response.status_code == 200
content = response.json()
results = content.pop("results")
assert content == {
"count": 1,
"next": None,
"previous": None,
}
assert len(results) == 1
assert results[0] == {
"id": str(document.id),
"abilities": {
"accesses_manage": True,
"accesses_view": True,
"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": False, # Can't move a deleted document
"partial_update": True,
"retrieve": True,
"update": True,
"versions_destroy": True,
"versions_list": True,
"versions_retrieve": True,
},
"created_at": document.created_at.isoformat().replace("+00:00", "Z"),
"creator": str(document.creator.id),
"depth": 1,
"excerpt": document.excerpt,
"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": ["owner"],
}
def test_api_documents_trashbin_authenticated_direct(django_assert_num_queries):
"""
The trashbin should only list deleted documents for which the current user is owner.
"""
now = timezone.now()
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document1, document2 = factories.DocumentFactory.create_batch(2, deleted_at=now)
models.DocumentAccess.objects.create(document=document1, user=user, role="owner")
models.DocumentAccess.objects.create(document=document2, user=user, role="owner")
# Unrelated documents
for reach in models.LinkReachChoices:
for role in models.LinkRoleChoices:
factories.DocumentFactory(link_reach=reach, link_role=role, deleted_at=now)
# Role other than "owner"
for role in models.RoleChoices.values:
if role == "owner":
continue
document_not_owner = factories.DocumentFactory(deleted_at=now)
models.DocumentAccess.objects.create(
document=document_not_owner, user=user, role=role
)
# Nested documents should also get listed
parent = factories.DocumentFactory(parent=document1)
document3 = factories.DocumentFactory(parent=parent, deleted_at=now)
models.DocumentAccess.objects.create(document=parent, user=user, role="owner")
# Permanently deleted documents should not be listed
fourty_days_ago = timezone.now() - timedelta(days=40)
permanently_deleted_document = factories.DocumentFactory(users=[(user, "owner")])
with mock.patch("django.utils.timezone.now", return_value=fourty_days_ago):
permanently_deleted_document.soft_delete()
expected_ids = {str(document1.id), str(document2.id), str(document3.id)}
with django_assert_num_queries(7):
response = client.get("/api/v1.0/documents/trashbin/")
with django_assert_num_queries(4):
response = client.get("/api/v1.0/documents/trashbin/")
assert response.status_code == 200
results = response.json()["results"]
results_ids = {result["id"] for result in results}
assert len(results) == 3
assert expected_ids == results_ids
def test_api_documents_trashbin_authenticated_via_team(
django_assert_num_queries, mock_user_teams
):
"""
Authenticated users should be able to list trashbin documents they own via a team.
"""
now = timezone.now()
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
mock_user_teams.return_value = ["team1", "team2", "unknown"]
deleted_document_team1 = factories.DocumentFactory(
teams=[("team1", "owner")], deleted_at=now
)
factories.DocumentFactory(teams=[("team1", "owner")])
factories.DocumentFactory(teams=[("team1", "administrator")], deleted_at=now)
factories.DocumentFactory(teams=[("team1", "administrator")])
deleted_document_team2 = factories.DocumentFactory(
teams=[("team2", "owner")], deleted_at=now
)
factories.DocumentFactory(teams=[("team2", "owner")])
factories.DocumentFactory(teams=[("team2", "administrator")], deleted_at=now)
factories.DocumentFactory(teams=[("team2", "administrator")])
expected_ids = {str(deleted_document_team1.id), str(deleted_document_team2.id)}
with django_assert_num_queries(5):
response = client.get("/api/v1.0/documents/trashbin/")
with django_assert_num_queries(3):
response = client.get("/api/v1.0/documents/trashbin/")
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
@mock.patch.object(PageNumberPagination, "get_page_size", return_value=2)
def test_api_documents_trashbin_pagination(
_mock_page_size,
):
"""Pagination should work as expected."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
document_ids = [
str(document.id)
for document in factories.DocumentFactory.create_batch(
3, deleted_at=timezone.now()
)
]
for document_id in document_ids:
models.DocumentAccess.objects.create(
document_id=document_id, user=user, role="owner"
)
# Get page 1
response = client.get("/api/v1.0/documents/trashbin/")
assert response.status_code == 200
content = response.json()
assert content["count"] == 3
assert content["next"] == "http://testserver/api/v1.0/documents/trashbin/?page=2"
assert content["previous"] is None
assert len(content["results"]) == 2
for item in content["results"]:
document_ids.remove(item["id"])
# Get page 2
response = client.get(
"/api/v1.0/documents/trashbin/?page=2",
)
assert response.status_code == 200
content = response.json()
assert content["count"] == 3
assert content["next"] is None
assert content["previous"] == "http://testserver/api/v1.0/documents/trashbin/"
assert len(content["results"]) == 1
document_ids.remove(content["results"][0]["id"])
assert document_ids == []
def test_api_documents_trashbin_distinct():
"""A document with several related users should only be listed once."""
user = factories.UserFactory()
client = APIClient()
client.force_login(user)
other_user = factories.UserFactory()
document = factories.DocumentFactory(
users=[(user, "owner"), other_user], deleted_at=timezone.now()
)
response = client.get(
"/api/v1.0/documents/trashbin/",
)
assert response.status_code == 200
content = response.json()
assert len(content["results"]) == 1
assert content["results"][0]["id"] == str(document.id)

94
src/backend/core/tests/test_api_utils_filter_root_paths.py Normal file
View File

@@ -0,0 +1,94 @@
"""
Unit tests for the filter_root_paths utility function.
"""
from core.api.utils import filter_root_paths
def test_api_utils_filter_root_paths_success():
"""
The `filter_root_paths` function should correctly identify root paths
from a given list of paths.
This test uses a list of paths with missing intermediate paths to ensure that
only the minimal set of root paths is returned.
"""
paths = [
"0001",
"00010001",
"000100010001",
"000100010002",
# missing 00010002
"000100020001",
"000100020002",
"0002",
"00020001",
"00020002",
# missing 0003
"00030001",
"000300010001",
"00030002",
# missing 0004
# missing 00040001
# missing 000400010001
# missing 000400010002
"000400010003",
"0004000100030001",
"000400010004",
]
filtered_paths = filter_root_paths(paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"0002",
"00030001",
"00030002",
"000400010003",
"000400010004",
]
def test_api_utils_filter_root_paths_sorting():
"""
The `filter_root_paths` function should fail is sorting is skipped and paths are not sorted.
This test verifies that when sorting is skipped, the function respects the input order, and
when sorting is enabled, the result is correctly ordered and minimal.
"""
paths = [
"0001",
"00010001",
"000100010001",
"000100020002",
"000100010002",
"000100020001",
"00020001",
"0002",
"00020002",
"000300010001",
"00030001",
"00030002",
"0004000100030001",
"000400010003",
"000400010004",
]
filtered_paths = filter_root_paths(paths, skip_sorting=True)
assert filtered_paths == [
"0001",
"00020001",
"0002",
"000300010001",
"00030001",
"00030002",
"0004000100030001",
"000400010003",
"000400010004",
]
filtered_paths = filter_root_paths(paths)
assert filtered_paths == [
"0001",
"0002",
"00030001",
"00030002",
"000400010003",
"000400010004",
]

207
src/backend/core/tests/test_models_documents.py
View File

@@ -2,12 +2,14 @@
Unit tests for the Document model
"""
import random
import smtplib
from logging import Logger
from unittest import mock
from django.contrib.auth.models import AnonymousUser
from django.core import mail
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.files.storage import default_storage
from django.utils import timezone
@@ -81,6 +83,44 @@ def test_models_documents_tree_alphabet():
assert models.Document.objects.count() == 124
@pytest.mark.parametrize("depth", range(5))
def test_models_documents_soft_delete(depth):
"""Trying to delete a document that is already deleted or is a descendant of
a deleted document should raise an error.
"""
documents = []
for i in range(depth + 1):
documents.append(
factories.DocumentFactory()
if i == 0
else factories.DocumentFactory(parent=documents[-1])
)
assert models.Document.objects.count() == depth + 1
# Delete any one of the documents...
deleted_document = random.choice(documents)
deleted_document.soft_delete()
with pytest.raises(RuntimeError):
documents[-1].soft_delete()
assert deleted_document.deleted_at is not None
assert deleted_document.ancestors_deleted_at == deleted_document.deleted_at
descendants = deleted_document.get_descendants()
for child in descendants:
assert child.deleted_at is None
assert child.ancestors_deleted_at is not None
assert child.ancestors_deleted_at == deleted_document.deleted_at
ancestors = deleted_document.get_ancestors()
for parent in ancestors:
assert parent.deleted_at is None
assert parent.ancestors_deleted_at is None
assert len(ancestors) + len(descendants) == depth
# get_abilities
@@ -95,15 +135,16 @@ def test_models_documents_tree_alphabet():
(False, "authenticated", "editor"),
],
)
def test_models_documents_get_abilities_forbidden(is_authenticated, reach, role):
def test_models_documents_get_abilities_forbidden(
is_authenticated, reach, role, django_assert_num_queries
):
"""
Check abilities returned for a document giving insufficient roles to link holders
i.e anonymous users or authenticated users who have no specific role on the document.
"""
document = factories.DocumentFactory(link_reach=reach, link_role=role)
user = factories.UserFactory() if is_authenticated else AnonymousUser()
abilities = document.get_abilities(user)
assert abilities == {
expected_abilities = {
"accesses_manage": False,
"accesses_view": False,
"ai_transform": False,
@@ -125,6 +166,12 @@ def test_models_documents_get_abilities_forbidden(is_authenticated, reach, role)
"versions_list": False,
"versions_retrieve": False,
}
nb_queries = 1 if is_authenticated else 0
with django_assert_num_queries(nb_queries):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert document.get_abilities(user) == expected_abilities
@pytest.mark.parametrize(
@@ -135,15 +182,16 @@ def test_models_documents_get_abilities_forbidden(is_authenticated, reach, role)
(True, "authenticated"),
],
)
def test_models_documents_get_abilities_reader(is_authenticated, reach):
def test_models_documents_get_abilities_reader(
is_authenticated, reach, django_assert_num_queries
):
"""
Check abilities returned for a document giving reader role to link holders
i.e anonymous users or authenticated users who have no specific role on the document.
"""
document = factories.DocumentFactory(link_reach=reach, link_role="reader")
user = factories.UserFactory() if is_authenticated else AnonymousUser()
abilities = document.get_abilities(user)
assert abilities == {
expected_abilities = {
"accesses_manage": False,
"accesses_view": False,
"ai_transform": False,
@@ -165,6 +213,12 @@ def test_models_documents_get_abilities_reader(is_authenticated, reach):
"versions_list": False,
"versions_retrieve": False,
}
nb_queries = 1 if is_authenticated else 0
with django_assert_num_queries(nb_queries):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert all(value is False for value in document.get_abilities(user).values())
@pytest.mark.parametrize(
@@ -175,15 +229,16 @@ def test_models_documents_get_abilities_reader(is_authenticated, reach):
(True, "authenticated"),
],
)
def test_models_documents_get_abilities_editor(is_authenticated, reach):
def test_models_documents_get_abilities_editor(
is_authenticated, reach, django_assert_num_queries
):
"""
Check abilities returned for a document giving editor role to link holders
i.e anonymous users or authenticated users who have no specific role on the document.
"""
document = factories.DocumentFactory(link_reach=reach, link_role="editor")
user = factories.UserFactory() if is_authenticated else AnonymousUser()
abilities = document.get_abilities(user)
assert abilities == {
expected_abilities = {
"accesses_manage": False,
"accesses_view": False,
"ai_transform": True,
@@ -205,14 +260,19 @@ def test_models_documents_get_abilities_editor(is_authenticated, reach):
"versions_list": False,
"versions_retrieve": False,
}
nb_queries = 1 if is_authenticated else 0
with django_assert_num_queries(nb_queries):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert all(value is False for value in document.get_abilities(user).values())
def test_models_documents_get_abilities_owner():
def test_models_documents_get_abilities_owner(django_assert_num_queries):
"""Check abilities returned for the owner of a document."""
user = factories.UserFactory()
access = factories.UserDocumentAccessFactory(role="owner", user=user)
abilities = access.document.get_abilities(access.user)
assert abilities == {
document = factories.DocumentFactory(users=[(user, "owner")])
expected_abilities = {
"accesses_manage": True,
"accesses_view": True,
"ai_transform": True,
@@ -234,13 +294,19 @@ def test_models_documents_get_abilities_owner():
"versions_list": True,
"versions_retrieve": True,
}
with django_assert_num_queries(1):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
expected_abilities["move"] = False
assert document.get_abilities(user) == expected_abilities
def test_models_documents_get_abilities_administrator():
def test_models_documents_get_abilities_administrator(django_assert_num_queries):
"""Check abilities returned for the administrator of a document."""
access = factories.UserDocumentAccessFactory(role="administrator")
abilities = access.document.get_abilities(access.user)
assert abilities == {
user = factories.UserFactory()
document = factories.DocumentFactory(users=[(user, "administrator")])
expected_abilities = {
"accesses_manage": True,
"accesses_view": True,
"ai_transform": True,
@@ -262,16 +328,18 @@ def test_models_documents_get_abilities_administrator():
"versions_list": True,
"versions_retrieve": True,
}
with django_assert_num_queries(1):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert all(value is False for value in document.get_abilities(user).values())
def test_models_documents_get_abilities_editor_user(django_assert_num_queries):
"""Check abilities returned for the editor of a document."""
access = factories.UserDocumentAccessFactory(role="editor")
with django_assert_num_queries(1):
abilities = access.document.get_abilities(access.user)
assert abilities == {
user = factories.UserFactory()
document = factories.DocumentFactory(users=[(user, "editor")])
expected_abilities = {
"accesses_manage": False,
"accesses_view": True,
"ai_transform": True,
@@ -293,24 +361,27 @@ def test_models_documents_get_abilities_editor_user(django_assert_num_queries):
"versions_list": True,
"versions_retrieve": True,
}
with django_assert_num_queries(1):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert all(value is False for value in document.get_abilities(user).values())
def test_models_documents_get_abilities_reader_user(django_assert_num_queries):
"""Check abilities returned for the reader of a document."""
access = factories.UserDocumentAccessFactory(
role="reader", document__link_role="reader"
user = factories.UserFactory()
document = factories.DocumentFactory(users=[(user, "reader")])
access_from_link = (
document.link_reach != "restricted" and document.link_role == "editor"
)
with django_assert_num_queries(1):
abilities = access.document.get_abilities(access.user)
assert abilities == {
expected_abilities = {
"accesses_manage": False,
"accesses_view": True,
"ai_transform": False,
"ai_translate": False,
"attachment_upload": False,
"children_create": False,
"ai_transform": access_from_link,
"ai_translate": access_from_link,
"attachment_upload": access_from_link,
"children_create": access_from_link,
"children_list": True,
"collaboration_auth": True,
"destroy": False,
@@ -319,13 +390,18 @@ def test_models_documents_get_abilities_reader_user(django_assert_num_queries):
"link_configuration": False,
"media_auth": True,
"move": False,
"partial_update": False,
"partial_update": access_from_link,
"retrieve": True,
"update": False,
"update": access_from_link,
"versions_destroy": False,
"versions_list": True,
"versions_retrieve": True,
}
with django_assert_num_queries(1):
assert document.get_abilities(user) == expected_abilities
document.soft_delete()
document.refresh_from_db()
assert all(value is False for value in document.get_abilities(user).values())
def test_models_documents_get_abilities_preset_role(django_assert_num_queries):
@@ -555,3 +631,62 @@ def test_models_documents__email_invitation__failed(mock_logger, _mock_send_mail
assert emails == ["guest3@example.com"]
assert isinstance(exception, smtplib.SMTPException)
# Document number of accesses
def test_models_documents_nb_accesses_cache_is_set_and_retrieved(
django_assert_num_queries,
):
"""Test that nb_accesses is cached after the first computation."""
document = factories.DocumentFactory()
key = f"document_{document.id!s}_nb_accesses"
nb_accesses = random.randint(1, 4)
factories.UserDocumentAccessFactory.create_batch(nb_accesses, document=document)
factories.UserDocumentAccessFactory() # An unrelated access should not be counted
# Initially, the nb_accesses should not be cached
assert cache.get(key) is None
# Compute the nb_accesses for the first time (this should set the cache)
with django_assert_num_queries(1):
assert document.nb_accesses == nb_accesses
# Ensure that the nb_accesses is now cached
with django_assert_num_queries(0):
assert document.nb_accesses == nb_accesses
assert cache.get(key) == nb_accesses
# The cache value should be invalidated when a document access is created
models.DocumentAccess.objects.create(
document=document, user=factories.UserFactory(), role="reader"
)
assert cache.get(key) is None # Cache should be invalidated
with django_assert_num_queries(1):
new_nb_accesses = document.nb_accesses
assert new_nb_accesses == nb_accesses + 1
assert cache.get(key) == new_nb_accesses # Cache should now contain the new value
def test_models_documents_nb_accesses_cache_is_invalidated_on_access_removal(
django_assert_num_queries,
):
"""Test that the cache is invalidated when a document access is deleted."""
document = factories.DocumentFactory()
key = f"document_{document.id!s}_nb_accesses"
access = factories.UserDocumentAccessFactory(document=document)
# Initially, the nb_accesses should be cached
assert document.nb_accesses == 1
assert cache.get(key) == 1
# Remove the access and check if cache is invalidated
access.delete()
assert cache.get(key) is None # Cache should be invalidated
# Recompute the nb_accesses (this should trigger a cache set)
with django_assert_num_queries(1):
new_nb_accesses = document.nb_accesses
assert new_nb_accesses == 0
assert cache.get(key) == 0 # Cache should now contain the new value